From 39b42aaacdbb55b86528b4294443696985c348ee Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Tue, 10 Nov 2026 03:53:20 +0000 Subject: [PATCH] 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 --- .../app-server/tests/suite/v2/thread_fork.rs | 10 +++++++ codex-rs/core/src/context_manager/history.rs | 8 ++++- .../core/src/context_manager/normalize.rs | 30 +++++++++---------- codex-rs/core/src/session/mod.rs | 1 + 4 files changed, 33 insertions(+), 16 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index 3d05a626b4..cb329b7a12 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/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::()?["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| { diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index 3e8221ba33..2ec53d71e9 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/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>, @@ -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) { 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); diff --git a/codex-rs/core/src/context_manager/normalize.rs b/codex-rs/core/src/context_manager/normalize.rs index 387bc8f75a..20fd98a6a6 100644 --- a/codex-rs/core/src/context_manager/normalize.rs +++ b/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) { +pub(crate) fn ensure_call_outputs_present( + items: &mut Vec, + 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) 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) 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}" )); diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 1f116885b0..1809f27bba 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/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);