diff --git a/codex-rs/cli/src/doctor.rs b/codex-rs/cli/src/doctor.rs index fd1f1c8fa1..af478a9d12 100644 --- a/codex-rs/cli/src/doctor.rs +++ b/codex-rs/cli/src/doctor.rs @@ -3305,6 +3305,7 @@ mod tests { ApiError::Stream(message()), ApiError::ResponseIncomplete(Box::new( codex_protocol::error::ResponseIncompleteError { + partial_output: vec![], reason: message(), response_id: Some(message()), token_usage: None, diff --git a/codex-rs/codex-api/src/sse/responses.rs b/codex-rs/codex-api/src/sse/responses.rs index 67f627e601..055feeca44 100644 --- a/codex-rs/codex-api/src/sse/responses.rs +++ b/codex-rs/codex-api/src/sse/responses.rs @@ -475,6 +475,15 @@ pub fn process_responses_event( .map(Into::into); return Err(ResponsesEventError::Api(ApiError::ResponseIncomplete( Box::new(ResponseIncompleteError { + partial_output: response + .and_then(|r| r.get("output")?.as_array()) + .into_iter() + .flatten() + .filter(|item| { + item.get("status").and_then(Value::as_str) == Some("incomplete") + }) + .cloned() + .collect(), reason: reason.to_string(), response_id, token_usage, @@ -1407,6 +1416,25 @@ mod tests { } } + #[tokio::test] + async fn incomplete_response_retains_partial_output_on_the_error() { + let call = json!({ + "type": "function_call", "status": "incomplete", "id": "fc_attempt", + "call_id": "attempt", "name": "unknown", "arguments": "{}", + }); + let terminal = json!({ + "type": "response.incomplete", + "response": { + "id": "resp_1", "incomplete_details": {"reason": "invalid_recipient"}, + "output": [{"type": "message", "status": "completed"}, call], + } + }); + let sse = format!("event: response.incomplete\ndata: {terminal}\n\n"); + let events = collect_events(&[sse.as_bytes()]).await; + assert_matches!(events.as_slice(), [Err(ApiError::ResponseIncomplete(error))] + if error.reason == "invalid_recipient" && error.partial_output == [call]); + } + #[tokio::test] async fn typed_errors_handle_missing_or_blank_message() { for (code, fallback) in [ diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 6f780e5de6..889a62680f 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -112,6 +112,7 @@ use futures::StreamExt; use http::HeaderMap as ApiHeaderMap; use http::HeaderValue; use http::StatusCode; +use serde::Deserialize; use std::time::Duration; use std::time::Instant; use tokio::sync::mpsc; @@ -156,6 +157,7 @@ use codex_model_provider_info::DEFAULT_WEBSOCKET_CONNECT_TIMEOUT_MS; use codex_model_provider_info::ModelProviderInfo; use codex_model_provider_info::WireApi; use codex_protocol::error::CodexErr; +use codex_protocol::error::CodexErrorDetails; use codex_protocol::error::Result; use codex_response_debug_context::extract_response_debug_context; use codex_response_debug_context::extract_response_debug_context_from_api_error; @@ -2793,11 +2795,30 @@ where feedback_tags!(last_model_request_id = upstream_request_id); } let mapped = provider.map_api_error(err); + // Terminal response output is trace evidence. Only streamed done items + // belong to the continuation cache or reach tool dispatch. + let completed_item_count = items_added.len(); + if inference_trace_attempt.needs_output_items() + && let CodexErrorDetails::ResponseIncomplete(incomplete) = mapped.details() + { + for raw in &incomplete.partial_output { + // Other discards unknown payloads and cannot be replayed as a trace item. + if let Ok(item) = ResponseItem::deserialize(raw) + && !matches!(item, ResponseItem::Other) + && !item.id().is_some_and(|id| { + items_added.iter().any(|seen| seen.id() == Some(id)) + }) + { + items_added.push(item); + } + } + } inference_trace_attempt.record_failed( &mapped, upstream_request_id, &items_added, ); + items_added.truncate(completed_item_count); if !logged_error { session_telemetry.see_event_completed_failed(&mapped); logged_error = true; diff --git a/codex-rs/core/src/client_tests.rs b/codex-rs/core/src/client_tests.rs index 9d4257dee2..b6a539f56c 100644 --- a/codex-rs/core/src/client_tests.rs +++ b/codex-rs/core/src/client_tests.rs @@ -1686,6 +1686,65 @@ async fn untraced_http_stream_releases_completed_output_before_response_ends() { assert_eq!(payload.strong_count(), 0); } +#[tokio::test] +async fn incomplete_response_traces_attempted_call_without_streaming_it() -> anyhow::Result<()> { + let item = json!({ + "type": "function_call", + "status": "incomplete", + "id": "fc_attempt", + "call_id": "attempt", + "name": "unknown_tool", + "arguments": "{}", + }); + let unknown = json!({"type": "future_response_item", "status": "incomplete"}); + for (partial_output, expected_call_ids) in [ + (vec![item.clone()], vec![Some("attempt")]), + (vec![unknown.clone(), item], vec![Some("attempt")]), + (vec![unknown], vec![]), + ] { + let temp = TempDir::new()?; + let attempt = started_inference_attempt(&temp)?; + let api_stream = futures::stream::iter([Err(codex_api::ApiError::ResponseIncomplete( + Box::new(codex_protocol::error::ResponseIncompleteError { + reason: "invalid_recipient".to_string(), + response_id: Some("response".to_string()), + token_usage: None, + partial_output, + }), + ))]); + let mut stream = super::map_response_events( + /*upstream_request_id*/ None, + api_stream, + test_session_telemetry(), + attempt, + test_model_provider(), + /*last_response_tx*/ None, + ); + assert!(matches!(stream.next().await, Some(Err(_)))); + assert!(stream.next().await.is_none()); + + let rollout = replay_bundle(temp.path())?; + let inference = rollout.inference_calls.values().next().unwrap(); + assert_eq!(inference.execution.status, ExecutionStatus::Failed); + let call_ids: Vec<_> = inference + .response_item_ids + .iter() + .map(|id| rollout.conversation_items[id].call_id.as_deref()) + .collect(); + assert_eq!(call_ids, expected_call_ids); + assert_eq!( + inference.raw_response_payload_id.is_some(), + !expected_call_ids.is_empty() + ); + assert_eq!( + rollout.threads["thread-root"].conversation_item_ids, + inference.request_item_ids + ); + assert!(rollout.tool_calls.is_empty()); + } + Ok(()) +} + #[tokio::test] async fn dropped_response_stream_traces_cancelled_partial_output() -> anyhow::Result<()> { let temp = TempDir::new()?; diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 85a3e6c91d..f84dc6e1bb 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -1246,6 +1246,7 @@ mod tests { let stream = response_stream(vec![Err(CodexErr::new( CodexErrorDetails::ResponseIncomplete(Box::new( codex_protocol::error::ResponseIncompleteError { + partial_output: vec![], reason: "max_output_tokens".to_string(), response_id: Some("resp-background".to_string()), token_usage: Some(TokenUsage { diff --git a/codex-rs/protocol/src/error.rs b/codex-rs/protocol/src/error.rs index a8a8827b23..e48cf5abd0 100644 --- a/codex-rs/protocol/src/error.rs +++ b/codex-rs/protocol/src/error.rs @@ -83,9 +83,12 @@ pub struct CodexErr { /// A terminal response that stopped before the model finished, with any reported usage. /// Content-filter retries and explicit interruptions have separate handling. -#[derive(Debug, Error, PartialEq, Eq)] +#[derive(derive_more::Debug, Error, PartialEq, Eq)] #[error("Incomplete response returned, reason: {reason}")] pub struct ResponseIncompleteError { + /// Provider output items explicitly marked incomplete, retained for failed inference traces. + #[debug(skip)] + pub partial_output: Vec, pub reason: String, pub response_id: Option, pub token_usage: Option,