openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-23 18:08:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Keep rollout trace terminal sessions distinct when process IDs are reused (#56011)

Browse Source 
## Why

Runtime process handles are recyclable and thread-local. Using them as terminal session identities can merge separate commands or associate `write_stdin` operations with the wrong exec.

## What changed

- Give each exec a distinct terminal session and track the latest session for each thread-local process handle.
- Record the exec selected by `write_stdin` at runtime and use it to correct session membership during replay, including when dispatch is delayed until a handle is reused.
- Fall back to the latest session for the thread and handle when replaying older traces without runtime target events.

## Testing

Extend reducer coverage for process ID reuse across threads and successive execs, stdin session membership, and delayed runtime target resolution.

GitOrigin-RevId: 330c6fd26a4eebdef03ae600a030dc76920e0dd9
This commit is contained in:
cassirer-openai
2026-10-21 23:13:31 +00:00
committed by copyberry
parent 082900149c
commit deede7201d
8 changed files with 306 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/rollout-trace/src/model/runtime.rs
+2 -2
View File
@@ -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<TerminalId>,
pub tool_call_id: ToolCallId,
pub kind: TerminalOperationKind,
codex-rs/rollout-trace/src/raw_event.rs
+6
View File
@@ -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,
..
codex-rs/rollout-trace/src/reducer/mod.rs
+12
View File
@@ -57,6 +57,8 @@ pub fn replay_bundle(bundle_dir: impl AsRef<Path>) -> Result<RolloutTrace> {
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<String, String>,
/// 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,
codex-rs/rollout-trace/src/reducer/tool.rs
+2 -3
View File
@@ -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,
codex-rs/rollout-trace/src/reducer/tool/terminal.rs
+122 -81
View File
@@ -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<Option<TerminalOperationId>> {
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<String>,
process_id: Option<String>,
request: TerminalRequest,
}
struct ParsedTerminalResponse {
terminal_id: Option<String>,
process_id: Option<String>,
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<Option<ParsedTerm
.context("write_stdin dispatch payload omitted function arguments")?;
let args: DispatchedWriteStdinArgs = serde_json::from_str(&arguments)
.context("parse write_stdin dispatch function arguments")?;
let terminal_id = terminal_id_from_json(&args.session_id)
let process_id = process_id_from_json(&args.session_id)
.context("write_stdin dispatch payload omitted session_id")?;
Ok(Some(ParsedTerminalRequest {
terminal_id: Some(terminal_id),
process_id: Some(process_id),
request: TerminalRequest::WriteStdin {
stdin: args.chars,
yield_time_ms: args.yield_time_ms,
@@ -456,7 +497,7 @@ fn parse_terminal_response_payload(
fn parse_protocol_terminal_response(payload: ExecCommandEndPayload) -> 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<ParsedTerminalRe
},
};
Ok(ParsedTerminalResponse {
terminal_id: None,
process_id: None,
result,
})
}
@@ -547,7 +588,7 @@ fn json_text_content(value: &JsonValue) -> Option<String> {
}
}
fn terminal_id_from_json(value: &JsonValue) -> Option<String> {
fn process_id_from_json(value: &JsonValue) -> Option<String> {
match value {
JsonValue::String(value) if !value.is_empty() => Some(value.clone()),
JsonValue::Number(value) => Some(value.to_string()),
codex-rs/rollout-trace/src/reducer/tool/terminal_tests.rs
+134 -79
View File
@@ -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::<Vec<_>>();
(session.thread_id.as_str(), calls)
})
.collect::<Vec<_>>();
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],
codex-rs/rollout-trace/src/thread.rs
+19
View File
@@ -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