diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 738a7c1c90..76d8e98b9a 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3705,6 +3705,7 @@ dependencies = [ "codex-protocol", "codex-state", "codex-utils-path", + "codex-utils-string", "pretty_assertions", "regex", "serde", diff --git a/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs b/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs index d06a91c898..c50a938aeb 100644 --- a/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs @@ -1,9 +1,12 @@ use super::*; use codex_protocol::parse_command::ParsedCommand; use codex_protocol::protocol::ExecCommandBeginEvent; +use codex_protocol::protocol::ExecCommandEndEvent; use codex_protocol::protocol::ExecCommandSource; +use codex_protocol::protocol::ExecCommandStatus; use pretty_assertions::assert_eq; use std::path::PathBuf; +use std::time::Duration; fn begin_event( cwd: PathUri, @@ -53,6 +56,53 @@ fn windows_command_event_renders_windows_native_cwd() { ); } +#[test] +fn persisted_windows_command_completion_renders_windows_native_cwd() { + let event = ExecCommandEndEvent { + call_id: "exec-1".to_string(), + process_id: None, + turn_id: "turn-1".to_string(), + completed_at_ms: 0, + command: vec![ + "pwsh.exe".to_string(), + "-Command".to_string(), + "pwd".to_string(), + ], + cwd: PathUri::parse("file:///C:/Research/space%20%23%25").expect("Windows cwd URI"), + path_convention: PathConvention::Windows, + parsed_cmd: vec![ParsedCommand::Unknown { + cmd: "pwsh.exe -Command pwd".to_string(), + }], + source: ExecCommandSource::Agent, + interaction_input: None, + stdout: String::new(), + stderr: String::new(), + aggregated_output: "C:\\Research\\space #%\r\n".to_string(), + exit_code: 0, + duration: Duration::from_millis(12), + formatted_output: String::new(), + status: ExecCommandStatus::Completed, + }; + + assert_eq!( + build_command_execution_end_item(&event), + ThreadItem::CommandExecution { + id: "exec-1".to_string(), + command: "pwsh.exe -Command pwd".to_string(), + cwd: ApiPathString::new(r"C:\Research\space #%"), + process_id: None, + source: CommandExecutionSource::Agent, + status: CommandExecutionStatus::Completed, + command_actions: vec![CommandAction::Unknown { + command: "pwsh.exe -Command pwd".to_string(), + }], + aggregated_output: Some("C:\\Research\\space #%\r\n".to_string()), + exit_code: Some(0), + duration_ms: Some(12), + } + ); +} + #[test] fn foreign_command_event_does_not_project_read_path_onto_host() { let (cwd, path_convention, native_cwd) = match PathConvention::native() { diff --git a/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs b/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs index 312267d0a6..32ab6198d9 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs @@ -28,6 +28,7 @@ use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadTurnsListParams; use codex_app_server_protocol::ThreadTurnsListResponse; use codex_app_server_protocol::TurnCompletedNotification; +use codex_app_server_protocol::TurnItemsView; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::UserInput as V2UserInput; @@ -44,7 +45,7 @@ use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); #[tokio::test] -async fn thread_shell_command_history_responses_exclude_persisted_command_executions() -> Result<()> +async fn thread_shell_command_history_responses_restore_persisted_command_execution() -> Result<()> { let tmp = TempDir::new()?; let codex_home = tmp.path().join("codex_home"); @@ -72,11 +73,12 @@ async fn thread_shell_command_history_responses_exclude_persisted_command_execut ) .await??; let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; + let thread_id = thread.id; let (shell_command, expected_output) = current_shell_output_command("hello from bang")?; let shell_id = mcp .send_thread_shell_command_request(ThreadShellCommandParams { - thread_id: thread.id.clone(), + thread_id: thread_id.clone(), command: shell_command, }) .await?; @@ -121,6 +123,7 @@ async fn thread_shell_command_history_responses_exclude_persisted_command_execut assert_eq!(status, &CommandExecutionStatus::Completed); assert_eq!(aggregated_output.as_deref(), Some(expected_output.as_str())); assert_eq!(*exit_code, Some(0)); + let expected_completed_item = completed.item; timeout( DEFAULT_READ_TIMEOUT, @@ -128,9 +131,13 @@ async fn thread_shell_command_history_responses_exclude_persisted_command_execut ) .await??; + drop(mcp); + let mut mcp = TestAppServer::new(codex_home.as_path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + let read_id = mcp .send_thread_read_request(ThreadReadParams { - thread_id: thread.id.clone(), + thread_id: thread_id.clone(), include_turns: true, }) .await?; @@ -141,15 +148,19 @@ async fn thread_shell_command_history_responses_exclude_persisted_command_execut .await??; let ThreadReadResponse { thread, .. } = to_response::(read_resp)?; assert_eq!(thread.turns.len(), 1); - assert_no_command_executions(&thread.turns[0].items, "thread/read"); + assert_only_command_execution( + &thread.turns[0].items, + &expected_completed_item, + "thread/read", + ); let turns_list_id = mcp .send_thread_turns_list_request(ThreadTurnsListParams { - thread_id: thread.id.clone(), + thread_id: thread_id.clone(), cursor: None, limit: None, sort_direction: Some(SortDirection::Asc), - items_view: None, + items_view: Some(TurnItemsView::Full), }) .await?; let turns_list_resp: JSONRPCResponse = timeout( @@ -160,11 +171,15 @@ async fn thread_shell_command_history_responses_exclude_persisted_command_execut let ThreadTurnsListResponse { data, .. } = to_response::(turns_list_resp)?; assert_eq!(data.len(), 1); - assert_no_command_executions(&data[0].items, "thread/turns/list"); + assert_only_command_execution( + &data[0].items, + &expected_completed_item, + "thread/turns/list", + ); let fork_id = mcp .send_thread_fork_request(ThreadForkParams { - thread_id: thread.id, + thread_id, ..Default::default() }) .await?; @@ -175,7 +190,11 @@ async fn thread_shell_command_history_responses_exclude_persisted_command_execut .await??; let ThreadForkResponse { thread, .. } = to_response::(fork_resp)?; assert_eq!(thread.turns.len(), 1); - assert_no_command_executions(&thread.turns[0].items, "thread/fork"); + assert_only_command_execution( + &thread.turns[0].items, + &expected_completed_item, + "thread/fork", + ); Ok(()) } @@ -340,6 +359,7 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> { }; assert_eq!(source, &CommandExecutionSource::UserShell); assert_eq!(aggregated_output.as_deref(), Some(expected_output.as_str())); + let expected_completed_item = completed.item; mcp.send_response( request_id, @@ -371,17 +391,23 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> { .await??; let ThreadReadResponse { thread, .. } = to_response::(read_resp)?; assert_eq!(thread.turns.len(), 1); - assert_no_command_executions(&thread.turns[0].items, "thread/read"); + assert!( + thread.turns[0].items.contains(&expected_completed_item), + "thread/read should include the persisted user shell command execution" + ); Ok(()) } -fn assert_no_command_executions(items: &[ThreadItem], context: &str) { - assert!( - items - .iter() - .all(|item| !matches!(item, ThreadItem::CommandExecution { .. })), - "{context} should always exclude command executions from returned turns" +fn assert_only_command_execution(items: &[ThreadItem], expected: &ThreadItem, context: &str) { + let command_executions = items + .iter() + .filter(|item| matches!(item, ThreadItem::CommandExecution { .. })) + .collect::>(); + assert_eq!( + command_executions, + vec![expected], + "{context} should include exactly the persisted command execution" ); } diff --git a/codex-rs/core/tests/remote_env_windows/wine_app_server_windows_exec_server_test.rs b/codex-rs/core/tests/remote_env_windows/wine_app_server_windows_exec_server_test.rs index bb0985206a..0d48ca2a8a 100644 --- a/codex-rs/core/tests/remote_env_windows/wine_app_server_windows_exec_server_test.rs +++ b/codex-rs/core/tests/remote_env_windows/wine_app_server_windows_exec_server_test.rs @@ -19,6 +19,8 @@ use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::SandboxPolicy; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadReadParams; +use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::TurnCompletedNotification; @@ -146,11 +148,12 @@ async fn exercise_app_server(websocket_url: String) -> Result<()> { ) .await??; let ThreadStartResponse { thread, .. } = to_response(thread_response)?; + let thread_id = thread.id; let first_items = submit_turn_and_collect_command( &mut app_server, TurnStartParams { - thread_id: thread.id.clone(), + thread_id: thread_id.clone(), input: vec![UserInput::Text { text: "Run the default Windows shell in a relative workdir.".to_string(), text_elements: Vec::new(), @@ -175,7 +178,7 @@ async fn exercise_app_server(websocket_url: String) -> Result<()> { let sticky_items = submit_turn_and_collect_command( &mut app_server, TurnStartParams { - thread_id: thread.id, + thread_id: thread_id.clone(), input: vec![UserInput::Text { text: "Use explicit powershell.exe in an absolute workdir.".to_string(), text_elements: Vec::new(), @@ -193,6 +196,71 @@ async fn exercise_app_server(websocket_url: String) -> Result<()> { "powershell.exe", ); + drop(app_server); + let mut app_server = TestAppServer::new_with_program_and_env( + codex_home.path(), + &app_server_program, + &[( + CODEX_EXEC_SERVER_URL_ENV_VAR, + Some(websocket_url.as_str()), + )], + ) + .await?; + timeout(APP_SERVER_TIMEOUT, app_server.initialize()).await??; + let read_request_id = app_server + .send_thread_read_request(ThreadReadParams { + thread_id, + include_turns: true, + }) + .await?; + let read_response: JSONRPCResponse = timeout( + APP_SERVER_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(read_request_id)), + ) + .await??; + let ThreadReadResponse { + thread: persisted_thread, + } = to_response(read_response)?; + let persisted_command_ids = persisted_thread + .turns + .iter() + .map(|turn| { + turn.items + .iter() + .filter_map(|item| match item { + ThreadItem::CommandExecution { id, .. } => Some(id.as_str()), + _ => None, + }) + .collect::>() + }) + .collect::>(); + assert_eq!( + persisted_command_ids, + vec![vec![FIRST_EXEC_CALL_ID], vec![STICKY_EXEC_CALL_ID]] + ); + let persisted_first = persisted_thread.turns[0] + .items + .iter() + .find(|item| matches!(item, ThreadItem::CommandExecution { id, .. } if id == FIRST_EXEC_CALL_ID)) + .context("persisted first command should be present")?; + assert_completed_command_item( + persisted_first, + FIRST_EXEC_CALL_ID, + FIRST_WINDOWS_CWD, + FIRST_OUTPUT_MARKER, + ); + let persisted_sticky = persisted_thread.turns[1] + .items + .iter() + .find(|item| matches!(item, ThreadItem::CommandExecution { id, .. } if id == STICKY_EXEC_CALL_ID)) + .context("persisted sticky command should be present")?; + assert_completed_command_item( + persisted_sticky, + STICKY_EXEC_CALL_ID, + ABSOLUTE_WINDOWS_CWD, + STICKY_OUTPUT_MARKER, + ); + let requests = response_mock.requests(); assert_eq!(requests.len(), 4); let first_environment_context = requests[0] @@ -234,6 +302,7 @@ async fn exercise_app_server(websocket_url: String) -> Result<()> { assert_no_linux_execution_artifacts(&request.body_json().to_string(), &linux_cwd); } assert_no_linux_execution_artifacts(&format!("{first_items:?}{sticky_items:?}"), &linux_cwd); + assert_no_linux_execution_artifacts(&format!("{persisted_thread:?}"), &linux_cwd); Ok(()) } diff --git a/codex-rs/rollout/Cargo.toml b/codex-rs/rollout/Cargo.toml index edc9059688..3598773407 100644 --- a/codex-rs/rollout/Cargo.toml +++ b/codex-rs/rollout/Cargo.toml @@ -22,6 +22,7 @@ codex-otel = { workspace = true } codex-protocol = { workspace = true } codex-state = { workspace = true } codex-utils-path = { workspace = true } +codex-utils-string = { workspace = true } regex = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 40ca77c97e..1de6d064e3 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -1,6 +1,21 @@ use crate::protocol::EventMsg; use crate::protocol::RolloutItem; use codex_protocol::models::ResponseItem; +use codex_utils_string::truncate_middle_chars; + +const PERSISTED_EXEC_AGGREGATED_OUTPUT_MAX_BYTES: usize = 10_000; + +fn truncate_persisted_exec_output(output: &str) -> String { + let mut retained_bytes = PERSISTED_EXEC_AGGREGATED_OUTPUT_MAX_BYTES; + loop { + let truncated = truncate_middle_chars(output, retained_bytes); + if truncated.len() <= PERSISTED_EXEC_AGGREGATED_OUTPUT_MAX_BYTES { + return truncated; + } + retained_bytes = retained_bytes + .saturating_sub(truncated.len() - PERSISTED_EXEC_AGGREGATED_OUTPUT_MAX_BYTES); + } +} /// Whether a rollout `item` should be persisted in rollout files. pub fn is_persisted_rollout_item(item: &RolloutItem) -> bool { @@ -20,12 +35,27 @@ pub fn persisted_rollout_items(items: &[RolloutItem]) -> Vec { let mut persisted = Vec::new(); for item in items { if is_persisted_rollout_item(item) { - persisted.push(item.clone()); + persisted.push(sanitize_rollout_item_for_persistence(item.clone())); } } persisted } +fn sanitize_rollout_item_for_persistence(item: RolloutItem) -> RolloutItem { + match item { + RolloutItem::EventMsg(EventMsg::ExecCommandEnd(mut event)) => { + // Rebuilt app-server history needs the aggregate, while the per-stream and + // model-formatted copies would only duplicate potentially large output. + event.aggregated_output = truncate_persisted_exec_output(&event.aggregated_output); + event.stdout.clear(); + event.stderr.clear(); + event.formatted_output.clear(); + RolloutItem::EventMsg(EventMsg::ExecCommandEnd(event)) + } + _ => item, + } +} + /// Whether a `ResponseItem` should be persisted in rollout files. #[inline] pub fn should_persist_response_item(item: &ResponseItem) -> bool { @@ -93,6 +123,7 @@ pub fn should_persist_event_msg(ev: &EventMsg) -> bool { | EventMsg::TurnComplete(_) | EventMsg::WebSearchEnd(_) | EventMsg::ImageGenerationEnd(_) + | EventMsg::ExecCommandEnd(_) | EventMsg::SubAgentActivity(_) => true, EventMsg::ItemCompleted(event) => { // Plan items are derived from streaming tags and are not part of the @@ -102,7 +133,6 @@ pub fn should_persist_event_msg(ev: &EventMsg) -> bool { } EventMsg::Error(_) | EventMsg::GuardianAssessment(_) - | EventMsg::ExecCommandEnd(_) | EventMsg::ViewImageToolCall(_) | EventMsg::CollabAgentSpawnEnd(_) | EventMsg::CollabAgentInteractionEnd(_) @@ -159,3 +189,7 @@ pub fn should_persist_event_msg(ev: &EventMsg) -> bool { | EventMsg::CollabResumeBegin(_) => false, } } + +#[cfg(test)] +#[path = "policy_tests.rs"] +mod tests; diff --git a/codex-rs/rollout/src/policy_tests.rs b/codex-rs/rollout/src/policy_tests.rs new file mode 100644 index 0000000000..483275b661 --- /dev/null +++ b/codex-rs/rollout/src/policy_tests.rs @@ -0,0 +1,61 @@ +use super::*; +use codex_protocol::parse_command::ParsedCommand; +use codex_protocol::protocol::ExecCommandEndEvent; +use codex_protocol::protocol::ExecCommandSource; +use codex_protocol::protocol::ExecCommandStatus; +use pretty_assertions::assert_eq; +use serde_json::Value; +use std::time::Duration; + +#[test] +fn persists_sanitized_exec_command_completion() { + let large_output = "😀0123456789".repeat(2_000); + let input_event = ExecCommandEndEvent { + call_id: "exec-1".to_string(), + process_id: Some("process-1".to_string()), + turn_id: "turn-1".to_string(), + completed_at_ms: 42, + command: vec![ + "pwsh.exe".to_string(), + "-Command".to_string(), + "pwd".to_string(), + ], + cwd: serde_json::from_value(Value::String("file:///C:/Research/workspace".to_string())) + .expect("Windows cwd URI"), + path_convention: serde_json::from_value(Value::String("windows".to_string())) + .expect("Windows path convention"), + parsed_cmd: vec![ParsedCommand::Unknown { + cmd: "pwsh.exe -Command pwd".to_string(), + }], + source: ExecCommandSource::Agent, + interaction_input: None, + stdout: large_output.clone(), + stderr: "stderr copy".to_string(), + aggregated_output: large_output, + exit_code: 0, + duration: Duration::from_millis(250), + formatted_output: "formatted copy".to_string(), + status: ExecCommandStatus::Completed, + }; + let input = RolloutItem::EventMsg(EventMsg::ExecCommandEnd(input_event.clone())); + + let persisted = persisted_rollout_items(std::slice::from_ref(&input)); + + let mut expected_event = input_event; + expected_event.aggregated_output = + truncate_persisted_exec_output(&expected_event.aggregated_output); + expected_event.stdout.clear(); + expected_event.stderr.clear(); + expected_event.formatted_output.clear(); + let expected = vec![RolloutItem::EventMsg(EventMsg::ExecCommandEnd( + expected_event, + ))]; + let RolloutItem::EventMsg(EventMsg::ExecCommandEnd(persisted_event)) = &persisted[0] else { + panic!("expected persisted command completion") + }; + assert!(persisted_event.aggregated_output.len() <= PERSISTED_EXEC_AGGREGATED_OUTPUT_MAX_BYTES); + assert_eq!( + serde_json::to_value(persisted).expect("serialize persisted items"), + serde_json::to_value(expected).expect("serialize expected items"), + ); +}