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

@@ -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);

codex-rs/core/src/session/handlers.rs

@@ -428,10 +428,10 @@ pub async fn shutdown(sess: &Arc<Session>, 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");
 }

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);

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<dyn codex_thread_store::ThreadStore> = 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]

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
 

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<JsonValue>,
+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,

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<JsonValue>,
+    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<JsonValue> {
+    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<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        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!(

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<i64>,
     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<AgentThreadId, AgentThread>,
     pub codex_turns: BTreeMap<CodexTurnId, CodexTurn>,
@@ -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(),

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 { .. }

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<Path>) -> Result<RolloutTrace> {
     let bundle_dir = bundle_dir.as_ref();
     let manifest: TraceBundleManifest =
@@ -85,21 +88,43 @@ pub fn replay_bundle(bundle_dir: impl AsRef<Path>) -> Result<RolloutTrace> {
     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,

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<Self> {
         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<TraceWriter>,
         root_thread_id: AgentThreadId,
         metadata: ThreadStartedTraceMetadata,
-    ) -> Self {
+    ) -> anyhow::Result<Self> {
         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<Self> {
         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<Self> {
         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(

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(())
 }
 

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<File>,
     next_seq: u64,
     next_payload_ordinal: u64,
+    // Preserve the first failure even when a producer cannot return it to its caller.
+    error: Option<String>,
 }
 
 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<RawPayloadRef> {
         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<RawTraceEvent> {
         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<T>(&mut self, result: Result<T>) -> Result<T> {
+        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(())
+    }
 }