Preserve failed write_stdin calls in rollout trace replay (#57193)

## Why

Invalid model arguments to `write_stdin` represent a tool failure, but parsing them could fail rollout trace replay.

## What changed

Skip terminal operation creation when dispatched `write_stdin` arguments cannot be parsed or lack a usable `session_id`. Preserve the generic tool call and its error, including the original arguments, so replay can continue.

## Testing

Add a regression test covering malformed JSON and missing or null `session_id` values. Verify that replay retains the failed call's arguments and error without creating terminal operations or sessions.

GitOrigin-RevId: 967e975a88072c169eb3e391d5c62e807d608fbf

codex-rs/rollout-trace/src/reducer/tool/terminal.rs

@@ -455,10 +455,14 @@ fn parse_dispatch_terminal_request(value: JsonValue) -> Result<Option<ParsedTerm
         .payload
         .arguments
         .context("write_stdin dispatch payload omitted function arguments")?;
-    let args: DispatchedWriteStdinArgs = serde_json::from_str(&arguments)
-        .context("parse write_stdin dispatch function arguments")?;
-    let process_id = process_id_from_json(&args.session_id)
-        .context("write_stdin dispatch payload omitted session_id")?;
+    // Invalid model arguments are a tool failure, not a corrupt trace. Keep the
+    // generic call and its error without inventing a terminal operation.
+    let Ok(args) = serde_json::from_str::<DispatchedWriteStdinArgs>(&arguments) else {
+        return Ok(None);
+    };
+    let Some(process_id) = process_id_from_json(&args.session_id) else {
+        return Ok(None);
+    };
 
     Ok(Some(ParsedTerminalRequest {
         process_id: Some(process_id),

codex-rs/rollout-trace/src/reducer/tool/terminal_tests.rs

@@ -1,7 +1,12 @@
+use std::sync::Arc;
+
 use pretty_assertions::assert_eq;
 use serde_json::json;
 use tempfile::TempDir;
 
+use crate::ToolDispatchInvocation;
+use crate::ToolDispatchPayload;
+use crate::ToolDispatchRequester;
 use crate::model::ExecutionStatus;
 use crate::model::ExecutionWindow;
 use crate::model::TerminalModelObservation;
@@ -15,6 +20,7 @@ use crate::model::ToolCallKind;
 use crate::model::ToolCallSummary;
 use crate::payload::RawPayloadKind;
 use crate::raw_event::RawTraceEventPayload;
+use crate::reducer::test_support::ROOT_THREAD_ID;
 use crate::reducer::test_support::create_started_writer;
 use crate::reducer::test_support::generic_summary;
 use crate::reducer::test_support::message;
@@ -24,6 +30,7 @@ use crate::reducer::test_support::start_turn_for_thread;
 use crate::reducer::test_support::trace_context;
 use crate::reducer::test_support::trace_context_for_thread;
 use crate::replay_bundle;
+use crate::tool_dispatch::ToolDispatchTraceContext;
 use crate::writer::TraceWriter;
 
 #[test]
@@ -491,6 +498,58 @@ fn dispatch_write_stdin_payload_reduces_to_terminal_operation() -> anyhow::Resul
     Ok(())
 }
 
+#[test]
+fn invalid_write_stdin_arguments_remain_failed_tool_calls() -> anyhow::Result<()> {
+    for arguments in [r#"{"session_id": "#, "{}", r#"{"session_id": null}"#] {
+        let temp = TempDir::new()?;
+        let writer = Arc::new(create_started_writer(&temp)?);
+        start_turn(&writer, "turn-1")?;
+        let dispatch = ToolDispatchTraceContext::start(
+            Arc::clone(&writer),
+            ToolDispatchInvocation {
+                thread_id: ROOT_THREAD_ID.to_string(),
+                codex_turn_id: "turn-1".to_string(),
+                tool_call_id: "call-stdin".to_string(),
+                tool_name: "write_stdin".to_string(),
+                tool_namespace: None,
+                requester: ToolDispatchRequester::Model {
+                    model_visible_call_id: "call-stdin".to_string(),
+                },
+                payload: ToolDispatchPayload::Function {
+                    arguments: arguments.to_string(),
+                },
+            },
+        );
+        dispatch.record_failed("invalid arguments");
+
+        let rollout = replay_bundle(temp.path())?;
+        let call = &rollout.tool_calls["call-stdin"];
+        assert_eq!(call.execution.status, ExecutionStatus::Failed);
+        assert_eq!(call.terminal_operation_id, None);
+        assert!(rollout.terminal_operations.is_empty());
+        assert!(rollout.terminal_sessions.is_empty());
+        // The failed call still retains its original arguments and error.
+        for (id, expected) in [
+            (
+                call.raw_invocation_payload_id.as_ref().unwrap(),
+                json!({"tool_name": "write_stdin", "tool_namespace": null,
+                    "payload": {"type": "function", "arguments": arguments}}),
+            ),
+            (
+                call.raw_result_payload_id.as_ref().unwrap(),
+                json!({"type": "error", "error": "invalid arguments"}),
+            ),
+        ] {
+            let payload = std::fs::read(temp.path().join(&rollout.raw_payloads[id].path))?;
+            assert_eq!(
+                serde_json::from_slice::<serde_json::Value>(&payload)?,
+                expected
+            );
+        }
+    }
+    Ok(())
+}
+
 #[test]
 fn code_mode_write_stdin_result_projects_structured_exec_fields() -> anyhow::Result<()> {
     let temp = TempDir::new()?;