diff --git a/codex-rs/rollout-trace/AGENTS.md b/codex-rs/rollout-trace/AGENTS.md new file mode 100644 index 0000000000..02d94d62b3 --- /dev/null +++ b/codex-rs/rollout-trace/AGENTS.md @@ -0,0 +1,16 @@ +# Rollout trace + +Only the exact Codex binary that wrote a diagnostic trace bundle runs its +reducer. There is no cross-binary replay compatibility, even between builds +with identical version numbers. + +- When changing a raw event or payload, update its writer, reducer, and affected + fixtures together. Do not add migrations, legacy parsers, or tests for another + binary's traces. A fallback needs a case the current writer can produce, such + as a recording interrupted between two events. +- Treat `state.json` as the consumer API and `manifest.json` identity metadata + as the bundle discovery API. For every `state.json` schema change, including + changes to types serialized in both raw events and reduced state, bump + `REDUCED_TRACE_SCHEMA_VERSION` and consider backward compatibility for + consumers. Breaking changes are allowed. Raw-only changes do not require a + reduced-state version bump. diff --git a/codex-rs/rollout-trace/README.md b/codex-rs/rollout-trace/README.md index 368102d4aa..010eaddaf6 100644 --- a/codex-rs/rollout-trace/README.md +++ b/codex-rs/rollout-trace/README.md @@ -11,6 +11,18 @@ during a Codex session. It records raw runtime evidence into a local bundle on disk, then replays that bundle into a semantic graph that a debugger or UI can inspect. +**Same-binary replay only.** The exact Codex executable that recorded a bundle +must reduce it. Matching release or schema versions do not make a different +binary compatible. The raw event log and payload formats are temporary +implementation details and can change with the writer and reducer. + +Viewers and other consumers read `state.json`, the reduced trace API. They can +use `manifest.json` for bundle discovery and identity, such as selecting a trace +by `root_thread_id`. Payload references in the reduced state let consumers +inspect the original evidence. Every change to the `state.json` schema bumps +`REDUCED_TRACE_SCHEMA_VERSION`. Consider the impact on existing consumers. +Breaking changes are allowed. + The key design choice is: **observe first, interpret later**. Hot-path Codex code does not try to build the final graph while the session is @@ -110,20 +122,20 @@ A trace bundle contains: - `trace.jsonl`: append-only raw events ordered by writer-assigned `seq`. - `payloads/*.json`: raw requests, responses, tool inputs/results, runtime events, terminal output, compaction data, and protocol snapshots. -- `state.json`: optional reducer output written by `codex debug trace-reduce`. +- `state.json`: reduced trace for consumers, written by `codex debug trace-reduce`. `trace_id` identifies this diagnostic artifact. `rollout_id` identifies the Codex rollout/session being observed. Keeping those separate lets us reason about the stored trace without confusing it with the product-level session identity. -To reduce a bundle: +Run the recording executable to reduce its bundle: ```bash codex debug trace-reduce ``` -By default this writes `/state.json`. Rust callers can also call -`codex_rollout_trace::replay_bundle` directly. +By default this writes `/state.json`. Within the recording binary, +`codex_rollout_trace::replay_bundle` returns the same reduced graph in memory. ## Raw Evidence vs Reduced Graph diff --git a/codex-rs/rollout-trace/src/bundle.rs b/codex-rs/rollout-trace/src/bundle.rs index 9f0df19c5e..2a72a5d6c2 100644 --- a/codex-rs/rollout-trace/src/bundle.rs +++ b/codex-rs/rollout-trace/src/bundle.rs @@ -8,9 +8,10 @@ use crate::model::AgentThreadId; pub(crate) const MANIFEST_FILE_NAME: &str = "manifest.json"; pub(crate) const RAW_EVENT_LOG_FILE_NAME: &str = "trace.jsonl"; pub(crate) const PAYLOADS_DIR_NAME: &str = "payloads"; -/// Conventional file name for a reducer-written `RolloutTrace` cache. +/// Conventional file name for the reducer's serialized `RolloutTrace`. pub const REDUCED_STATE_FILE_NAME: &str = "state.json"; pub(crate) const TRACE_MANIFEST_SCHEMA_VERSION: u32 = 2; +/// Bump for every change to the consumer-facing `state.json` schema. pub(crate) const REDUCED_TRACE_SCHEMA_VERSION: u32 = 2; /// Manifest stored at the root of a trace bundle. diff --git a/codex-rs/rollout-trace/src/lib.rs b/codex-rs/rollout-trace/src/lib.rs index 54094f0525..1318f70058 100644 --- a/codex-rs/rollout-trace/src/lib.rs +++ b/codex-rs/rollout-trace/src/lib.rs @@ -4,7 +4,8 @@ //! small writer API here; semantic replay and viewer projections stay outside //! `codex-core`. //! -//! See `README.md` for the system diagram and reducer model. +//! Only the recording executable replays raw events. Consumers read the reduced +//! `state.json` and use manifest metadata for bundle discovery. See `README.md`. mod bundle; mod code_cell; @@ -20,7 +21,7 @@ mod thread; mod tool_dispatch; mod writer; -/// Conventional reduced-state cache name written next to a raw trace bundle. +/// Conventional file name for the consumer-facing reduced trace. pub use bundle::REDUCED_STATE_FILE_NAME; /// No-op-capable handle for recording one code-mode runtime cell. pub use code_cell::CodeCellTraceContext; @@ -54,7 +55,7 @@ pub use raw_event::RawTraceEvent; pub use raw_event::RawTraceEventContext; /// Typed payload for one raw trace event. pub use raw_event::RawTraceEventPayload; -/// Replay a raw trace bundle and write/read its reduced `RolloutTrace`. +/// Replay a bundle recorded by this executable into a `RolloutTrace`. pub use reducer::replay_bundle; /// Raw payload captured when a child agent reports completion to its parent. pub use thread::AgentResultTracePayload; diff --git a/codex-rs/rollout-trace/src/protocol_event.rs b/codex-rs/rollout-trace/src/protocol_event.rs index 58481aacb1..5cfe318aac 100644 --- a/codex-rs/rollout-trace/src/protocol_event.rs +++ b/codex-rs/rollout-trace/src/protocol_event.rs @@ -162,8 +162,7 @@ impl Serialize for ToolRuntimePayload<'_> { /// Rollout-trace representation of an exec begin event. /// -/// Rollout traces share the rollout compatibility requirement that paths remain path-flavored -/// strings on disk, even though live events carry `PathUri` internally. +/// The trace records filesystem paths as strings. Live events carry `PathUri`. #[derive(Serialize)] struct ExecCommandBeginTracePayload<'a> { call_id: &'a str, diff --git a/codex-rs/rollout-trace/src/reducer/acceptance_tests.rs b/codex-rs/rollout-trace/src/reducer/acceptance_tests.rs index 5c1113ebc3..74c96d6ebf 100644 --- a/codex-rs/rollout-trace/src/reducer/acceptance_tests.rs +++ b/codex-rs/rollout-trace/src/reducer/acceptance_tests.rs @@ -252,18 +252,9 @@ fn inherited_output_is_child_prefill_without_copying_parent_sampling_provenance( } #[test] -fn legacy_bundles_cannot_claim_accepted_history() -> anyhow::Result<()> { +fn replay_rejects_a_mismatched_manifest_schema() -> anyhow::Result<()> { let temp = TempDir::new()?; - let writer = create_started_writer(&temp)?; - start_turn(&writer, "turn-1")?; - append_completed_inference( - &writer, - "thread-root", - "turn-1", - "call", - vec![message("user", "answer")], - vec![message("assistant", "unproven")], - )?; + create_started_writer(&temp)?; let manifest_path = temp.path().join("manifest.json"); let mut manifest: serde_json::Value = serde_json::from_reader(std::fs::File::open(&manifest_path)?)?; @@ -273,7 +264,7 @@ fn legacy_bundles_cannot_claim_accepted_history() -> anyhow::Result<()> { replay_bundle(temp.path()) .unwrap_err() .to_string() - .contains("cannot establish accepted history") + .contains("unsupported manifest schema") ); Ok(()) } diff --git a/codex-rs/rollout-trace/src/reducer/code_cell.rs b/codex-rs/rollout-trace/src/reducer/code_cell.rs index c06a1c8b6a..742dc7a6b9 100644 --- a/codex-rs/rollout-trace/src/reducer/code_cell.rs +++ b/codex-rs/rollout-trace/src/reducer/code_cell.rs @@ -560,35 +560,6 @@ impl TraceReducer { self.add_code_cell_output_item(&code_cell_id, item_id) } - /// Resolves the owning thread for a code-cell runtime event. - /// - /// Runtime events should carry a thread id, but older/raw paths may only have - /// the turn id. The fallback keeps replay strict while avoiding duplicate logic - /// in every code-cell event arm. - pub(super) fn code_cell_event_thread_id( - &self, - thread_id: Option, - codex_turn_id: Option<&str>, - runtime_cell_id: &str, - event_name: &str, - ) -> Result { - if let Some(thread_id) = thread_id { - return Ok(thread_id); - } - let Some(codex_turn_id) = codex_turn_id else { - bail!("{event_name} {runtime_cell_id} did not include a thread id"); - }; - self.rollout - .codex_turns - .get(codex_turn_id) - .map(|turn| turn.thread_id.clone()) - .with_context(|| { - format!( - "{event_name} {runtime_cell_id} referenced unknown Codex turn {codex_turn_id}" - ) - }) - } - /// Derives the stable reduced code-cell id from the model-visible exec call id. pub(super) fn reduced_code_cell_id_for_model_visible_call( model_visible_call_id: &str, diff --git a/codex-rs/rollout-trace/src/reducer/mod.rs b/codex-rs/rollout-trace/src/reducer/mod.rs index cfd0dac15a..d57608ae6d 100644 --- a/codex-rs/rollout-trace/src/reducer/mod.rs +++ b/codex-rs/rollout-trace/src/reducer/mod.rs @@ -49,7 +49,7 @@ use self::tool::ObservedAgentResultEdge; use self::tool::PendingAgentInteractionEdge; use self::tool::ToolCallStarted; -/// Replays a local trace bundle into a reduced rollout graph. +/// Replays a trace bundle written by this executable into a reduced rollout graph. /// A valid prefix has `recording_complete = false`; malformed events or missing /// or invalid referenced payloads return an error. pub fn replay_bundle(bundle_dir: impl AsRef) -> Result { @@ -57,10 +57,9 @@ pub fn replay_bundle(bundle_dir: impl AsRef) -> Result { let manifest: TraceBundleManifest = serde_json::from_reader(File::open(bundle_dir.join(MANIFEST_FILE_NAME))?) .with_context(|| format!("read {}", bundle_dir.join(MANIFEST_FILE_NAME).display()))?; - // Earlier bundles cannot distinguish accepted output from transport output. ensure!( manifest.schema_version == TRACE_MANIFEST_SCHEMA_VERSION, - "trace schema {} cannot establish accepted history; use its matching reducer or record a new bundle", + "unsupported manifest schema {} (expected {TRACE_MANIFEST_SCHEMA_VERSION}). Run the exact Codex executable that recorded this bundle", manifest.schema_version ); let mut reducer = TraceReducer { @@ -112,7 +111,7 @@ pub fn replay_bundle(bundle_dir: impl AsRef) -> Result { .with_context(|| format!("parse trace event line {line_number}"))?; ensure!( event.schema_version == RAW_TRACE_EVENT_SCHEMA_VERSION, - "unsupported trace event schema {} at line {line_number}", + "unsupported trace event schema {} at line {line_number} (expected {RAW_TRACE_EVENT_SCHEMA_VERSION}). Run the exact Codex executable that recorded this bundle", event.schema_version, ); event_count += 1; @@ -472,12 +471,9 @@ impl TraceReducer { model_visible_call_id, source_js, } => { - let thread_id = self.code_cell_event_thread_id( - event.thread_id, - event.codex_turn_id.as_deref(), - &runtime_cell_id, - "code cell start", - )?; + let thread_id = event.thread_id.with_context(|| { + format!("code cell start {runtime_cell_id} did not include a thread id") + })?; let reduced_code_cell_id = Self::reduced_code_cell_id_for_model_visible_call(&model_visible_call_id); self.record_runtime_code_cell_id( @@ -503,12 +499,11 @@ impl TraceReducer { status, response_payload, } => { - let thread_id = self.code_cell_event_thread_id( - event.thread_id, - event.codex_turn_id.as_deref(), - &runtime_cell_id, - "code cell initial response", - )?; + let thread_id = event.thread_id.with_context(|| { + format!( + "code cell initial response {runtime_cell_id} did not include a thread id" + ) + })?; let code_cell_id = self.code_cell_id_for_runtime_cell_id( &thread_id, &runtime_cell_id, @@ -528,12 +523,9 @@ impl TraceReducer { status, response_payload, } => { - let thread_id = self.code_cell_event_thread_id( - event.thread_id, - event.codex_turn_id.as_deref(), - &runtime_cell_id, - "code cell end", - )?; + let thread_id = event.thread_id.with_context(|| { + format!("code cell end {runtime_cell_id} did not include a thread id") + })?; let code_cell_id = self.code_cell_id_for_runtime_cell_id( &thread_id, &runtime_cell_id, diff --git a/codex-rs/rollout-trace/src/reducer/tool/agents.rs b/codex-rs/rollout-trace/src/reducer/tool/agents.rs index c7274825aa..b395eb15d2 100644 --- a/codex-rs/rollout-trace/src/reducer/tool/agents.rs +++ b/codex-rs/rollout-trace/src/reducer/tool/agents.rs @@ -6,14 +6,12 @@ use codex_protocol::protocol::CollabAgentInteractionEndEvent; use codex_protocol::protocol::CollabAgentSpawnEndEvent; use codex_protocol::protocol::CollabCloseBeginEvent; use codex_protocol::protocol::CollabCloseEndEvent; -use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::SubAgentActivityEvent; use codex_protocol::protocol::SubAgentActivityKind; use serde::Deserialize; use serde_json::Value; use super::super::TraceReducer; -use crate::model::ConversationItem; use crate::model::ConversationItemKind; use crate::model::ConversationPart; use crate::model::ConversationRole; @@ -705,13 +703,25 @@ impl TraceReducer { fn inter_agent_message_item(&self, item_id: &str) -> Option<(String, String, String)> { let item = self.rollout.conversation_items.get(item_id)?; - let (author_agent_path, recipient_agent_path, message_content) = - inter_agent_message_fields(item)?; + let delivery = item.agent_message.as_ref()?; let thread = self.rollout.threads.get(&item.thread_id)?; - if recipient_agent_path != thread.agent_path { + if delivery.recipient != thread.agent_path { return None; } - Some((item.thread_id.clone(), author_agent_path, message_content)) + let content = match item.body.parts.as_slice() { + [ConversationPart::Text { text }] => text, + [ConversationPart::Encoded { label, value }] if label == "encrypted_content" => value, + [ + ConversationPart::Text { .. }, + ConversationPart::Encoded { label, value }, + ] if label == "encrypted_content" => value, + _ => return None, + }; + Some(( + item.thread_id.clone(), + delivery.author.clone(), + content.clone(), + )) } fn agent_path_for_thread(&self, thread_id: &str) -> Result { @@ -771,44 +781,6 @@ fn push_unique(items: &mut Vec, item: &str) { } } -fn inter_agent_message_fields(item: &ConversationItem) -> Option<(String, String, String)> { - if item.role != ConversationRole::Assistant || item.kind != ConversationItemKind::Message { - return None; - } - if let Some(agent_message) = &item.agent_message { - let message_content = match item.body.parts.as_slice() { - [ConversationPart::Text { text }] => text, - [ConversationPart::Encoded { label, value }] if label == "encrypted_content" => value, - [ - ConversationPart::Text { .. }, - ConversationPart::Encoded { label, value }, - ] if label == "encrypted_content" => value, - _ => return None, - }; - return Some(( - agent_message.author.clone(), - agent_message.recipient.clone(), - message_content.clone(), - )); - } - - // Older traces store multi-agent v2 deliveries as assistant messages whose - // text is serialized `InterAgentCommunication`. Treat only that exact - // transport shape as an edge target; ordinary assistant JSON must not be - // mistaken for cross-thread delivery. - let [ConversationPart::Text { text }] = item.body.parts.as_slice() else { - return None; - }; - let communication = serde_json::from_str::(text).ok()?; - Some(( - communication.author.to_string(), - communication.recipient.to_string(), - communication - .encrypted_content - .unwrap_or(communication.content), - )) -} - #[cfg(test)] #[path = "agents_tests.rs"] mod tests; diff --git a/codex-rs/rollout-trace/src/reducer/tool/agents_tests.rs b/codex-rs/rollout-trace/src/reducer/tool/agents_tests.rs index 4733233ce1..293a78927d 100644 --- a/codex-rs/rollout-trace/src/reducer/tool/agents_tests.rs +++ b/codex-rs/rollout-trace/src/reducer/tool/agents_tests.rs @@ -104,18 +104,13 @@ fn spawn_runtime_payload_targets_delivered_child_message() -> anyhow::Result<()> "019d0000-0000-7000-8000-000000000002", "turn-child-1", )?; - let delivered = inter_agent_message( - "/root", - "/root/repo_file_counter", - "count", - /*trigger_turn*/ true, - ); + let delivered = inter_agent_message("/root", "/root/repo_file_counter", "count"); append_inference_request( &writer, "019d0000-0000-7000-8000-000000000002", "turn-child-1", "inference-child-1", - vec![message("assistant", &delivered)], + vec![delivered], )?; let replayed = replay_bundle(temp.path())?; @@ -403,14 +398,13 @@ fn send_message_runtime_payload_targets_delivered_child_message() -> anyhow::Res "019d0000-0000-7000-8000-000000000002", "turn-child-1", )?; - let delivered = - inter_agent_message("/root", "/root/child", "hello", /*trigger_turn*/ false); + let delivered = inter_agent_message("/root", "/root/child", "hello"); append_inference_request( &writer, "019d0000-0000-7000-8000-000000000002", "turn-child-1", "inference-child-1", - vec![message("assistant", &delivered)], + vec![delivered], )?; let replayed = replay_bundle(temp.path())?; @@ -485,18 +479,13 @@ fn send_message_activity_targets_delivered_child_message() -> anyhow::Result<()> )?; start_thread(&writer, child_thread_id, "/root/child")?; start_turn_for_thread(&writer, child_thread_id, "turn-child-1")?; - let delivered = inter_agent_message( - "/root", - "/root/child", - "hello again", - /*trigger_turn*/ false, - ); + let delivered = inter_agent_message("/root", "/root/child", "hello again"); append_inference_request( &writer, child_thread_id, "turn-child-1", "inference-child-1", - vec![message("assistant", &delivered)], + vec![delivered], )?; let replayed = replay_bundle(temp.path())?; @@ -571,18 +560,13 @@ fn followup_activity_targets_delivered_child_message() -> anyhow::Result<()> { )?; start_thread(&writer, child_thread_id, "/root/child")?; start_turn_for_thread(&writer, child_thread_id, "turn-child-1")?; - let delivered = inter_agent_message( - "/root", - "/root/child", - "continue", - /*trigger_turn*/ true, - ); + let delivered = inter_agent_message("/root", "/root/child", "continue"); append_inference_request( &writer, child_thread_id, "turn-child-1", "inference-child-1", - vec![message("assistant", &delivered)], + vec![delivered], )?; let replayed = replay_bundle(temp.path())?; @@ -774,18 +758,13 @@ fn agent_result_edge_links_child_result_to_parent_notification() -> anyhow::Resu )?; start_agent_turn(&writer, "turn-root-1")?; - let delivered = inter_agent_message( - "/root/child", - "/root", - notification, - /*trigger_turn*/ false, - ); + let delivered = inter_agent_message("/root/child", "/root", notification); append_inference_request( &writer, "019d0000-0000-7000-8000-000000000001", "turn-root-1", "inference-root-1", - vec![message("assistant", &delivered)], + vec![delivered], )?; let replayed = replay_bundle(temp.path())?; @@ -872,18 +851,13 @@ fn agent_result_edge_falls_back_to_child_thread_without_result_message() -> anyh // mailbox item. The target should remain that precise parent-side // ConversationItem even though the source falls back to the child thread. start_agent_turn(&writer, "turn-root-1")?; - let delivered = inter_agent_message( - "/root/child", - "/root", - notification, - /*trigger_turn*/ false, - ); + let delivered = inter_agent_message("/root/child", "/root", notification); append_inference_request( &writer, "019d0000-0000-7000-8000-000000000001", "turn-root-1", "inference-root-1", - vec![message("assistant", &delivered)], + vec![delivered], )?; let replayed = replay_bundle(temp.path())?; @@ -1009,15 +983,13 @@ fn append_spawn_agent_tool_lifecycle( }) } -fn inter_agent_message(author: &str, recipient: &str, content: &str, trigger_turn: bool) -> String { +fn inter_agent_message(author: &str, recipient: &str, content: &str) -> serde_json::Value { json!({ + "type": "agent_message", "author": author, "recipient": recipient, - "other_recipients": [], - "content": content, - "trigger_turn": trigger_turn, + "content": [{"type": "input_text", "text": content}] }) - .to_string() } fn target_conversation_item_id(anchor: &TraceAnchor) -> &String { diff --git a/codex-rs/rollout-trace/src/reducer/tool/terminal.rs b/codex-rs/rollout-trace/src/reducer/tool/terminal.rs index d2b21cc127..536438fc5a 100644 --- a/codex-rs/rollout-trace/src/reducer/tool/terminal.rs +++ b/codex-rs/rollout-trace/src/reducer/tool/terminal.rs @@ -6,8 +6,9 @@ //! `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. +//! distinct reduced session. A stdin invocation initially joins by handle. Once +//! the runtime resolves the target, that event identifies the exact exec. A +//! partial recording can end between invocation and resolution. use anyhow::Context; use anyhow::Result;