codex: capture only committed live events

Live stream translation writes traffic artifacts while initial Anthropic
framing is still buffered. Empty completion retries discard those bytes, but
the capture previously presented them as events delivered to the client.

Translate the buffered prefix without a live capture sink, then parse and
record its Anthropic SSE events only when the prefix is committed downstream.
Later stream events continue to use direct capture. Extend the retry smoke test
to prove discarded end-turn framing stays out of downstream artifacts.

Client-visible streaming and retry behavior is unchanged.

Author: raine

src/providers/codex/mod.rs

@@ -20,6 +20,7 @@ use std::sync::Arc;
 
 use crate::anthropic::error::json_error;
 use crate::anthropic::schema::{CountTokensResponse, MessagesRequest};
+use crate::anthropic::sse::parse_sse_events;
 use crate::config;
 use crate::logging::create_logger;
 use crate::monitor::usage_from_anthropic_sse;
@@ -563,7 +564,7 @@ async fn live_stream_response_once(
             generation_started = true;
         }
         append_upstream_sse_payload(&mut upstream_sse_body, &payload);
-        let (chunk, terminal) = match translate_live_stream_payload(&mut translator, &payload, &ctx)
+        let (chunk, terminal) = match translate_live_stream_payload(&mut translator, &payload, None)
         {
             Ok(result) => result,
             Err(message) => {
@@ -619,6 +620,7 @@ async fn live_stream_response_once(
             };
         }
         if translator.has_semantic_output() && !pending_chunk.is_empty() {
+            record_live_stream_downstream_capture(&ctx, &pending_chunk);
             record_live_stream_progress(&ctx, &pending_chunk);
             if terminal {
                 update_continuation_from_upstream(
@@ -652,6 +654,7 @@ async fn live_stream_response_once(
             if pending_chunk.is_empty() {
                 return LiveStreamStart::Response(empty_live_stream_response());
             }
+            record_live_stream_downstream_capture(&ctx, &pending_chunk);
             record_live_stream_progress(&ctx, &pending_chunk);
             return LiveStreamStart::Response(single_live_stream_response(pending_chunk));
         }
@@ -688,13 +691,31 @@ fn codex_generation_event(payload: &serde_json::Value) -> bool {
 fn translate_live_stream_payload(
     translator: &mut LiveStreamTranslator,
     payload: &serde_json::Value,
-    ctx: &RequestContext,
+    traffic: Option<&crate::traffic::TrafficCapture>,
 ) -> Result<(Vec<u8>, bool), String> {
-    let chunk = translator.accept(payload, ctx.traffic.as_deref())?;
+    let chunk = translator.accept(payload, traffic)?;
     let terminal = is_codex_terminal_event(payload) || translator.is_finished();
     Ok((chunk, terminal))
 }
 
+fn record_live_stream_downstream_capture(ctx: &RequestContext, chunk: &[u8]) {
+    let Some(traffic) = ctx.traffic.as_ref() else {
+        return;
+    };
+    for event in parse_sse_events(chunk) {
+        let Ok(data) = serde_json::from_str::<serde_json::Value>(&event.data) else {
+            continue;
+        };
+        traffic.write_json_event(
+            "050-downstream-event",
+            &serde_json::json!({
+                "event": event.event.as_deref().unwrap_or("message"),
+                "data": data,
+            }),
+        );
+    }
+}
+
 fn record_live_stream_progress(ctx: &RequestContext, chunk: &[u8]) {
     if let Some(monitor) = ctx.monitor.as_ref() {
         let (input_tokens, output_tokens) = usage_from_anthropic_sse(chunk);
@@ -744,28 +765,31 @@ fn remaining_live_stream_response(
             match item {
                 Ok(payload) => {
                     append_upstream_sse_payload(&mut upstream_sse_body, &payload);
-                    let (chunk, terminal) =
-                        match translate_live_stream_payload(&mut translator, &payload, &ctx) {
-                            Ok(result) => result,
-                            Err(message) => {
-                                abort_request_state(
-                                    ctx.session_id.as_deref(),
-                                    turn_id,
-                                    compact_boundary,
-                                    &request_body,
-                                );
-                                let chunk = translator.error_chunk(
-                                    &message,
-                                    "api_error",
-                                    ctx.traffic.as_deref(),
-                                );
-                                if !chunk.is_empty() {
-                                    record_live_stream_progress(&ctx, &chunk);
-                                    let _ = tx.send(Ok(Bytes::from(chunk))).await;
-                                }
-                                return;
+                    let (chunk, terminal) = match translate_live_stream_payload(
+                        &mut translator,
+                        &payload,
+                        ctx.traffic.as_deref(),
+                    ) {
+                        Ok(result) => result,
+                        Err(message) => {
+                            abort_request_state(
+                                ctx.session_id.as_deref(),
+                                turn_id,
+                                compact_boundary,
+                                &request_body,
+                            );
+                            let chunk = translator.error_chunk(
+                                &message,
+                                "api_error",
+                                ctx.traffic.as_deref(),
+                            );
+                            if !chunk.is_empty() {
+                                record_live_stream_progress(&ctx, &chunk);
+                                let _ = tx.send(Ok(Bytes::from(chunk))).await;
                             }
-                        };
+                            return;
+                        }
+                    };
                     if !chunk.is_empty() {
                         record_live_stream_progress(&ctx, &chunk);
                         if tx.send(Ok(Bytes::from(chunk))).await.is_err() {

tests/smoke_cutover.rs

@@ -1580,13 +1580,16 @@ async fn smoke_codex_websocket_stream_retries_empty_close_with_full_context() {
 async fn smoke_codex_websocket_stream_retries_terminal_only_completion_with_full_context() {
     let _guard = env_lock();
     let config = TempDir::new().unwrap();
+    let state = TempDir::new().unwrap();
     write_auth(config.path(), "codex");
     clear_codex_websocket_pool_for_tests();
     clear_all_continuations_for_tests();
 
     let captured = Arc::new(Mutex::new(Vec::new()));
     let upstream = spawn_websocket_empty_completion_then_retry_upstream(captured.clone()).await;
 
+    let _traffic_env = EnvGuard::set("CCP_TRAFFIC_LOG", "1");
+    let _state_env = EnvGuard::set("XDG_STATE_HOME", state.path());
     let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path());
     let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream);
     let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "websocket");
@@ -1625,6 +1628,22 @@ async fn smoke_codex_websocket_stream_retries_terminal_only_completion_with_full
         String::from_utf8_lossy(&second_body)
     );
 
+    let downstream_end_turns = traffic_files(state.path())
+        .into_iter()
+        .filter(|path| {
+            path.file_name()
+                .and_then(|name| name.to_str())
+                .is_some_and(|name| name.ends_with("050-downstream-event.json"))
+        })
+        .filter_map(|path| std::fs::read(path).ok())
+        .filter_map(|bytes| serde_json::from_slice::<Value>(&bytes).ok())
+        .filter(|event| event["data"]["delta"]["stop_reason"] == "end_turn")
+        .count();
+    assert_eq!(
+        downstream_end_turns, 2,
+        "discarded empty attempts must not be captured as downstream events"
+    );
+
     let guard = captured.lock().unwrap();
     assert_eq!(guard.len(), 3, "expected full-context retry request");
     assert!(guard[0].get("previous_response_id").is_none());