diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index 37a4759b82..5c4a429b3f 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -996,6 +996,15 @@ impl UnifiedExecProcessManager { } = self .prepare_process_handles(process_id, &locked_process) .await?; + context + .session + .services + .rollout_thread_trace + .record_stdin_target( + &context.step_context.turn.sub_id, + &context.call_id, + &call_id, + ); let remote_exited = process.has_exited() && output.drain_policy == OutputDrainPolicy::WaitForOutputClosure; // A write racing with remote exit still collects the final output. diff --git a/codex-rs/rollout-trace/src/model/runtime.rs b/codex-rs/rollout-trace/src/model/runtime.rs index 297166e5b7..36b399ac47 100644 --- a/codex-rs/rollout-trace/src/model/runtime.rs +++ b/codex-rs/rollout-trace/src/model/runtime.rs @@ -215,7 +215,7 @@ pub enum ToolCallSummary { }, } -/// Reusable terminal process/session returned by the runtime. +/// One terminal process lifetime, including its command and subsequent polls. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct TerminalSession { pub terminal_id: TerminalId, @@ -230,7 +230,7 @@ pub struct TerminalSession { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct TerminalOperation { pub operation_id: TerminalOperationId, - /// Runtime terminal/process ID. `None` is legal only while the operation that creates it is starting. + /// Reduced session identity, distinct from the recyclable runtime process ID. pub terminal_id: Option, pub tool_call_id: ToolCallId, pub kind: TerminalOperationKind, diff --git a/codex-rs/rollout-trace/src/raw_event.rs b/codex-rs/rollout-trace/src/raw_event.rs index bd9010d7bb..b8ec8f315e 100644 --- a/codex-rs/rollout-trace/src/raw_event.rs +++ b/codex-rs/rollout-trace/src/raw_event.rs @@ -144,6 +144,11 @@ pub enum RawTraceEventPayload { tool_call_id: ToolCallId, mcp_call_id: McpCallId, }, + /// Exec selected by the runtime after resolving a recyclable process handle. + StdinTargetResolved { + tool_call_id: ToolCallId, + exec_tool_call_id: ToolCallId, + }, ToolCallRuntimeStarted { tool_call_id: ToolCallId, /// Runtime/protocol observation for how Codex began executing the tool. @@ -243,6 +248,7 @@ impl RawTraceEventPayload { | RawTraceEventPayload::CompactionRequestFailed { .. } | RawTraceEventPayload::CodeCellStarted { .. } | RawTraceEventPayload::McpToolCallCorrelationAssigned { .. } + | RawTraceEventPayload::StdinTargetResolved { .. } | RawTraceEventPayload::AgentResultObserved { carried_payload: None, .. diff --git a/codex-rs/rollout-trace/src/reducer/mod.rs b/codex-rs/rollout-trace/src/reducer/mod.rs index 91eeba55d4..123c0cbff6 100644 --- a/codex-rs/rollout-trace/src/reducer/mod.rs +++ b/codex-rs/rollout-trace/src/reducer/mod.rs @@ -57,6 +57,8 @@ pub fn replay_bundle(bundle_dir: impl AsRef) -> Result { bundle_dir: bundle_dir.to_path_buf(), next_conversation_item_ordinal: 1, next_terminal_operation_ordinal: 1, + terminal_ids_by_process: BTreeMap::new(), + process_ids_by_terminal: BTreeMap::new(), thread_conversation_snapshots: BTreeMap::new(), pending_compaction_replacement_item_ids: BTreeMap::new(), code_cell_ids_by_runtime: BTreeMap::new(), @@ -90,6 +92,10 @@ struct TraceReducer { bundle_dir: PathBuf, next_conversation_item_ordinal: u64, next_terminal_operation_ordinal: u64, + /// Latest session for each thread-local runtime handle; each exec replaces it. + terminal_ids_by_process: BTreeMap<(String, String), String>, + /// Retains the begin/end consistency check when a handle is later reused. + process_ids_by_terminal: BTreeMap, /// Last model-visible conversation snapshot per thread. /// /// Requests and responses both advance this sequence because both are @@ -267,6 +273,12 @@ impl TraceReducer { } => { self.assign_mcp_tool_call_correlation(tool_call_id, mcp_call_id)?; } + RawTraceEventPayload::StdinTargetResolved { + tool_call_id, + exec_tool_call_id, + } => { + self.resolve_stdin_target(&tool_call_id, &exec_tool_call_id)?; + } RawTraceEventPayload::ToolCallRuntimeStarted { tool_call_id, runtime_payload, diff --git a/codex-rs/rollout-trace/src/reducer/tool.rs b/codex-rs/rollout-trace/src/reducer/tool.rs index 20e1e2af47..45913b0a2b 100644 --- a/codex-rs/rollout-trace/src/reducer/tool.rs +++ b/codex-rs/rollout-trace/src/reducer/tool.rs @@ -101,9 +101,8 @@ impl TraceReducer { self.thread_mut(&thread_id)?; - // Some terminal-like tools, notably write_stdin, do not emit a richer - // runtime begin event. For those tools the canonical invocation is the - // only place to recover the terminal/session join key. + // write_stdin has no protocol runtime-begin event, so its request comes + // from the canonical invocation. Runtime resolution can refine its session link. let terminal_operation_id = self.start_terminal_operation_from_invocation( seq, wall_time_unix_ms, diff --git a/codex-rs/rollout-trace/src/reducer/tool/terminal.rs b/codex-rs/rollout-trace/src/reducer/tool/terminal.rs index f9c6bc3788..13f56d7461 100644 --- a/codex-rs/rollout-trace/src/reducer/tool/terminal.rs +++ b/codex-rs/rollout-trace/src/reducer/tool/terminal.rs @@ -5,6 +5,9 @@ //! runtime details. Direct tools without protocol observations, such as //! `write_stdin`, can still form a terminal row from the canonical dispatch //! invocation/result payloads when those payloads carry the session join key. +//! Runtime process handles are recyclable and thread-local. Each exec creates a +//! distinct reduced session; stdin uses the runtime's resolved exec when recorded, +//! falling back to the latest session for its handle in older traces. use anyhow::Context; use anyhow::Result; @@ -114,7 +117,7 @@ impl TraceReducer { ) -> Result> { let operation_id = self.next_terminal_operation_id(); let ParsedTerminalRequest { - terminal_id, + process_id, request, } = start.request; @@ -122,7 +125,7 @@ impl TraceReducer { operation_id.clone(), TerminalOperation { operation_id: operation_id.clone(), - terminal_id: terminal_id.clone(), + terminal_id: None, tool_call_id: start.tool_call_id.to_string(), kind: start.operation_kind, execution: ExecutionWindow { @@ -139,14 +142,8 @@ impl TraceReducer { }, ); - if let Some(terminal_id) = terminal_id { - self.ensure_terminal_session( - start.thread_id, - &terminal_id, - &operation_id, - start.wall_time_unix_ms, - start.seq, - )?; + if let Some(process_id) = process_id { + self.attach_terminal_session(start.thread_id, &process_id, &operation_id)?; } Ok(Some(operation_id)) @@ -185,7 +182,7 @@ impl TraceReducer { }) .transpose()?; - let (terminal_id, started_at_unix_ms, started_seq) = { + let process_id = { let Some(operation) = self.rollout.terminal_operations.get_mut(operation_id) else { bail!("terminal end referenced unknown operation {operation_id}"); }; @@ -195,84 +192,128 @@ impl TraceReducer { if let Some((raw_payload_id, response)) = response { push_unique(&mut operation.raw_payload_ids, &raw_payload_id); - // If begin and end both report a process id they must name the - // same terminal. If begin omitted it, the end event completes - // the session join key for this operation. - match (&operation.terminal_id, response.terminal_id.as_deref()) { - (Some(existing), Some(process_id)) if existing != process_id => { - bail!( - "terminal operation {operation_id} changed process id from \ - {existing} to {process_id}" - ); - } - (None, Some(process_id)) => { - operation.terminal_id = Some(process_id.to_string()); - } - (Some(_), Some(_)) | (Some(_), None) | (None, None) => {} - } operation.result = Some(response.result); + response.process_id + } else { + None } - - ( - operation.terminal_id.clone(), - operation.execution.started_at_unix_ms, - operation.execution.started_seq, - ) }; - if let Some(terminal_id) = terminal_id { - self.ensure_terminal_session( - thread_id, - &terminal_id, - operation_id, - started_at_unix_ms, - started_seq, - )?; + if let Some(process_id) = process_id { + self.attach_terminal_session(thread_id, &process_id, operation_id)?; } Ok(()) } - fn ensure_terminal_session( + fn attach_terminal_session( &mut self, thread_id: &str, - terminal_id: &str, + process_id: &str, operation_id: &str, - started_at_unix_ms: i64, - started_seq: RawEventSeq, ) -> Result<()> { - if !self.rollout.terminal_sessions.contains_key(terminal_id) { - self.rollout.terminal_sessions.insert( - terminal_id.to_string(), - TerminalSession { - terminal_id: terminal_id.to_string(), - thread_id: thread_id.to_string(), - created_by_operation_id: operation_id.to_string(), - operation_ids: Vec::new(), - execution: ExecutionWindow { - started_at_unix_ms, - started_seq, - // Current raw events do not report a terminal/session - // shutdown boundary, so the session remains open even - // after individual operations complete. - ended_at_unix_ms: None, - ended_seq: None, - status: ExecutionStatus::Running, - }, - }, - ); + let Some(operation) = self.rollout.terminal_operations.get_mut(operation_id) else { + bail!("terminal session referenced unknown operation {operation_id}"); + }; + if let Some(terminal_id) = &operation.terminal_id { + let existing = &self.process_ids_by_terminal[terminal_id]; + if existing != process_id { + bail!( + "terminal operation {operation_id} changed process id from {existing} to {process_id}" + ); + } + return Ok(()); } - let Some(session) = self.rollout.terminal_sessions.get_mut(terminal_id) else { - bail!("terminal session {terminal_id} disappeared during reduction"); + let key = (thread_id.to_string(), process_id.to_string()); + let existing = match operation.kind { + TerminalOperationKind::ExecCommand => None, + TerminalOperationKind::WriteStdin => self.terminal_ids_by_process.get(&key).cloned(), }; - if session.thread_id != thread_id { - bail!( - "terminal session {terminal_id} belongs to thread {}, not {thread_id}", - session.thread_id - ); - } + let terminal_id = existing.unwrap_or_else(|| format!("terminal_session:{operation_id}")); + let session = self + .rollout + .terminal_sessions + .entry(terminal_id.clone()) + .or_insert_with(|| TerminalSession { + terminal_id: terminal_id.clone(), + thread_id: thread_id.to_string(), + created_by_operation_id: operation_id.to_string(), + operation_ids: Vec::new(), + execution: ExecutionWindow { + started_at_unix_ms: operation.execution.started_at_unix_ms, + started_seq: operation.execution.started_seq, + // Raw events do not report session release: buffered output + // can still be collected after the process exits. + ended_at_unix_ms: None, + ended_seq: None, + status: ExecutionStatus::Running, + }, + }); push_unique(&mut session.operation_ids, operation_id); + operation.terminal_id = Some(terminal_id.clone()); + self.process_ids_by_terminal + .insert(terminal_id.clone(), process_id.to_string()); + self.terminal_ids_by_process.insert(key, terminal_id); + Ok(()) + } + + /// Replaces the provisional handle-based join with the runtime's actual target. + pub(in crate::reducer) fn resolve_stdin_target( + &mut self, + tool_call_id: &str, + exec_tool_call_id: &str, + ) -> Result<()> { + let stdin = self + .rollout + .tool_calls + .get(tool_call_id) + .with_context(|| format!("stdin target referenced unknown call {tool_call_id}"))?; + let exec = self + .rollout + .tool_calls + .get(exec_tool_call_id) + .with_context(|| format!("stdin target referenced unknown exec {exec_tool_call_id}"))?; + let (Some(operation_id), Some(exec_operation_id)) = + (&stdin.terminal_operation_id, &exec.terminal_operation_id) + else { + // A failed payload write can leave either call without a terminal projection. + return Ok(()); + }; + let Some(terminal_id) = self.rollout.terminal_operations[exec_operation_id] + .terminal_id + .clone() + else { + return Ok(()); + }; + let operation = self + .rollout + .terminal_operations + .get_mut(operation_id) + .with_context(|| format!("stdin target referenced unknown operation {operation_id}"))?; + if operation.terminal_id.as_ref() == Some(&terminal_id) { + return Ok(()); + } + // Move this operation without rewinding the handle lookup: a newer exec + // may already own that handle by the time resolution is recorded. + if let Some(previous) = operation.terminal_id.replace(terminal_id.clone()) { + let session = self + .rollout + .terminal_sessions + .get_mut(&previous) + .with_context(|| format!("stdin target referenced unknown session {previous}"))?; + session.operation_ids.retain(|id| id != operation_id); + if session.operation_ids.is_empty() { + self.rollout.terminal_sessions.remove(&previous); + self.process_ids_by_terminal.remove(&previous); + } + } + self.rollout + .terminal_sessions + .get_mut(&terminal_id) + .with_context(|| format!("stdin target referenced unknown session {terminal_id}"))? + .operation_ids + .push(operation_id.clone()); Ok(()) } @@ -354,12 +395,12 @@ struct TerminalOperationStart<'a> { } struct ParsedTerminalRequest { - terminal_id: Option, + process_id: Option, request: TerminalRequest, } struct ParsedTerminalResponse { - terminal_id: Option, + process_id: Option, result: TerminalResult, } @@ -369,7 +410,7 @@ fn parse_protocol_terminal_request( ) -> ParsedTerminalRequest { // Startup/poll paths usually include a process id at begin time, but plain // exec starts may only learn it in the matching end event. - let terminal_id = payload.process_id.clone(); + let process_id = payload.process_id.clone(); let request = match operation_kind { TerminalOperationKind::ExecCommand => TerminalRequest::ExecCommand { display_command: payload.command.join(" "), @@ -385,7 +426,7 @@ fn parse_protocol_terminal_request( }, }; ParsedTerminalRequest { - terminal_id, + process_id, request, } } @@ -416,11 +457,11 @@ fn parse_dispatch_terminal_request(value: JsonValue) -> Result ParsedTerminalResponse { ParsedTerminalResponse { - terminal_id: payload.process_id, + process_id: payload.process_id, result: TerminalResult { exit_code: Some(payload.exit_code), stdout: payload.aggregated_output, @@ -502,7 +543,7 @@ fn parse_dispatch_terminal_response(value: JsonValue) -> Result Option { } } -fn terminal_id_from_json(value: &JsonValue) -> Option { +fn process_id_from_json(value: &JsonValue) -> Option { match value { JsonValue::String(value) if !value.is_empty() => Some(value.clone()), JsonValue::Number(value) => Some(value.to_string()), diff --git a/codex-rs/rollout-trace/src/reducer/tool/terminal_tests.rs b/codex-rs/rollout-trace/src/reducer/tool/terminal_tests.rs index a42f3bbd0d..cd7ac59146 100644 --- a/codex-rs/rollout-trace/src/reducer/tool/terminal_tests.rs +++ b/codex-rs/rollout-trace/src/reducer/tool/terminal_tests.rs @@ -18,8 +18,11 @@ use crate::raw_event::RawTraceEventPayload; use crate::reducer::test_support::create_started_writer; use crate::reducer::test_support::generic_summary; use crate::reducer::test_support::message; +use crate::reducer::test_support::start_thread; use crate::reducer::test_support::start_turn; +use crate::reducer::test_support::start_turn_for_thread; use crate::reducer::test_support::trace_context; +use crate::reducer::test_support::trace_context_for_thread; use crate::replay_bundle; use crate::writer::TraceWriter; @@ -157,7 +160,7 @@ fn exec_tool_reduces_to_terminal_operation_and_session() -> anyhow::Result<()> { rollout.terminal_operations[&operation_id], TerminalOperation { operation_id: operation_id.clone(), - terminal_id: Some("pty-1".to_string()), + terminal_id: Some("terminal_session:terminal_operation:1".to_string()), tool_call_id: "tool-1".to_string(), kind: TerminalOperationKind::ExecCommand, execution: ExecutionWindow { @@ -193,9 +196,9 @@ fn exec_tool_reduces_to_terminal_operation_and_session() -> anyhow::Result<()> { }, ); assert_eq!( - rollout.terminal_sessions["pty-1"], + rollout.terminal_sessions["terminal_session:terminal_operation:1"], TerminalSession { - terminal_id: "pty-1".to_string(), + terminal_id: "terminal_session:terminal_operation:1".to_string(), thread_id: "thread-root".to_string(), created_by_operation_id: operation_id.clone(), operation_ids: vec![operation_id], @@ -213,90 +216,113 @@ fn exec_tool_reduces_to_terminal_operation_and_session() -> anyhow::Result<()> { } #[test] -fn write_stdin_operation_reuses_existing_terminal_session() -> anyhow::Result<()> { +fn reused_process_ids_keep_sessions_and_stdin_joins_distinct() -> anyhow::Result<()> { let temp = TempDir::new()?; let writer = create_started_writer(&temp)?; start_turn(&writer, "turn-1")?; + start_thread(&writer, "child", "/root/child")?; + start_turn_for_thread(&writer, "child", "child-turn")?; + let mut reused_stdin = None; - let startup_payload = writer.write_json_payload( - RawPayloadKind::ToolRuntimeEvent, - &json!({ - "call_id": "tool-start", - "process_id": "pty-1", - "turn_id": "turn-1", - "command": ["bash"], - "cwd": "/repo" - }), - )?; - writer.append_with_context( - trace_context("turn-1"), - RawTraceEventPayload::ToolCallStarted { - tool_call_id: "tool-start".to_string(), - model_visible_call_id: None, - code_mode_runtime_tool_id: None, - requester: crate::raw_event::RawToolCallRequester::Model, - kind: ToolCallKind::ExecCommand, - summary: generic_summary("exec_command"), - invocation_payload: None, - }, - )?; - writer.append_with_context( - trace_context("turn-1"), - RawTraceEventPayload::ToolCallRuntimeStarted { - tool_call_id: "tool-start".to_string(), - runtime_payload: startup_payload, - }, - )?; - - let stdin_payload = writer.write_json_payload( - RawPayloadKind::ToolRuntimeEvent, - &json!({ - "call_id": "tool-stdin", - "process_id": "pty-1", - "turn_id": "turn-1", - "command": ["bash"], - "cwd": "/repo", - "interaction_input": "echo hi\n" - }), - )?; - let _stdin_start = writer.append_with_context( - trace_context("turn-1"), - RawTraceEventPayload::ToolCallStarted { - tool_call_id: "tool-stdin".to_string(), - model_visible_call_id: None, - code_mode_runtime_tool_id: None, - requester: crate::raw_event::RawToolCallRequester::Model, - kind: ToolCallKind::WriteStdin, - summary: generic_summary("write_stdin"), - invocation_payload: None, - }, - )?; - let stdin_runtime_start = writer.append_with_context( - trace_context("turn-1"), - RawTraceEventPayload::ToolCallRuntimeStarted { - tool_call_id: "tool-stdin".to_string(), - runtime_payload: stdin_payload, - }, - )?; + // Real rollouts reuse process IDs both across threads and after an exec + // finishes. Polls must stay with the corresponding exec, not the last + // occurrence of the process number anywhere in the rollout. + for (index, (thread, turn, kind)) in [ + ("thread-root", "turn-1", ToolCallKind::ExecCommand), + ("child", "child-turn", ToolCallKind::ExecCommand), + ("thread-root", "turn-1", ToolCallKind::WriteStdin), + ("child", "child-turn", ToolCallKind::WriteStdin), + ("thread-root", "turn-1", ToolCallKind::ExecCommand), + ("thread-root", "turn-1", ToolCallKind::WriteStdin), + ] + .into_iter() + .enumerate() + { + if index == 4 { + let response = writer.write_json_payload( + RawPayloadKind::ToolRuntimeEvent, + &json!({"process_id": "123", "exit_code": 0}), + )?; + writer.append_with_context( + trace_context("turn-1"), + RawTraceEventPayload::ToolCallRuntimeEnded { + tool_call_id: "tool-0".to_string(), + status: ExecutionStatus::Completed, + runtime_payload: response, + }, + )?; + } + let tool_call_id = format!("tool-{index}"); + writer.append_with_context( + trace_context_for_thread(thread, turn), + RawTraceEventPayload::ToolCallStarted { + tool_call_id: tool_call_id.clone(), + model_visible_call_id: None, + code_mode_runtime_tool_id: None, + requester: crate::raw_event::RawToolCallRequester::Model, + kind, + summary: generic_summary("terminal"), + invocation_payload: None, + }, + )?; + let request = writer.write_json_payload( + RawPayloadKind::ToolRuntimeEvent, + &json!({ + "process_id": "123", + "command": ["bash"], + "cwd": "/repo", + "interaction_input": (index == 5).then_some("echo hi\n"), + }), + )?; + let request_id = request.raw_payload_id.clone(); + let runtime_start = writer.append_with_context( + trace_context_for_thread(thread, turn), + RawTraceEventPayload::ToolCallRuntimeStarted { + tool_call_id, + runtime_payload: request, + }, + )?; + if index == 5 { + reused_stdin = Some((runtime_start, request_id)); + } + } let rollout = replay_bundle(temp.path())?; - let startup_operation_id = "terminal_operation:1".to_string(); - let stdin_operation_id = "terminal_operation:2".to_string(); - + let groups = rollout + .terminal_sessions + .values() + .map(|session| { + let calls = session + .operation_ids + .iter() + .map(|id| { + let operation = &rollout.terminal_operations[id]; + assert_eq!(operation.terminal_id.as_ref(), Some(&session.terminal_id)); + operation.tool_call_id.as_str() + }) + .collect::>(); + (session.thread_id.as_str(), calls) + }) + .collect::>(); assert_eq!( - rollout.terminal_sessions["pty-1"].operation_ids, - vec![startup_operation_id, stdin_operation_id.clone()], + groups, + vec![ + ("thread-root", vec!["tool-0", "tool-2"]), + ("child", vec!["tool-1", "tool-3"]), + ("thread-root", vec!["tool-4", "tool-5"]), + ], ); + let (runtime_start, request_id) = reused_stdin.unwrap(); assert_eq!( - rollout.terminal_operations[&stdin_operation_id], + rollout.terminal_operations["terminal_operation:6"], TerminalOperation { - operation_id: stdin_operation_id.clone(), - terminal_id: Some("pty-1".to_string()), - tool_call_id: "tool-stdin".to_string(), + operation_id: "terminal_operation:6".to_string(), + terminal_id: Some("terminal_session:terminal_operation:5".to_string()), + tool_call_id: "tool-5".to_string(), kind: TerminalOperationKind::WriteStdin, execution: ExecutionWindow { - started_at_unix_ms: stdin_runtime_start.wall_time_unix_ms, - started_seq: stdin_runtime_start.seq, + started_at_unix_ms: runtime_start.wall_time_unix_ms, + started_seq: runtime_start.seq, ended_at_unix_ms: None, ended_seq: None, status: ExecutionStatus::Running, @@ -308,10 +334,39 @@ fn write_stdin_operation_reuses_existing_terminal_session() -> anyhow::Result<() }, result: None, model_observations: Vec::new(), - raw_payload_ids: vec!["raw_payload:2".to_string()], + raw_payload_ids: vec![request_id], }, ); + // Stdin started while tool-0 owned the handle, but dispatch was delayed + // until tool-4 replaced it. Only its session membership should change. + writer.append_with_context( + trace_context("turn-1"), + RawTraceEventPayload::StdinTargetResolved { + tool_call_id: "tool-2".to_string(), + exec_tool_call_id: "tool-4".to_string(), + }, + )?; + let mut expected = rollout; + let previous = "terminal_session:terminal_operation:1"; + let resolved = "terminal_session:terminal_operation:5"; + expected + .terminal_sessions + .get_mut(previous) + .unwrap() + .operation_ids = vec!["terminal_operation:1".to_string()]; + expected + .terminal_sessions + .get_mut(resolved) + .unwrap() + .operation_ids + .push("terminal_operation:3".to_string()); + expected + .terminal_operations + .get_mut("terminal_operation:3") + .unwrap() + .terminal_id = Some(resolved.to_string()); + assert_eq!(replay_bundle(temp.path())?, expected); Ok(()) } @@ -389,7 +444,7 @@ fn dispatch_write_stdin_payload_reduces_to_terminal_operation() -> anyhow::Resul rollout.terminal_operations[&operation_id], TerminalOperation { operation_id: operation_id.clone(), - terminal_id: Some("123".to_string()), + terminal_id: Some("terminal_session:terminal_operation:1".to_string()), tool_call_id: "tool-stdin".to_string(), kind: TerminalOperationKind::WriteStdin, execution: ExecutionWindow { @@ -417,9 +472,9 @@ fn dispatch_write_stdin_payload_reduces_to_terminal_operation() -> anyhow::Resul }, ); assert_eq!( - rollout.terminal_sessions["123"], + rollout.terminal_sessions["terminal_session:terminal_operation:1"], TerminalSession { - terminal_id: "123".to_string(), + terminal_id: "terminal_session:terminal_operation:1".to_string(), thread_id: "thread-root".to_string(), created_by_operation_id: operation_id.clone(), operation_ids: vec![operation_id], diff --git a/codex-rs/rollout-trace/src/thread.rs b/codex-rs/rollout-trace/src/thread.rs index 514b8d382b..4a38bacaff 100644 --- a/codex-rs/rollout-trace/src/thread.rs +++ b/codex-rs/rollout-trace/src/thread.rs @@ -247,6 +247,25 @@ impl ThreadTraceContext { context.append_with_context_best_effort(codex_turn_id.into(), payload); } + /// Records the exec actually selected by stdin, since its process handle can be reused. + pub fn record_stdin_target( + &self, + codex_turn_id: &str, + tool_call_id: &str, + exec_tool_call_id: &str, + ) { + let ThreadTraceContextState::Enabled(context) = &self.state else { + return; + }; + context.append_with_context_best_effort( + codex_turn_id.to_string(), + RawTraceEventPayload::StdinTargetResolved { + tool_call_id: tool_call_id.to_string(), + exec_tool_call_id: exec_tool_call_id.to_string(), + }, + ); + } + /// Emits the v2 child-to-parent completion message as an explicit graph edge. /// /// The notification is runtime delivery from a completed child turn into