Handle unfinished tool calls in restored history (#60519)

## Why

Forked and recovered sessions can restore a tool call before its output has been recorded. Missing outputs for these calls should be treated as interrupted work rather than an error.

## What changed

Track the restored history's sequence cutoff so custom tool and local shell calls in that history can receive synthetic `aborted` outputs without reporting a missing-output error. Keep the existing checks for newly recorded calls and clamp the cutoff when history is replaced.

## Testing

Extend the active-turn fork test with a pending `apply_patch` call and verify that the fork's model input includes its `aborted` output.

GitOrigin-RevId: d989fa0991db72f0f8fa8a6d5de403fb307c40c7

codex-rs/app-server/tests/suite/v2/thread_fork.rs

@@ -1958,6 +1958,11 @@ async fn thread_fork_freezes_active_paginated_turn_as_interrupted(
         &completed_user_item("before-fork", /*completed_at_ms*/ 1),
     )
     .await?;
+    let pending_call = serde_json::from_value(json!({
+        "type": "response_item",
+        "payload": {"type": "custom_tool_call", "call_id": "pending", "name": "apply_patch", "input": "patch"}
+    }))?;
+    append_rollout_item_to_path(&source_path, &pending_call).await?;
     let mut mcp = TestAppServer::builder()
         .with_codex_home(codex_home.path())
         .without_auto_env()
@@ -2093,6 +2098,11 @@ async fn thread_fork_freezes_active_paginated_turn_as_interrupted(
         .body_json::<Value>()?["input"]
         .clone();
     let serialized_input = serde_json::to_string(&input)?;
+    assert!(input.as_array().expect("model input").iter().any(|item| {
+        item["type"] == "custom_tool_call_output"
+            && item["call_id"] == "pending"
+            && item["output"] == "aborted"
+    }));
     assert!(serialized_input.contains("before-fork model input"));
     assert!(!serialized_input.contains("after-fork model input"));
     assert!(input.as_array().is_some_and(|items| {

codex-rs/core/src/context_manager/history.rs

@@ -105,6 +105,8 @@ fn next_guardian_review_context_revision() -> u64 {
 pub(crate) struct ContextManager {
     item_id_generator: ItemIdGenerator,
     sampling: super::sampling::SamplingProgress,
+    /// Inclusive sequence cutoff for calls that may still be running in the fork source.
+    pub(crate) restored_through_sequence: u64,
     /// The oldest items are at the beginning of the vector. Snapshots share the vector until a
     /// caller needs to mutate it, avoiding deep copies for read-only history consumers.
     items: Arc<Vec<ResponseItemEnvelope>>,
@@ -319,6 +321,7 @@ impl ContextManager {
         Self {
             item_id_generator,
             sampling: Default::default(),
+            restored_through_sequence: 0,
             items: Arc::new(Vec::new()),
             review_history: None,
             review_budgeting: ContextBudgeting::Fixed,
@@ -876,6 +879,9 @@ impl ContextManager {
 
     pub(crate) fn replace_annotated(&mut self, mut items: Vec<ResponseItemEnvelope>) {
         self.sampling = super::sampling::SamplingProgress::restore(&mut items);
+        self.restored_through_sequence = self
+            .restored_through_sequence
+            .min(self.sampling.last_sequence);
         self.retained_context = Arc::default();
         self.user_message_revision = self.user_message_revision.saturating_add(1);
         if let Some(review_history) = &mut self.review_history {
@@ -1174,7 +1180,7 @@ impl ContextManager {
         let items = Arc::make_mut(&mut self.items);
 
         // all function/tool calls must have a corresponding output
-        normalize::ensure_call_outputs_present(items);
+        normalize::ensure_call_outputs_present(items, self.restored_through_sequence);
 
         // Paired outputs must have a corresponding call; named external outputs stand alone.
         normalize::remove_orphan_outputs(items);

codex-rs/core/src/context_manager/normalize.rs

@@ -18,7 +18,19 @@ use tracing::info;
 // Changing this value would change model-visible IDs and invalidate prompt caches.
 const SYNTHETIC_OUTPUT_ID_NAMESPACE: Uuid = Uuid::from_u128(0x90d38d3e_6a5b_4d52_bfe2_2f1e634bfac4);
 
-pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItemEnvelope>) {
+pub(crate) fn ensure_call_outputs_present(
+    items: &mut Vec<ResponseItemEnvelope>,
+    restored_through_sequence: u64,
+) {
+    // Forks and recovered sessions can inherit a call before its output was written.
+    let may_be_unfinished = |envelope: &ResponseItemEnvelope| {
+        envelope.metadata.as_ref().is_some_and(|metadata| {
+            metadata.inherited_user_message
+                || metadata
+                    .history_sequence
+                    .is_some_and(|sequence| sequence <= restored_through_sequence)
+        })
+    };
     let mut function_output_ids = HashSet::new();
     let mut tool_search_output_ids = HashSet::new();
     let mut custom_tool_output_ids = HashSet::new();
@@ -87,13 +99,7 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItemEnvelope>)
             ResponseItem::CustomToolCall { id, call_id, .. }
                 if !custom_tool_output_ids.contains(call_id.as_str()) =>
             {
-                // Preserved-prefix forks mark copied calls as inherited. A call still
-                // running in the parent has no output in the child's snapshot.
-                if !envelope
-                    .metadata
-                    .as_ref()
-                    .is_some_and(|metadata| metadata.inherited_user_message)
-                {
+                if !may_be_unfinished(envelope) {
                     error_or_panic(format!(
                         "Custom tool call output is missing for call id: {call_id}"
                     ));
@@ -115,13 +121,7 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItemEnvelope>)
                 call_id: Some(call_id),
                 ..
             } if !function_output_ids.contains(call_id.as_str()) => {
-                // Preserved-prefix forks mark copied calls as inherited. A call still
-                // running in the parent has no output in the child's snapshot.
-                if !envelope
-                    .metadata
-                    .as_ref()
-                    .is_some_and(|metadata| metadata.inherited_user_message)
-                {
+                if !may_be_unfinished(envelope) {
                     error_or_panic(format!(
                         "Local shell call output is missing for call id: {call_id}"
                     ));

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

@@ -2037,6 +2037,7 @@ impl Session {
                 reference_context_item,
                 HistoryReplacement::Reset,
             );
+            state.history.restored_through_sequence = state.history.input_sequence();
             state
                 .history
                 .restore_rejected_audio_file_ids(rejected_audio_file_ids);