From 297f5e33a5f917863a7b83f262c9ef1431bc0267 Mon Sep 17 00:00:00 2001 From: cassirer-openai Date: Thu, 29 Oct 2026 19:43:45 +0000 Subject: [PATCH] Make rollout recording failures and completion explicit (#58166) ## Why Enabled rollout tracing could silently fail to initialize, and a terminal task status did not establish whether all recording had finished. Trace consumers need to distinguish a complete recording from a partial bundle, including when a task is interrupted. ## What changed - Fail thread startup or resume if an enabled recorder cannot initialize, while cleaning up session resources. - Stop recording after the first write or serialization failure and preserve that failure across shared producers. - Append a closing record when the last writer handle is released without recording errors. Expose `recording_complete` in `state.json` independently of task success, and mark submission-channel closure as an aborted rollout. - Reject replay logs with missing or reordered events, mismatched rollout IDs, events after completion, or missing or invalid JSON payloads. Valid prefixes remain replayable with `recording_complete` set to `false`. ## Testing Add coverage for startup failure and recovery, shared-writer failures, finalization after child handles close, incomplete logs, and missing or malformed payloads. Extend session and CLI tests to check recording completion after shutdown, interruption, stream failure, and submission-channel closure. GitOrigin-RevId: 30d62fa95912fb0442ad1199776d110884066c30 --- codex-rs/cli/tests/rollout_trace.rs | 43 ++++++++ codex-rs/core/src/session/handlers.rs | 5 +- codex-rs/core/src/session/session.rs | 21 +++- codex-rs/core/src/session/tests.rs | 32 +++++- codex-rs/rollout-trace/README.md | 9 +- codex-rs/rollout-trace/src/compaction.rs | 9 +- codex-rs/rollout-trace/src/inference.rs | 42 ++++---- codex-rs/rollout-trace/src/model/mod.rs | 4 + codex-rs/rollout-trace/src/raw_event.rs | 5 +- codex-rs/rollout-trace/src/reducer/mod.rs | 36 ++++++- codex-rs/rollout-trace/src/thread.rs | 59 +++++------ codex-rs/rollout-trace/src/thread_tests.rs | 73 +++++++++++++- codex-rs/rollout-trace/src/writer.rs | 112 +++++++++++++++++++-- 13 files changed, 368 insertions(+), 82 deletions(-) diff --git a/codex-rs/cli/tests/rollout_trace.rs b/codex-rs/cli/tests/rollout_trace.rs index c9bfeb2492..8f8e106d16 100644 --- a/codex-rs/cli/tests/rollout_trace.rs +++ b/codex-rs/cli/tests/rollout_trace.rs @@ -6,6 +6,7 @@ use app_test_support::TestAppServer; use app_test_support::create_mock_responses_server_sequence; use codex_app_server_protocol::AgentMessageDeltaNotification; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::TurnCompletedNotification; @@ -27,6 +28,46 @@ use serde_json::json; use tempfile::TempDir; use tokio::sync::oneshot; +#[tokio::test] +async fn enabled_recording_cannot_silently_fail_to_start() -> Result<()> { + let codex = codex_utils_cargo_bin::cargo_bin("codex")?; + let home = TempDir::new()?; + let server = create_mock_responses_server_sequence(Vec::new()).await; + MockResponsesConfig::new(&server.uri()).write(home.path())?; + let trace_root = home.path().join("traces"); + std::fs::write(&trace_root, "file")?; + let mut app = TestAppServer::builder() + .with_program(&codex) + .with_plugin_startup_tasks() + .with_args(&["app-server"]) + .with_codex_home(home.path()) + .with_env_overrides(&[("CODEX_ROLLOUT_TRACE_ROOT", trace_root.to_str())]) + .build_initialized() + .await?; + let request_id = app + .send_thread_start_request_with_auto_env(ThreadStartParams::default()) + .await?; + let error = app + .read_stream_until_error_message(RequestId::Integer(request_id)) + .await?; + assert!( + error + .error + .message + .contains("failed to start rollout recording") + ); + // The same app-server can start a new recording once storage is available. + std::fs::remove_file(&trace_root)?; + std::fs::create_dir(&trace_root)?; + app.start_thread(ThreadStartParams::default()).await?; + assert!(app.shutdown_gracefully().await?.success()); + let mut bundles = std::fs::read_dir(trace_root)?; + let bundle = bundles.next().unwrap()?.path(); + assert!(bundles.next().is_none()); + assert!(codex_rollout_trace::replay_bundle(bundle)?.recording_complete); + Ok(()) +} + #[tokio::test] async fn accepted_output_survives_success_and_stream_failure() -> Result<()> { let codex = codex_utils_cargo_bin::cargo_bin("codex")?; @@ -116,6 +157,7 @@ async fn accepted_output_survives_success_and_stream_failure() -> Result<()> { ); let rollout: RolloutTrace = serde_json::from_reader(std::fs::File::open(bundle.join("state.json"))?)?; + assert!(rollout.recording_complete); let call = rollout.inference_calls.values().next().unwrap(); let output_id = &call.response_item_ids[0]; let output = &rollout.conversation_items[output_id]; @@ -234,6 +276,7 @@ async fn accepted_output_survives_interruption_and_shutdown() -> Result<()> { ); let rollout: RolloutTrace = serde_json::from_reader(std::fs::File::open(bundle.join("state.json"))?)?; + assert!(rollout.recording_complete); assert_eq!(rollout.inference_calls.len(), 1); let call = rollout.inference_calls.values().next().unwrap(); assert_eq!(call.execution.status, ExecutionStatus::Cancelled); diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index b66d272d6f..2564c5663a 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -428,10 +428,10 @@ pub async fn shutdown(sess: &Arc, sub_id: String) -> bool { sess.services .rollout_thread_trace .record_protocol_event(&event.msg); - sess.deliver_event_raw(event).await; sess.services .rollout_thread_trace .record_ended(codex_rollout_trace::RolloutStatus::Completed); + sess.deliver_event_raw(event).await; true } @@ -793,6 +793,9 @@ pub(super) async fn submission_loop( ); warn!("failed to shutdown thread persistence after submission channel closed: {err}"); } + sess.services + .rollout_thread_trace + .record_ended(codex_rollout_trace::RolloutStatus::Aborted); } debug!("Agent loop exited"); } diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 37c22bc42d..86b6cd600f 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1779,13 +1779,18 @@ impl Session { // parent had no trace bundle, do not create an orphan child bundle // that looks like an independent rollout. if matches!(&initial_history, InitialHistory::Resumed(_)) { - parent_rollout_thread_trace - .resume_child_thread_trace_or_disabled(thread_id.to_string()) + parent_rollout_thread_trace.resume_child_thread_trace(thread_id.to_string()) } else { - parent_rollout_thread_trace.start_child_thread_trace_or_disabled(trace_metadata) + parent_rollout_thread_trace.start_child_thread_trace(trace_metadata) } } else { - ThreadTraceContext::start_root_or_disabled(trace_metadata) + ThreadTraceContext::start_root(trace_metadata) + }; + // Extensions have already started. On error, finish constructing the + // session so normal teardown can release everything they registered. + let (rollout_thread_trace, trace_error) = match rollout_thread_trace { + Ok(trace) => (trace, None), + Err(err) => (ThreadTraceContext::disabled(), Some(err)), }; let services = SessionServices { // Start with an empty connection set. The initialized set is @@ -1934,6 +1939,14 @@ impl Session { if let Some(startup) = &startup { startup.set_session(Arc::clone(&sess)); } + if let Some(err) = trace_error { + if startup.is_none() { + super::handlers::shutdown_session_runtime(&sess).await; + } + return Err( + CodexErr::Fatal(format!("failed to start rollout recording: {err:#}")).into(), + ); + } if let Some(network_policy_decider_session) = network_policy_decider_session { let mut guard = network_policy_decider_session.write().await; *guard = Arc::downgrade(&sess); diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 0546aa2edc..9cf370f6e6 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -104,6 +104,10 @@ use codex_protocol::turn_input::TurnInput as SubmittedTurnInput; use codex_protocol::turn_input::TurnInputMode; use codex_protocol::turn_input::TurnInputRequest; use codex_protocol::turn_input::TurnInputSubmission; +use codex_rollout_trace::RolloutStatus; +use codex_rollout_trace::ThreadStartedTraceMetadata; +use codex_rollout_trace::ThreadTraceContext; +use codex_rollout_trace::replay_bundle; use codex_tools::ToolSpec; use codex_utils_path_uri::PathUri; use std::collections::BTreeMap; @@ -8925,7 +8929,7 @@ async fn shutdown_complete_does_not_append_to_thread_store_after_shutdown() { } #[tokio::test] -async fn submission_loop_channel_close_runs_full_thread_teardown() { +async fn submission_loop_channel_close_runs_full_thread_teardown() -> anyhow::Result<()> { struct SessionStopMarker; struct ThreadStopMarker; @@ -8955,6 +8959,24 @@ async fn submission_loop_channel_close_runs_full_thread_teardown() { let store = Arc::new(codex_thread_store::InMemoryThreadStore::default()); let thread_store: Arc = store.clone(); let config = session.get_config().await; + let traces = tempfile::tempdir()?; + session.services.rollout_thread_trace = ThreadTraceContext::start_root_in_root_for_test( + traces.path(), + ThreadStartedTraceMetadata { + thread_id: session.thread_id.to_string(), + agent_path: "/root".to_string(), + task_name: None, + nickname: None, + agent_role: None, + session_source: SessionSource::Exec, + cwd: config.cwd.to_path_buf(), + rollout_path: None, + model: turn_context.model_info().slug.clone(), + provider_name: config.model_provider_id.clone(), + approval_policy: "never".to_string(), + sandbox_policy: "danger-full-access".to_string(), + }, + )?; let live_thread = LiveThread::create( Arc::clone(&thread_store), CreateThreadParams { @@ -9014,6 +9036,13 @@ async fn submission_loop_channel_close_runs_full_thread_teardown() { let session = Arc::new(session); submission_loop(session, Arc::clone(&turn_context.config), rx_sub).await; + // Channel closure aborts the rollout, but releasing the session still + // closes its recording normally, so the resulting bundle is complete. + let bundle = std::fs::read_dir(traces.path())?.next().unwrap()?.path(); + let rollout = replay_bundle(bundle)?; + assert_eq!(rollout.status, RolloutStatus::Aborted); + assert!(rollout.recording_complete); + assert_eq!(1, calls.load(std::sync::atomic::Ordering::SeqCst)); assert_eq!( codex_thread_store::InMemoryThreadStoreCalls { @@ -9023,6 +9052,7 @@ async fn submission_loop_channel_close_runs_full_thread_teardown() { }, store.calls().await ); + Ok(()) } #[tokio::test] diff --git a/codex-rs/rollout-trace/README.md b/codex-rs/rollout-trace/README.md index 540a494638..368102d4aa 100644 --- a/codex-rs/rollout-trace/README.md +++ b/codex-rs/rollout-trace/README.md @@ -96,10 +96,11 @@ one from `CODEX_ROLLOUT_TRACE_ROOT`; fresh spawned child threads derive their own context from the parent's context so the whole rollout tree shares one writer. Disabled contexts accept the same calls and record nothing. -Trace startup and writes are best-effort. Rollout tracing must never make a -Codex session fail just because diagnostic recording failed. Core emits raw -observations; this crate owns the bundle schema, trace-context APIs, writer, and -reducer. +When tracing is enabled, a thread can start or resume only if its recorder +initializes successfully. A later write failure stops recording. Partial +bundles remain available for replay. The reducer sets `recording_complete` in +`state.json` when it reads the closing record written by the last trace handle, +including when the recorded task was interrupted. ## Bundle Layout diff --git a/codex-rs/rollout-trace/src/compaction.rs b/codex-rs/rollout-trace/src/compaction.rs index c8f18795ab..a5c18186f8 100644 --- a/codex-rs/rollout-trace/src/compaction.rs +++ b/codex-rs/rollout-trace/src/compaction.rs @@ -12,10 +12,9 @@ use std::sync::atomic::Ordering; use codex_protocol::models::ResponseItem; use serde::Serialize; -use serde_json::Value as JsonValue; use tracing::warn; -use crate::inference::trace_response_item_json; +use crate::inference::TraceResponseItems; use crate::model::AgentThreadId; use crate::model::CodexTurnId; use crate::model::CompactionId; @@ -71,8 +70,8 @@ struct EnabledCompactionTraceAttempt { } #[derive(Serialize)] -struct TracedCompactionCompleted { - output_items: Vec, +struct TracedCompactionCompleted<'a> { + output_items: TraceResponseItems<'a>, } /// History replacement checkpoint persisted when compaction installs new live history. @@ -215,7 +214,7 @@ impl CompactionTraceAttempt { return; }; let response_payload = TracedCompactionCompleted { - output_items: output_items.iter().map(trace_response_item_json).collect(), + output_items: TraceResponseItems(output_items), }; let Some(response_payload) = write_json_payload_best_effort( &attempt.context.writer, diff --git a/codex-rs/rollout-trace/src/inference.rs b/codex-rs/rollout-trace/src/inference.rs index aec924605f..5411420f61 100644 --- a/codex-rs/rollout-trace/src/inference.rs +++ b/codex-rs/rollout-trace/src/inference.rs @@ -15,6 +15,9 @@ use codex_protocol::protocol::TokenUsage; use http::HeaderMap; use http::HeaderValue; use serde::Serialize; +use serde::Serializer; +use serde::ser::Error; +use serde::ser::SerializeSeq; use serde_json::Value as JsonValue; use uuid::Uuid; @@ -89,7 +92,7 @@ struct TracedResponseStreamOutput<'a> { response_id: Option<&'a str>, upstream_request_id: Option<&'a str>, token_usage: Option<&'a TokenUsage>, - output_items: Vec, + output_items: TraceResponseItems<'a>, } impl InferenceTraceContext { @@ -324,12 +327,8 @@ impl InferenceTraceAttempt { /// The protocol serializer intentionally omits some readable reasoning content /// when shaping items for later model requests. Rollout traces need the item as /// Codex received it, so this helper restores that content in the raw payload. -pub(crate) fn trace_response_item_json(item: &ResponseItem) -> JsonValue { - let mut value = serde_json::to_value(item).unwrap_or_else(|err| { - serde_json::json!({ - "serialization_error": err.to_string(), - }) - }); +fn trace_response_item_json(item: &ResponseItem) -> serde_json::Result { + let mut value = serde_json::to_value(item)?; if let ResponseItem::Reasoning { content: Some(content), @@ -337,17 +336,24 @@ pub(crate) fn trace_response_item_json(item: &ResponseItem) -> JsonValue { } = item && let JsonValue::Object(object) = &mut value { - object.insert( - "content".to_string(), - serde_json::to_value(content).unwrap_or_else(|err| { - serde_json::json!({ - "serialization_error": err.to_string(), - }) - }), - ); + object.insert("content".to_string(), serde_json::to_value(content)?); } - value + Ok(value) +} + +/// Serializes response evidence one item at a time, propagating failures to the writer. +pub(crate) struct TraceResponseItems<'a>(pub &'a [ResponseItem]); + +impl Serialize for TraceResponseItems<'_> { + fn serialize(&self, serializer: S) -> Result { + let mut sequence = serializer.serialize_seq(Some(self.0.len()))?; + for item in self.0 { + let value = trace_response_item_json(item).map_err(S::Error::custom)?; + sequence.serialize_element(&value)?; + } + sequence.end() + } } fn next_inference_call_id() -> InferenceCallId { @@ -373,7 +379,7 @@ fn write_response_payload_best_effort( response_id, upstream_request_id, token_usage, - output_items: output_items.iter().map(trace_response_item_json).collect(), + output_items: TraceResponseItems(output_items), }; write_json_payload_best_effort( &attempt.context.writer, @@ -515,7 +521,7 @@ mod tests { }; let normal = serde_json::to_value(&item).expect("response item serializes"); - let traced = trace_response_item_json(&item); + let traced = trace_response_item_json(&item).expect("trace item serializes"); assert_eq!(normal.get("content"), None); assert_eq!( diff --git a/codex-rs/rollout-trace/src/model/mod.rs b/codex-rs/rollout-trace/src/model/mod.rs index 0166121da9..438b0aca64 100644 --- a/codex-rs/rollout-trace/src/model/mod.rs +++ b/codex-rs/rollout-trace/src/model/mod.rs @@ -67,6 +67,9 @@ pub struct RolloutTrace { /// Wall-clock timestamp for terminal rollout status. `None` means running or partial trace. pub ended_at_unix_ms: Option, pub status: RolloutStatus, + /// All recorder handles closed without a recorded write failure; not task success. + #[serde(default)] + pub recording_complete: bool, pub root_thread_id: AgentThreadId, pub threads: BTreeMap, pub codex_turns: BTreeMap, @@ -105,6 +108,7 @@ impl RolloutTrace { started_at_unix_ms, ended_at_unix_ms: None, status: RolloutStatus::Running, + recording_complete: false, root_thread_id, threads: BTreeMap::new(), codex_turns: BTreeMap::new(), diff --git a/codex-rs/rollout-trace/src/raw_event.rs b/codex-rs/rollout-trace/src/raw_event.rs index 1d70d9ec79..ba634b63a1 100644 --- a/codex-rs/rollout-trace/src/raw_event.rs +++ b/codex-rs/rollout-trace/src/raw_event.rs @@ -67,6 +67,8 @@ pub enum RawToolCallRequester { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "snake_case", tag = "type")] pub enum RawTraceEventPayload { + /// Final record after all producers release a writer with no recorded errors. + RecordingCompleted, RolloutStarted { trace_id: String, root_thread_id: AgentThreadId, @@ -251,7 +253,8 @@ impl RawTraceEventPayload { /// Raw payload refs that must exist before this raw event is appended. pub(crate) fn raw_payload_refs(&self) -> Vec<&RawPayloadRef> { match self { - RawTraceEventPayload::RolloutStarted { .. } + RawTraceEventPayload::RecordingCompleted + | RawTraceEventPayload::RolloutStarted { .. } | RawTraceEventPayload::RolloutEnded { .. } | RawTraceEventPayload::ThreadEnded { .. } | RawTraceEventPayload::ThreadResumed { .. } diff --git a/codex-rs/rollout-trace/src/reducer/mod.rs b/codex-rs/rollout-trace/src/reducer/mod.rs index 467796ea23..81d052db66 100644 --- a/codex-rs/rollout-trace/src/reducer/mod.rs +++ b/codex-rs/rollout-trace/src/reducer/mod.rs @@ -12,6 +12,7 @@ use anyhow::Context; use anyhow::Result; use anyhow::bail; use anyhow::ensure; +use serde::de::IgnoredAny; use serde_json::Value; use crate::bundle::MANIFEST_FILE_NAME; @@ -49,6 +50,8 @@ use self::tool::PendingAgentInteractionEdge; use self::tool::ToolCallStarted; /// Replays a local trace bundle 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 { let bundle_dir = bundle_dir.as_ref(); let manifest: TraceBundleManifest = @@ -85,21 +88,43 @@ pub fn replay_bundle(bundle_dir: impl AsRef) -> Result { let event_log_path = bundle_dir.join(RAW_EVENT_LOG_FILE_NAME); let event_log = File::open(&event_log_path) .with_context(|| format!("open trace event log {}", event_log_path.display()))?; + let mut event_count = 0; for (line_index, line) in BufReader::new(event_log).lines().enumerate() { - let line = line.with_context(|| format!("read trace event line {}", line_index + 1))?; + let line_number = line_index + 1; + let line = line.with_context(|| format!("read trace event line {line_number}"))?; if line.trim().is_empty() { continue; } + ensure!( + !reducer.rollout.recording_complete, + "trace event after recording completed" + ); let event: RawTraceEvent = serde_json::from_str(&line) - .with_context(|| format!("parse trace event line {}", line_index + 1))?; + .with_context(|| format!("parse trace event line {line_number}"))?; ensure!( event.schema_version == RAW_TRACE_EVENT_SCHEMA_VERSION, - "unsupported trace event schema {} at line {}", + "unsupported trace event schema {} at line {line_number}", event.schema_version, - line_index + 1 + ); + event_count += 1; + ensure!( + event.seq == event_count, + "missing or reordered trace event {event_count}" + ); + ensure!( + event.rollout_id == reducer.rollout.rollout_id, + "trace event belongs to another rollout" ); reducer.apply_event(event)?; } + for payload in reducer.rollout.raw_payloads.values() { + let file = File::open(reducer.bundle_dir.join(&payload.path)) + .with_context(|| format!("missing payload {}", payload.raw_payload_id))?; + // Validate even opaque bodies without allocating their strings or objects. + // A nonempty file alone is not evidence that a copy finished writing JSON. + serde_json::from_reader::<_, IgnoredAny>(BufReader::new(file)) + .with_context(|| format!("invalid payload {}", payload.raw_payload_id))?; + } // Spawn edges prefer the child task message as their target, but a child can // fail before that message is ever reduced. Only after replaying the whole // bundle do we know which spawn deliveries need the child-thread fallback. @@ -210,6 +235,9 @@ impl TraceReducer { } match event.payload { + RawTraceEventPayload::RecordingCompleted => { + self.rollout.recording_complete = true; + } RawTraceEventPayload::RolloutStarted { trace_id, root_thread_id, diff --git a/codex-rs/rollout-trace/src/thread.rs b/codex-rs/rollout-trace/src/thread.rs index db03f06da1..c96cc30176 100644 --- a/codex-rs/rollout-trace/src/thread.rs +++ b/codex-rs/rollout-trace/src/thread.rs @@ -100,23 +100,12 @@ impl ThreadTraceContext { } } - /// Starts a root thread trace from `CODEX_ROLLOUT_TRACE_ROOT`, or disables tracing. - /// - /// Trace startup is best-effort. A tracing failure must not make the Codex - /// session unusable, because traces are diagnostic and can be enabled while - /// debugging unrelated production failures. - pub fn start_root_or_disabled(metadata: ThreadStartedTraceMetadata) -> Self { + /// Starts tracing when configured; an enabled recorder must initialize successfully. + pub fn start_root(metadata: ThreadStartedTraceMetadata) -> anyhow::Result { let Some(root) = std::env::var_os(CODEX_ROLLOUT_TRACE_ROOT_ENV) else { - return Self::disabled(); + return Ok(Self::disabled()); }; - let root = PathBuf::from(root); - match start_root_in_root(root.as_path(), metadata) { - Ok(context) => context, - Err(err) => { - warn!("failed to initialize rollout trace bundle: {err:#}"); - Self::disabled() - } - } + start_root_in_root(&PathBuf::from(root), metadata) } /// Starts a root trace in a known directory. @@ -135,16 +124,17 @@ impl ThreadTraceContext { writer: Arc, root_thread_id: AgentThreadId, metadata: ThreadStartedTraceMetadata, - ) -> Self { + ) -> anyhow::Result { let context = EnabledThreadTraceContext { writer, root_thread_id, thread_id: metadata.thread_id.clone(), }; record_thread_started(&context, metadata); - Self { + context.writer.check()?; + Ok(Self { state: ThreadTraceContextState::Enabled(context), - } + }) } /// Returns whether this handle will write trace events. @@ -157,12 +147,12 @@ impl ThreadTraceContext { } /// Starts a new child thread in this context's rollout tree. - pub fn start_child_thread_trace_or_disabled( + pub fn start_child_thread_trace( &self, metadata: ThreadStartedTraceMetadata, - ) -> Self { + ) -> anyhow::Result { match &self.state { - ThreadTraceContextState::Disabled => Self::disabled(), + ThreadTraceContextState::Disabled => Ok(Self::disabled()), ThreadTraceContextState::Enabled(context) => Self::start( Arc::clone(&context.writer), context.root_thread_id.clone(), @@ -172,26 +162,29 @@ impl ThreadTraceContext { } /// Reattaches an evicted child to this rollout without starting a new thread. - pub fn resume_child_thread_trace_or_disabled(&self, thread_id: String) -> Self { + pub fn resume_child_thread_trace(&self, thread_id: String) -> anyhow::Result { let ThreadTraceContextState::Enabled(parent) = &self.state else { - return Self::disabled(); + return Ok(Self::disabled()); }; let context = EnabledThreadTraceContext { writer: Arc::clone(&parent.writer), root_thread_id: parent.root_thread_id.clone(), thread_id: thread_id.clone(), }; - context.append_best_effort(RawTraceEventPayload::ThreadResumed { thread_id }); - Self { + context + .writer + .append(RawTraceEventPayload::ThreadResumed { thread_id })?; + Ok(Self { state: ThreadTraceContextState::Enabled(context), - } + }) } - /// Emits terminal trace events for graceful thread shutdown. + /// Emits terminal trace events when the thread stops. /// /// Spawned child sessions share their root bundle, so only the root - /// thread end closes the rollout. Child thread ends update the child thread - /// execution state without marking the whole bundle complete. + /// thread end closes the rollout. This is not recording finalization: child + /// threads and detached producers may still hold the shared writer. + /// Recording errors stay with the writer and prevent the closing event. pub fn record_ended(&self, status: RolloutStatus) { let ThreadTraceContextState::Enabled(context) = &self.state else { return; @@ -499,15 +492,13 @@ fn start_root_in_root( )?; let writer = Arc::new(writer); - if let Err(err) = writer.append(RawTraceEventPayload::RolloutStarted { + writer.append(RawTraceEventPayload::RolloutStarted { trace_id, root_thread_id: thread_id.clone(), - }) { - warn!("failed to append rollout trace event: {err:#}"); - } + })?; debug!("recording rollout trace at {}", bundle_dir.display()); - Ok(ThreadTraceContext::start(writer, thread_id, metadata)) + ThreadTraceContext::start(writer, thread_id, metadata) } fn record_thread_started( diff --git a/codex-rs/rollout-trace/src/thread_tests.rs b/codex-rs/rollout-trace/src/thread_tests.rs index 2ffb232db4..26ef9bf7d0 100644 --- a/codex-rs/rollout-trace/src/thread_tests.rs +++ b/codex-rs/rollout-trace/src/thread_tests.rs @@ -63,6 +63,39 @@ fn create_in_root_writes_replayable_lifecycle_events() -> anyhow::Result<()> { assert_eq!(replayed.root_thread_id, thread_id.to_string()); assert_eq!(replayed.threads[&thread_id.to_string()].agent_path, "/root"); assert_eq!(replayed.raw_payloads.len(), 1); + assert!(!replayed.recording_complete); + + thread_trace.record_protocol_event(&EventMsg::ShutdownComplete); + drop(thread_trace); + let complete = replay_bundle(&bundle_dir)?; + assert_eq!(complete.status, RolloutStatus::Completed); + assert!(complete.recording_complete); + + // A syntactically valid prefix is still an incomplete export. + let path = bundle_dir.join("trace.jsonl"); + let events = fs::read_to_string(&path)?; + let (prefix, _) = events.trim_end().rsplit_once('\n').unwrap(); + fs::write(&path, prefix)?; + assert!(!replay_bundle(&bundle_dir)?.recording_complete); + fs::write(&path, &events)?; + + // Completion closes the log; even a correctly numbered later event is invalid. + let mut trailing: serde_json::Value = serde_json::from_str(events.lines().last().unwrap())?; + trailing["seq"] = (trailing["seq"].as_u64().unwrap() + 1).into(); + fs::write(&path, format!("{events}{trailing}\n"))?; + assert!(replay_bundle(&bundle_dir).is_err()); + fs::write(&path, &events)?; + + // Protocol breadcrumbs are opaque to replay, but their payloads are still required. + let opaque = complete + .raw_payloads + .values() + .find(|payload| payload.kind == RawPayloadKind::ProtocolEvent) + .unwrap(); + fs::write(bundle_dir.join(&opaque.path), "{\"truncated\":")?; + assert!(replay_bundle(&bundle_dir).is_err()); + fs::remove_file(bundle_dir.join(&opaque.path))?; + assert!(replay_bundle(&bundle_dir).is_err()); Ok(()) } @@ -77,7 +110,7 @@ fn spawned_thread_start_appends_to_root_bundle() -> anyhow::Result<()> { minimal_metadata(root_thread_id), )?; - let child_trace = root_trace.start_child_thread_trace_or_disabled(ThreadStartedTraceMetadata { + let child_trace = root_trace.start_child_thread_trace(ThreadStartedTraceMetadata { thread_id: child_thread_id.to_string(), agent_path: "/root/repo_file_counter".to_string(), task_name: Some("repo_file_counter".to_string()), @@ -98,7 +131,7 @@ fn spawned_thread_start_appends_to_root_bundle() -> anyhow::Result<()> { provider_name: "test-provider".to_string(), approval_policy: "never".to_string(), sandbox_policy: format!("{:?}", SandboxPolicy::DangerFullAccess), - }); + })?; child_trace.record_ended(RolloutStatus::Completed); let bundle_dir = single_bundle_dir(temp.path())?; let replayed = replay_bundle(&bundle_dir)?; @@ -118,6 +151,42 @@ fn spawned_thread_start_appends_to_root_bundle() -> anyhow::Result<()> { ); assert_eq!(replayed.raw_payloads.len(), 2); + root_trace.record_ended(RolloutStatus::Completed); + drop(root_trace); + // Root shutdown does not close the recording while a child still owns it. + assert!(!replay_bundle(&bundle_dir)?.recording_complete); + drop(child_trace); + assert!(replay_bundle(&bundle_dir)?.recording_complete); + + Ok(()) +} + +#[test] +fn resumed_child_rejects_a_failed_shared_writer() -> anyhow::Result<()> { + let temp = TempDir::new()?; + let root = ThreadTraceContext::start_root_in_root_for_test( + temp.path(), + minimal_metadata(ThreadId::new()), + )?; + let child_id = ThreadId::new(); + let child = root.start_child_thread_trace(minimal_metadata(child_id))?; + child.record_ended(RolloutStatus::Completed); + drop(child); + + // A parent payload write fails while the child is evicted. + let bundle = single_bundle_dir(temp.path())?; + let payloads = bundle.join("payloads"); + let backup = temp.path().join("saved-payloads"); + fs::rename(&payloads, &backup)?; + root.record_protocol_event(&EventMsg::ShutdownComplete); + fs::rename(backup, payloads)?; + + assert!( + root.resume_child_thread_trace(child_id.to_string()) + .is_err() + ); + drop(root); + assert!(!replay_bundle(&bundle)?.recording_complete); Ok(()) } diff --git a/codex-rs/rollout-trace/src/writer.rs b/codex-rs/rollout-trace/src/writer.rs index fe09cf5b03..f3cae91c78 100644 --- a/codex-rs/rollout-trace/src/writer.rs +++ b/codex-rs/rollout-trace/src/writer.rs @@ -1,4 +1,7 @@ //! Hot-path trace bundle writer. +//! +//! All producers share one writer. Write failures prevent further recording and +//! suppress the final closing event. use std::fs::File; use std::fs::OpenOptions; @@ -14,6 +17,7 @@ use std::time::UNIX_EPOCH; use anyhow::Context; use anyhow::Result; +use anyhow::bail; use serde::Serialize; use crate::bundle::MANIFEST_FILE_NAME; @@ -44,6 +48,8 @@ struct TraceWriterInner { event_log: BufWriter, next_seq: u64, next_payload_ordinal: u64, + // Preserve the first failure even when a producer cannot return it to its caller. + error: Option, } impl TraceWriter { @@ -62,14 +68,13 @@ impl TraceWriter { let started_at_unix_ms = unix_time_ms(); let manifest = TraceBundleManifest::new(trace_id, rollout_id, root_thread_id, started_at_unix_ms); - write_json_file(&bundle_dir.join(MANIFEST_FILE_NAME), &manifest)?; - let event_log_path = bundle_dir.join(RAW_EVENT_LOG_FILE_NAME); let event_log = OpenOptions::new() - .create(true) - .append(true) + .create_new(true) + .write(true) .open(&event_log_path) .with_context(|| format!("open trace event log {}", event_log_path.display()))?; + write_json_file(&bundle_dir.join(MANIFEST_FILE_NAME), &manifest)?; Ok(Self { inner: Mutex::new(TraceWriterInner { @@ -78,6 +83,7 @@ impl TraceWriter { event_log: BufWriter::new(event_log), next_seq: 1, next_payload_ordinal: 1, + error: None, }), }) } @@ -89,6 +95,7 @@ impl TraceWriter { value: &impl Serialize, ) -> Result { let mut inner = self.lock_inner(); + inner.check()?; let ordinal = inner.next_payload_ordinal; inner.next_payload_ordinal += 1; let raw_payload_id = format!("raw_payload:{ordinal}"); @@ -97,7 +104,8 @@ impl TraceWriter { // Payload files are created before the event that references them. A // replay interrupted after an event is appended should never point at a // payload file that the writer planned but had not written yet. - write_json_file(&absolute_path, value)?; + let result = write_json_file(&absolute_path, value); + inner.record_result(result)?; Ok(RawPayloadRef { raw_payload_id, kind, @@ -117,6 +125,7 @@ impl TraceWriter { payload: RawTraceEventPayload, ) -> Result { let mut inner = self.lock_inner(); + inner.check()?; let event = RawTraceEvent { schema_version: RAW_TRACE_EVENT_SCHEMA_VERSION, seq: inner.next_seq, @@ -126,13 +135,22 @@ impl TraceWriter { codex_turn_id: context.codex_turn_id, payload, }; + let result = (|| { + serde_json::to_writer(&mut inner.event_log, &event)?; + inner.event_log.write_all(b"\n")?; + inner.event_log.flush()?; + Ok(()) + })(); + inner.record_result(result)?; inner.next_seq += 1; - serde_json::to_writer(&mut inner.event_log, &event)?; - inner.event_log.write_all(b"\n")?; - inner.event_log.flush()?; Ok(event) } + /// Reports any earlier recording failure, including failures in detached producers. + pub(crate) fn check(&self) -> Result<()> { + self.lock_inner().check() + } + fn lock_inner(&self) -> MutexGuard<'_, TraceWriterInner> { // Preserve the event log after a panic in tracing code. Dropping the // writer would lose subsequent diagnostic events in exactly the session @@ -141,6 +159,34 @@ impl TraceWriter { } } +impl TraceWriterInner { + fn check(&self) -> Result<()> { + if let Some(error) = &self.error { + bail!("rollout recording failed: {error}"); + } + Ok(()) + } + + fn record_result(&mut self, result: Result) -> Result { + if let Err(error) = &result { + let error = format!("{error:#}"); + tracing::error!("rollout recording failed: {error}"); + self.error = Some(error); + } + result + } +} + +impl Drop for TraceWriter { + fn drop(&mut self) { + // The last producer releases the writer, even for interrupted turns. + // This marks recorder closure, not successful task completion. + if !self.inner.is_poisoned() && !std::thread::panicking() { + let _ = self.append(RawTraceEventPayload::RecordingCompleted); + } + } +} + fn write_json_file(path: &Path, value: &impl Serialize) -> Result<()> { let file = File::create(path).with_context(|| format!("create {}", path.display()))?; let mut writer = BufWriter::new(file); @@ -266,4 +312,54 @@ mod tests { Ok(()) } + + #[test] + fn payload_failure_prevents_finalization_even_after_storage_recovers() -> anyhow::Result<()> { + let temp = TempDir::new()?; + let writer = + TraceWriter::create(temp.path(), "trace".into(), "rollout".into(), "root".into())?; + std::fs::remove_dir(temp.path().join("payloads"))?; + assert!( + writer + .write_json_payload(RawPayloadKind::ProtocolEvent, &json!({})) + .is_err() + ); + std::fs::create_dir(temp.path().join("payloads"))?; + assert!(writer.check().is_err()); + assert!( + writer + .write_json_payload(RawPayloadKind::ProtocolEvent, &json!({})) + .is_err() + ); + drop(writer); + assert_eq!( + std::fs::read_to_string(temp.path().join("trace.jsonl"))?, + "" + ); + Ok(()) + } + + #[test] + fn event_write_failure_prevents_finalization() -> anyhow::Result<()> { + let temp = TempDir::new()?; + let writer = + TraceWriter::create(temp.path(), "trace".into(), "rollout".into(), "root".into())?; + // A read-only file makes the real buffered write fail on every platform. + writer.lock_inner().event_log = + std::io::BufWriter::new(std::fs::File::open(temp.path().join("trace.jsonl"))?); + assert!( + writer + .append(RawTraceEventPayload::RolloutEnded { + status: RolloutStatus::Completed + }) + .is_err() + ); + assert!(writer.check().is_err()); + drop(writer); + assert_eq!( + std::fs::read_to_string(temp.path().join("trace.jsonl"))?, + "" + ); + Ok(()) + } }