openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-13 11:47:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

rollout: persist bounded command execution history

Browse Source
This commit is contained in:
Adam Perry
2026-06-13 06:42:25 +00:00
parent 7acc90f497
commit bc6931247b
7 changed files with 262 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
codex-rs/Cargo.lock generated
View File

@@ -3705,6 +3705,7 @@ dependencies = [
"codex-protocol",
"codex-state",
"codex-utils-path",
"codex-utils-string",
"pretty_assertions",
"regex",
"serde",

50
codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs
View File

@@ -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() {

58
codex-rs/app-server/tests/suite/v2/thread_shell_command.rs
View File

@@ -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::<ThreadStartResponse>(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::<ThreadReadResponse>(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::<ThreadTurnsListResponse>(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::<ThreadForkResponse>(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::<ThreadReadResponse>(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::<Vec<_>>();
assert_eq!(
command_executions,
vec![expected],
"{context} should include exactly the persisted command execution"
);
}

73
codex-rs/core/tests/remote_env_windows/wine_app_server_windows_exec_server_test.rs
View File

@@ -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::<Vec<_>>()
})
.collect::<Vec<_>>();
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(())
}

1
codex-rs/rollout/Cargo.toml
View File

@@ -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 }

38
codex-rs/rollout/src/policy.rs
View File

@@ -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<RolloutItem> {
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;

61
codex-rs/rollout/src/policy_tests.rs Normal file
View File

@@ -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"),
);
}