diff --git a/codex-rs/codex-api/src/endpoint/responses.rs b/codex-rs/codex-api/src/endpoint/responses.rs index 57a44d2e2755..686b24e9f995 100644 --- a/codex-rs/codex-api/src/endpoint/responses.rs +++ b/codex-rs/codex-api/src/endpoint/responses.rs @@ -81,9 +81,9 @@ impl ResponsesClient { let mut body = serde_json::to_value(&request) .map_err(|e| ApiError::Stream(format!("failed to encode responses request: {e}")))?; - if request.store && self.session.provider().is_azure_responses_endpoint() { - attach_item_ids(&mut body, &request.input); - } + // Replayed model-generated items must preserve their original ids so + // the Responses API can validate reasoning/call lineage across retries. + attach_item_ids(&mut body, &request.input); let mut headers = extra_headers; if let Some(ref conv_id) = conversation_id { diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index dce3d2f3b282..4f5f31b0fa21 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -6487,7 +6487,7 @@ async fn run_sampling_request( let base_instructions = sess.get_base_instructions().await; - let prompt = build_prompt( + let mut prompt = build_prompt( input, router.as_ref(), turn_context.as_ref(), @@ -6598,6 +6598,17 @@ async fn run_sampling_request( .await; } tokio::time::sleep(delay).await; + // The failed attempt may have already recorded partial reasoning, + // calls, or tool outputs. Rebuild the prompt from live history so + // retries replay the same causal chain back to the model. + prompt = build_prompt( + sess.clone_history() + .await + .for_prompt(&turn_context.model_info.input_modalities), + router.as_ref(), + turn_context.as_ref(), + sess.get_base_instructions().await, + ); } else { return Err(err); } @@ -7431,6 +7442,10 @@ async fn try_run_sampling_request( needs_follow_up |= output_result.needs_follow_up; } ResponseEvent::OutputItemAdded(item) => { + if matches!(item, ResponseItem::Reasoning { .. }) { + sess.record_into_history(std::slice::from_ref(&item), turn_context.as_ref()) + .await; + } if let Some(turn_item) = handle_non_tool_response_item( sess.as_ref(), turn_context.as_ref(), diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index 4d7f4c558a5b..32d1f7cb199e 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -101,6 +101,17 @@ impl ContextManager { } let processed = self.process_item(item_ref, policy); + if let Some(reasoning_id) = reasoning_item_id(&processed) + && let Some(existing) = self + .items + .iter_mut() + .rev() + .find(|existing| reasoning_item_id(existing) == Some(reasoning_id)) + { + *existing = processed; + continue; + } + self.items.push(processed); } } @@ -428,6 +439,13 @@ fn is_api_message(message: &ResponseItem) -> bool { } } +fn reasoning_item_id(item: &ResponseItem) -> Option<&str> { + match item { + ResponseItem::Reasoning { id, .. } if !id.is_empty() => Some(id.as_str()), + _ => None, + } +} + fn estimate_reasoning_length(encoded_len: usize) -> usize { encoded_len .saturating_mul(3) diff --git a/codex-rs/core/src/context_manager/history_tests.rs b/codex-rs/core/src/context_manager/history_tests.rs index 066272748f47..6ae4b8b1b7af 100644 --- a/codex-rs/core/src/context_manager/history_tests.rs +++ b/codex-rs/core/src/context_manager/history_tests.rs @@ -101,6 +101,21 @@ fn reasoning_with_encrypted_content(len: usize) -> ResponseItem { } } +fn reasoning_with_id(id: &str, summary: &str, raw_content: Option<&str>) -> ResponseItem { + ResponseItem::Reasoning { + id: id.to_string(), + summary: vec![ReasoningItemReasoningSummary::SummaryText { + text: summary.to_string(), + }], + content: raw_content.map(|text| { + vec![ReasoningItemContent::ReasoningText { + text: text.to_string(), + }] + }), + encrypted_content: raw_content.map(|text| BASE64_STANDARD.encode(format!("padding{text}"))), + } +} + fn truncate_exec_output(content: &str) -> String { truncate::truncate_text(content, TruncationPolicy::Tokens(EXEC_FORMAT_MAX_TOKENS)) } @@ -167,6 +182,19 @@ fn filters_non_api_messages() { ); } +#[test] +fn record_items_replaces_reasoning_item_with_same_non_empty_id() { + let mut history = ContextManager::new(); + let policy = TruncationPolicy::Tokens(10_000); + let partial = reasoning_with_id("reason-1", "partial", None); + let completed = reasoning_with_id("reason-1", "complete", Some("raw detail")); + + history.record_items([&partial], policy); + history.record_items([&completed], policy); + + assert_eq!(history.raw_items(), &[completed]); +} + #[test] fn non_last_reasoning_tokens_return_zero_when_no_user_messages() { let history = create_history_with_items(vec![reasoning_with_encrypted_content(800)]); diff --git a/codex-rs/core/tests/suite/abort_tasks.rs b/codex-rs/core/tests/suite/abort_tasks.rs index af3c70b74e32..88c9ae8107b0 100644 --- a/codex-rs/core/tests/suite/abort_tasks.rs +++ b/codex-rs/core/tests/suite/abort_tasks.rs @@ -7,6 +7,8 @@ use codex_protocol::protocol::Op; use codex_protocol::user_input::UserInput; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_function_call; +use core_test_support::responses::ev_reasoning_item_added; +use core_test_support::responses::ev_reasoning_summary_text_delta; use core_test_support::responses::ev_response_created; use core_test_support::responses::mount_sse_once; use core_test_support::responses::mount_sse_sequence; @@ -72,6 +74,7 @@ async fn interrupt_long_running_tool_emits_turn_aborted() { async fn interrupt_tool_records_history_entries() { let command = "sleep 60"; let call_id = "call-history"; + let reasoning_id = "reasoning-history"; let args = json!({ "command": command, @@ -80,6 +83,8 @@ async fn interrupt_tool_records_history_entries() { .to_string(); let first_body = sse(vec![ ev_response_created("resp-history"), + ev_reasoning_item_added(reasoning_id, &[""]), + ev_reasoning_summary_text_delta("thinking"), ev_function_call(call_id, "shell_command", &args), ev_completed("resp-history"), ]); @@ -136,6 +141,20 @@ async fn interrupt_tool_records_history_entries() { requests.len() ); + let follow_up_request = &requests[1]; + let follow_up_reasoning_items = follow_up_request.inputs_of_type("reasoning"); + assert_eq!( + follow_up_reasoning_items.len(), + 1, + "expected follow-up request to replay the interrupted reasoning item" + ); + assert_eq!( + follow_up_reasoning_items[0] + .get("id") + .and_then(serde_json::Value::as_str), + Some(reasoning_id) + ); + assert!( response_mock.saw_function_call(call_id), "function call not recorded in responses payload" @@ -162,6 +181,33 @@ async fn interrupt_tool_records_history_entries() { secs >= 0.1, "expected at least one tenth of a second of elapsed time, got {secs}" ); + + let follow_up_input = follow_up_request.input(); + let reasoning_index = follow_up_input + .iter() + .position(|item| { + item.get("type").and_then(serde_json::Value::as_str) == Some("reasoning") + && item.get("id").and_then(serde_json::Value::as_str) == Some(reasoning_id) + }) + .expect("missing reasoning item in follow-up request"); + let function_call_index = follow_up_input + .iter() + .position(|item| { + item.get("type").and_then(serde_json::Value::as_str) == Some("function_call") + && item.get("call_id").and_then(serde_json::Value::as_str) == Some(call_id) + }) + .expect("missing function_call in follow-up request"); + let function_output_index = follow_up_input + .iter() + .position(|item| { + item.get("type").and_then(serde_json::Value::as_str) == Some("function_call_output") + && item.get("call_id").and_then(serde_json::Value::as_str) == Some(call_id) + }) + .expect("missing function_call_output in follow-up request"); + assert!( + reasoning_index < function_call_index && function_call_index < function_output_index, + "expected reasoning -> function_call -> function_call_output ordering in follow-up replay" + ); } /// After an interrupt we persist a model-visible `` marker in the conversation diff --git a/codex-rs/core/tests/suite/stream_no_completed.rs b/codex-rs/core/tests/suite/stream_no_completed.rs index 5d1b21481111..60af3fb2b9ab 100644 --- a/codex-rs/core/tests/suite/stream_no_completed.rs +++ b/codex-rs/core/tests/suite/stream_no_completed.rs @@ -9,12 +9,18 @@ use codex_protocol::user_input::UserInput; use codex_utils_cargo_bin::find_resource; use core_test_support::load_sse_fixture; use core_test_support::responses; +use core_test_support::responses::ev_function_call; +use core_test_support::responses::ev_reasoning_item_added; +use core_test_support::responses::ev_reasoning_summary_text_delta; +use core_test_support::responses::ev_response_created; use core_test_support::skip_if_no_network; use core_test_support::streaming_sse::StreamingSseChunk; use core_test_support::streaming_sse::start_streaming_sse_server; use core_test_support::test_codex::TestCodex; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; +use serde_json::Value; +use serde_json::json; fn sse_incomplete() -> String { let fixture = find_resource!("tests/fixtures/incomplete_sse.json") @@ -26,7 +32,19 @@ fn sse_incomplete() -> String { async fn retries_on_early_close() { skip_if_no_network!(); - let incomplete_sse = sse_incomplete(); + let call_id = "call-retry"; + let reasoning_id = "reasoning-retry"; + let args = json!({ + "command": "printf retry", + "timeout_ms": 1_000 + }) + .to_string(); + let incomplete_sse = responses::sse(vec![ + ev_response_created("resp_incomplete"), + ev_reasoning_item_added(reasoning_id, &[""]), + ev_reasoning_summary_text_delta("thinking"), + ev_function_call(call_id, "shell_command", &args), + ]); let completed_sse = responses::sse_completed("resp_ok"); let (server, _) = start_streaming_sse_server(vec![ @@ -94,6 +112,35 @@ async fn retries_on_early_close() { 2, "expected retry after incomplete SSE stream" ); + let second_body: Value = serde_json::from_slice(&requests[1]).expect("parse second request"); + let second_input = second_body["input"] + .as_array() + .expect("second request input should be an array"); + let reasoning_index = second_input + .iter() + .position(|item| { + item.get("type").and_then(Value::as_str) == Some("reasoning") + && item.get("id").and_then(Value::as_str) == Some(reasoning_id) + }) + .expect("retry request should include the interrupted reasoning item"); + let function_call_index = second_input + .iter() + .position(|item| { + item.get("type").and_then(Value::as_str) == Some("function_call") + && item.get("call_id").and_then(Value::as_str) == Some(call_id) + }) + .expect("retry request should include the interrupted function call"); + let function_output_index = second_input + .iter() + .position(|item| { + item.get("type").and_then(Value::as_str) == Some("function_call_output") + && item.get("call_id").and_then(Value::as_str) == Some(call_id) + }) + .expect("retry request should include the tool output from the interrupted call"); + assert!( + reasoning_index < function_call_index && function_call_index < function_output_index, + "expected retry replay order reasoning -> function_call -> function_call_output" + ); server.shutdown().await; }