From 2a14c1b3668e006a9b4c06d205c6adcdac590be6 Mon Sep 17 00:00:00 2001 From: mulfyx Date: Sat, 8 Aug 2026 21:52:55 +0500 Subject: [PATCH 1/2] fix(codex): unify stream event semantics --- .../codex/chat_completions/response.rs | 36 +- src/providers/codex/client.rs | 222 +++++----- src/providers/codex/events.rs | 386 +++++++++++++----- src/providers/codex/mod.rs | 195 ++++----- src/providers/codex/translate/live_stream.rs | 83 ++-- src/providers/codex/translate/reducer.rs | 173 ++------ src/providers/codex/websocket.rs | 22 +- tests/codex_agent_continuation.rs | 12 +- tests/smoke_cutover.rs | 318 ++++++++++++++- 9 files changed, 957 insertions(+), 490 deletions(-) diff --git a/src/providers/codex/chat_completions/response.rs b/src/providers/codex/chat_completions/response.rs index a865cdbe..0afb6f8b 100644 --- a/src/providers/codex/chat_completions/response.rs +++ b/src/providers/codex/chat_completions/response.rs @@ -1,6 +1,8 @@ +use http::StatusCode; use serde_json::{Value, json}; use crate::anthropic::sse::parse_sse_events; +use crate::providers::codex::events::classify_event_failure; use super::ChatError; @@ -46,8 +48,16 @@ impl CompletionState { pub fn observe(&mut self, event: &Value) -> Result, ChatError> { let kind = event.get("type").and_then(Value::as_str); - if matches!(kind, Some("response.failed" | "response.error" | "error")) { - return Err(event_error(event)); + if let Some(failure) = classify_event_failure(event) { + let status = + StatusCode::from_u16(failure.client_status()).unwrap_or(StatusCode::BAD_GATEWAY); + return Err(ChatError::new( + status, + failure.client_error_type(), + failure.message, + None, + None, + )); } match kind { Some("response.created" | "response.in_progress") => { @@ -62,7 +72,7 @@ impl CompletionState { self.text.push_str(delta); return Ok(Some(delta.to_string())); } - Some("response.completed" | "response.incomplete") => { + Some("response.completed" | "response.done") => { let response = event.get("response").ok_or_else(|| { ChatError::upstream("Codex completion event did not contain a response") })?; @@ -70,11 +80,6 @@ impl CompletionState { if response.get("status").and_then(Value::as_str) == Some("failed") { return Err(event_error(event)); } - if kind == Some("response.incomplete") - || response.get("status").and_then(Value::as_str) == Some("incomplete") - { - self.finish_reason = "length"; - } self.completed = true; } _ => { @@ -199,10 +204,19 @@ mod tests { } #[test] - fn incomplete_maps_to_length() { + fn incomplete_is_an_upstream_error() { let body = b"data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\ndata: {\"type\":\"response.incomplete\",\"response\":{\"status\":\"incomplete\"}}\n\n"; - let value = aggregate_sse(body, "model").unwrap(); - assert_eq!(value["choices"][0]["finish_reason"], "length"); + let error = aggregate_sse(body, "model").unwrap_err(); + assert!(error.message.contains("Incomplete response")); + } + + #[test] + fn buffered_chat_preserves_typed_terminal_failure() { + let body = b"data: {\"type\":\"response.failed\",\"response\":{\"status\":\"failed\",\"error\":{\"code\":\"context_length_exceeded\",\"message\":\"input exceeds context window\"}}}\n\n"; + let error = aggregate_sse(body, "model").unwrap_err(); + assert_eq!(error.status, StatusCode::PAYLOAD_TOO_LARGE); + assert_eq!(error.kind, "request_too_large"); + assert_eq!(error.message, "input exceeds context window"); } #[test] diff --git a/src/providers/codex/client.rs b/src/providers/codex/client.rs index bebaeed8..8f37ab81 100644 --- a/src/providers/codex/client.rs +++ b/src/providers/codex/client.rs @@ -1462,6 +1462,7 @@ impl CodexHttpClient { ); } + let event_kind = super::events::classify_stream_event(&payload); let failure = super::events::classify_event_failure(&payload); if !semantic_output_forwarded && let Some(failure) = failure.as_ref() @@ -1471,30 +1472,53 @@ impl CodexHttpClient { break 'read_attempt codex_event_failure_error(failure.clone()); } - let terminal = event_closes_http_stream(&payload); - if !semantic_output_forwarded && failure.is_some() { - pending_events.clear(); - if tx.send(Ok(payload)).await.is_err() { - return; + let terminal = super::events::event_is_terminal(&payload); + match event_kind { + super::events::CodexStreamEventKind::TerminalFailure => { + if !semantic_output_forwarded { + pending_events.clear(); + } + if tx.send(Ok(payload)).await.is_err() { + return; + } } - } else if !semantic_output_forwarded - && http_event_starts_semantic_output(&payload) - { - semantic_output_forwarded = true; - for pending in pending_events.drain(..) { - if tx.send(Ok(pending)).await.is_err() { + super::events::CodexStreamEventKind::TerminalSuccess => { + for pending in pending_events.drain(..) { + if tx.send(Ok(pending)).await.is_err() { + return; + } + } + if tx.send(Ok(payload)).await.is_err() { return; } } - if tx.send(Ok(payload)).await.is_err() { - return; + super::events::CodexStreamEventKind::Semantic => { + if !semantic_output_forwarded { + semantic_output_forwarded = true; + for pending in pending_events.drain(..) { + if tx.send(Ok(pending)).await.is_err() { + return; + } + } + } + if tx.send(Ok(payload)).await.is_err() { + return; + } } - } else if semantic_output_forwarded || http_event_is_control(&payload) { - if tx.send(Ok(payload)).await.is_err() { - return; + super::events::CodexStreamEventKind::Control => { + if tx.send(Ok(payload)).await.is_err() { + return; + } + } + super::events::CodexStreamEventKind::Structural => { + if semantic_output_forwarded { + if tx.send(Ok(payload)).await.is_err() { + return; + } + } else { + pending_events.push(payload); + } } - } else { - pending_events.push(payload); } if terminal { @@ -2081,6 +2105,7 @@ impl CodexHttpClient { } }; + let mut pending_events = Vec::new(); loop { let item = tokio::select! { biased; @@ -2174,20 +2199,71 @@ impl CodexHttpClient { continue 'attempt; } - if item.as_ref().is_ok_and(event_closes_live_retry_window) { - forwarded_any = true; - } let terminal = item.as_ref().is_err() || item.as_ref().is_ok_and(super::websocket::is_terminal_event); - if tx.send(item).await.is_err() { - if let Some(reservation) = continuation.as_ref() { - super::websocket::invalidate_codex_websocket_pool_socket( - reservation, - stream.socket_id(), - ); + match item { + Ok(payload) => match super::events::classify_stream_event(&payload) { + super::events::CodexStreamEventKind::TerminalFailure => { + if !forwarded_any { + pending_events.clear(); + } + if tx.send(Ok(payload)).await.is_err() { + return; + } + } + super::events::CodexStreamEventKind::TerminalSuccess => { + for pending in pending_events.drain(..) { + if tx.send(Ok(pending)).await.is_err() { + return; + } + } + if tx.send(Ok(payload)).await.is_err() { + return; + } + } + super::events::CodexStreamEventKind::Semantic => { + if !forwarded_any { + forwarded_any = true; + for pending in pending_events.drain(..) { + if tx.send(Ok(pending)).await.is_err() { + return; + } + } + } + if tx.send(Ok(payload)).await.is_err() { + return; + } + } + super::events::CodexStreamEventKind::Control => { + if tx.send(Ok(payload)).await.is_err() { + return; + } + } + super::events::CodexStreamEventKind::Structural => { + if forwarded_any { + if tx.send(Ok(payload)).await.is_err() { + return; + } + } else { + pending_events.push(payload); + } + } + }, + Err(err) => { + if tx.send(Err(err)).await.is_err() { + if let Some(reservation) = continuation.as_ref() { + super::websocket::invalidate_codex_websocket_pool_socket( + reservation, + stream.socket_id(), + ); + } + abort_abandoned_live_continuation( + continuation.as_ref(), + &socket_id_publisher, + ); + return; + } } - abort_abandoned_live_continuation(continuation.as_ref(), &socket_id_publisher); - return; } if terminal { return; @@ -2458,62 +2534,6 @@ fn response_headers(resp: &reqwest::Response) -> Vec<(String, String)> { .collect() } -fn event_closes_http_stream(payload: &serde_json::Value) -> bool { - matches!( - payload.get("type").and_then(|value| value.as_str()), - Some( - "response.completed" - | "response.incomplete" - | "response.done" - | "response.failed" - | "response.error" - | "error" - ) - ) -} - -fn http_event_is_control(payload: &serde_json::Value) -> bool { - matches!( - payload.get("type").and_then(|value| value.as_str()), - Some( - "keepalive" - | "response.created" - | "response.in_progress" - | "codex.rate_limits" - | "response.web_search_call.in_progress" - | "response.web_search_call.searching" - | "response.web_search_call.completed" - ) - ) -} - -fn http_event_starts_semantic_output(payload: &serde_json::Value) -> bool { - match payload.get("type").and_then(|value| value.as_str()) { - Some("response.output_item.added") => matches!( - payload - .pointer("/item/type") - .and_then(|value| value.as_str()), - Some("message" | "function_call") - ), - Some( - "response.reasoning_summary_text.delta" - | "response.output_text.delta" - | "response.function_call_arguments.delta", - ) => payload - .get("delta") - .and_then(|value| value.as_str()) - .is_some_and(|delta| !delta.is_empty()), - Some("response.output_item.done") => matches!( - payload - .pointer("/item/type") - .and_then(|value| value.as_str()), - Some("reasoning" | "message" | "function_call") - ), - Some("response.completed" | "response.incomplete" | "response.done") => true, - _ => false, - } -} - fn codex_event_failure_error(failure: super::events::CodexEventFailure) -> CodexError { CodexError { status: failure.status, @@ -2953,11 +2973,9 @@ fn invalidate_live_continuation_pool( super::websocket::invalidate_codex_websocket_pool_turn_for_owner(owner, continuation.turn_id()); } +#[cfg(test)] fn event_closes_live_retry_window(payload: &serde_json::Value) -> bool { - !matches!( - payload.get("type").and_then(|value| value.as_str()), - Some("codex.rate_limits" | "keepalive") - ) + super::events::classify_stream_event(payload) == super::events::CodexStreamEventKind::Semantic } pub(super) fn is_continuation_retry_error(err: &CodexError) -> bool { @@ -3432,7 +3450,7 @@ mod tests { .unwrap(); write_http_chunk( &mut stream, - b"data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_up\"}}\n\n", + b"data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"hello\"}\n\n", ) .await; release_rx.await.unwrap(); @@ -3462,7 +3480,7 @@ mod tests { .unwrap(); assert_eq!( first_upstream.get("type").and_then(|value| value.as_str()), - Some("response.output_item.added") + Some("response.output_text.delta") ); release_tx.send(()).unwrap(); @@ -5244,9 +5262,29 @@ mod tests { assert!(!event_closes_live_retry_window(&serde_json::json!({ "type": "keepalive" }))); - assert!(event_closes_live_retry_window(&serde_json::json!({ + assert!(!event_closes_live_retry_window(&serde_json::json!({ "type": "response.created" }))); + assert!(!event_closes_live_retry_window(&serde_json::json!({ + "type": "response.output_item.added", + "output_index": 0, + "item": {"type": "message", "id": "msg_1"} + }))); + assert!(!event_closes_live_retry_window(&serde_json::json!({ + "type": "response.output_item.added", + "output_index": 0, + "item": {"type": "function_call", "call_id": "call_1", "name": "Read"} + }))); + assert!(event_closes_live_retry_window(&serde_json::json!({ + "type": "response.output_text.delta", + "output_index": 0, + "delta": "hello" + }))); + assert!(event_closes_live_retry_window(&serde_json::json!({ + "type": "response.function_call_arguments.delta", + "output_index": 0, + "delta": "{}" + }))); } #[test] diff --git a/src/providers/codex/events.rs b/src/providers/codex/events.rs index 51befdf1..39f0def5 100644 --- a/src/providers/codex/events.rs +++ b/src/providers/codex/events.rs @@ -8,6 +8,15 @@ pub(crate) enum CodexFailureKind { Permanent, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum CodexStreamEventKind { + Control, + Structural, + Semantic, + TerminalSuccess, + TerminalFailure, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct CodexEventFailure { pub kind: CodexFailureKind, @@ -21,49 +30,111 @@ impl CodexEventFailure { pub fn retryable(&self) -> bool { !matches!(self.kind, CodexFailureKind::Permanent) } + + pub fn client_status(&self) -> u16 { + self.status + } + + pub fn client_error_type(&self) -> &'static str { + match self.client_status() { + 400 | 422 => "invalid_request_error", + 401 => "authentication_error", + 403 => "permission_error", + 413 => "request_too_large", + 429 => "rate_limit_error", + 529 => "overloaded_error", + _ => "api_error", + } + } } -pub(crate) fn is_terminal_rate_limit_event(payload: &Value) -> bool { - payload.get("type").and_then(Value::as_str) == Some("codex.rate_limits") - && payload - .pointer("/rate_limits/limit_reached") - .and_then(Value::as_bool) - == Some(true) - && payload - .pointer("/credits/has_credits") - .and_then(Value::as_bool) - != Some(true) - && payload - .pointer("/credits/unlimited") - .and_then(Value::as_bool) - != Some(true) +pub(crate) fn classify_stream_event(payload: &Value) -> CodexStreamEventKind { + if classify_event_failure(payload).is_some() { + return CodexStreamEventKind::TerminalFailure; + } + match payload.get("type").and_then(Value::as_str) { + Some("response.completed" | "response.done") => CodexStreamEventKind::TerminalSuccess, + Some( + "keepalive" + | "response.created" + | "response.in_progress" + | "codex.rate_limits" + | "response.web_search_call.in_progress" + | "response.web_search_call.searching" + | "response.web_search_call.completed", + ) => CodexStreamEventKind::Control, + Some( + "response.reasoning_summary_text.delta" + | "response.output_text.delta" + | "response.function_call_arguments.delta", + ) if payload + .get("delta") + .and_then(Value::as_str) + .is_some_and(|delta| !delta.is_empty()) => + { + CodexStreamEventKind::Semantic + } + Some("response.output_item.done") + if matches!( + payload.pointer("/item/type").and_then(Value::as_str), + Some("function_call" | "web_search_call") + ) => + { + CodexStreamEventKind::Semantic + } + _ => CodexStreamEventKind::Structural, + } +} + +pub(crate) fn event_is_terminal(payload: &Value) -> bool { + matches!( + classify_stream_event(payload), + CodexStreamEventKind::TerminalSuccess | CodexStreamEventKind::TerminalFailure + ) +} + +pub(crate) fn event_is_success_terminal(payload: &Value) -> bool { + classify_stream_event(payload) == CodexStreamEventKind::TerminalSuccess } pub(crate) fn event_error(payload: &Value) -> Option<&Value> { payload .get("error") - .or_else(|| payload.pointer("/response/error")) + .filter(|error| !error.is_null()) + .or_else(|| { + payload + .pointer("/response/error") + .filter(|error| !error.is_null()) + }) } pub(crate) fn classify_event_failure(payload: &Value) -> Option { let event_type = payload.get("type").and_then(Value::as_str)?; + let response_status = payload.pointer("/response/status").and_then(Value::as_str); if event_type == "codex.rate_limits" { - if !is_terminal_rate_limit_event(payload) { - return None; - } + return None; + } + if event_type == "response.incomplete" || response_status == Some("incomplete") { + let reason = payload + .pointer("/response/incomplete_details/reason") + .and_then(Value::as_str) + .unwrap_or("unknown"); return Some(CodexEventFailure { - kind: CodexFailureKind::RateLimit, - explicit_status: Some(429), - status: 429, - message: "rate limit reached".to_string(), - retry_after: scalar_string(payload.pointer("/rate_limits/primary/reset_after_seconds")), + kind: CodexFailureKind::Transient, + explicit_status: None, + status: 503, + message: format!("Incomplete response returned, reason: {reason}"), + retry_after: None, }); } - if !matches!(event_type, "response.failed" | "response.error" | "error") { + let error = event_error(payload); + if !matches!(event_type, "response.failed" | "response.error" | "error") + && response_status != Some("failed") + && error.is_none() + { return None; } - let error = event_error(payload); let explicit_status = numeric_status(payload) .or_else(|| { error @@ -71,20 +142,47 @@ pub(crate) fn classify_event_failure(payload: &Value) -> Option { + "This request has been flagged for possible cybersecurity risk.".to_string() + } + (Some("cyber_policy"), Some(message)) if message.trim().is_empty() => { + "This request has been flagged for possible cybersecurity risk.".to_string() + } + (Some("invalid_prompt" | "bio_policy"), None) => "Invalid request.".to_string(), + (_, Some(message)) => message.to_string(), + _ => "Upstream error".to_string(), + }; let lower = message.to_ascii_lowercase(); + let context_window = code == Some("context_length_exceeded") + || lower.contains("context window") + || lower.contains("context length exceeded"); - let kind = if explicit_status == Some(429) || lower.contains("rate limit") { + let kind = if context_window + || matches!( + code, + Some( + "context_length_exceeded" + | "insufficient_quota" + | "usage_not_included" + | "cyber_policy" + | "invalid_prompt" + | "bio_policy" + ) + ) { + CodexFailureKind::Permanent + } else if matches!(code, Some("server_is_overloaded" | "slow_down")) { + CodexFailureKind::Overloaded + } else if explicit_status == Some(429) || lower.contains("rate limit") { CodexFailureKind::RateLimit } else if explicit_status == Some(529) || code == Some("overloaded_error") @@ -104,15 +202,28 @@ pub(crate) fn classify_event_failure(payload: &Value) -> Option 400, + Some("insufficient_quota") => 429, + Some("usage_not_included") => 403, + _ => match kind { + CodexFailureKind::RateLimit => 429, + CodexFailureKind::Overloaded => 529, + CodexFailureKind::Transient => 503, + CodexFailureKind::Permanent => 500, + }, + }) }; - let status = explicit_status.unwrap_or(match kind { - CodexFailureKind::RateLimit => 429, - CodexFailureKind::Overloaded => 529, - CodexFailureKind::Transient => 503, - CodexFailureKind::Permanent => 500, - }); let retry_after = error .and_then(|value| value.get("retry_after")) .and_then(scalar_string_value) @@ -135,6 +246,10 @@ pub(crate) fn classify_event_failure(payload: &Value) -> Option Option { + first_event_failure(body).filter(CodexEventFailure::retryable) +} + +pub(crate) fn first_event_failure(body: &[u8]) -> Option { for event in crate::anthropic::sse::parse_sse_events(body) { if event.data == "[DONE]" { continue; @@ -142,9 +257,7 @@ pub(crate) fn first_retryable_failure(body: &[u8]) -> Option let Ok(payload) = serde_json::from_str::(&event.data) else { continue; }; - if let Some(failure) = classify_event_failure(&payload) - && failure.retryable() - { + if let Some(failure) = classify_event_failure(&payload) { return Some(failure); } } @@ -203,14 +316,6 @@ mod tests { #[test] fn classifies_retryable_failure_kinds() { - let rate = classify_event_failure(&serde_json::json!({ - "type": "codex.rate_limits", - "rate_limits": {"limit_reached": true, "primary": {"reset_after_seconds": 1.5}} - })) - .unwrap(); - assert_eq!(rate.kind, CodexFailureKind::RateLimit); - assert_eq!(rate.retry_after.as_deref(), Some("1.5")); - let overload = classify_event_failure(&serde_json::json!({ "type": "response.failed", "response": {"error": {"type": "overloaded_error", "message": "busy"}} @@ -221,55 +326,12 @@ mod tests { } #[test] - fn terminal_rate_limit_honors_credits() { - // No credits field at all: legacy payload stays terminal. - assert!(is_terminal_rate_limit_event(&serde_json::json!({ - "type": "codex.rate_limits", - "rate_limits": {"limit_reached": true} - }))); - - // Credits exhausted: terminal. - assert!(is_terminal_rate_limit_event(&serde_json::json!({ - "type": "codex.rate_limits", - "rate_limits": {"limit_reached": true}, - "credits": {"has_credits": false, "unlimited": false} - }))); - - // Usable credits remain: informational. - assert!(!is_terminal_rate_limit_event(&serde_json::json!({ - "type": "codex.rate_limits", - "rate_limits": {"limit_reached": true}, - "credits": {"has_credits": true, "unlimited": false} - }))); - - // Unlimited plan: informational. - assert!(!is_terminal_rate_limit_event(&serde_json::json!({ - "type": "codex.rate_limits", - "rate_limits": {"limit_reached": true}, - "credits": {"has_credits": false, "unlimited": true} - }))); - - // Limit not reached: never terminal, credits irrelevant. - assert!(!is_terminal_rate_limit_event(&serde_json::json!({ - "type": "codex.rate_limits", - "rate_limits": {"limit_reached": false}, - "credits": {"has_credits": false, "unlimited": false} - }))); - - // Wrong event type never matches. - assert!(!is_terminal_rate_limit_event(&serde_json::json!({ - "type": "response.completed", - "rate_limits": {"limit_reached": true} - }))); - } - - #[test] - fn classifier_skips_credited_rate_limit_snapshots() { + fn rate_limit_snapshots_are_always_telemetry() { assert!( classify_event_failure(&serde_json::json!({ "type": "codex.rate_limits", "rate_limits": {"limit_reached": true}, - "credits": {"has_credits": true, "unlimited": false} + "credits": {"has_credits": false, "unlimited": false} })) .is_none() ); @@ -291,4 +353,140 @@ mod tests { .unwrap(); assert!(!failure.retryable()); } + + #[test] + fn classifies_progress_and_terminal_semantics() { + assert_eq!( + classify_stream_event(&serde_json::json!({"type":"response.created"})), + CodexStreamEventKind::Control + ); + assert_eq!( + classify_stream_event(&serde_json::json!({ + "type":"response.output_item.added", + "item":{"type":"function_call"} + })), + CodexStreamEventKind::Structural + ); + assert_eq!( + classify_stream_event(&serde_json::json!({ + "type":"response.function_call_arguments.delta", + "delta":"{}" + })), + CodexStreamEventKind::Semantic + ); + assert_eq!( + classify_stream_event(&serde_json::json!({"type":"response.incomplete"})), + CodexStreamEventKind::TerminalFailure + ); + assert_eq!( + classify_stream_event(&serde_json::json!({ + "type":"response.completed", + "error": null, + "response": { + "status":"failed", + "error":{"status":400,"code":"invalid_prompt","message":"rejected"} + } + })), + CodexStreamEventKind::TerminalFailure + ); + } + + #[test] + fn nested_response_error_survives_null_top_level_error() { + let failure = classify_event_failure(&serde_json::json!({ + "type":"response.completed", + "error": null, + "response": { + "status":"failed", + "error":{"status":400,"code":"invalid_prompt","message":"nested rejection"} + } + })) + .unwrap(); + assert_eq!(failure.client_status(), 400); + assert_eq!(failure.client_error_type(), "invalid_request_error"); + assert_eq!(failure.message, "nested rejection"); + assert!(!failure.retryable()); + } + + #[test] + fn completed_event_with_nested_error_is_not_success() { + let payload = serde_json::json!({ + "type":"response.completed", + "error": null, + "response": { + "status":"completed", + "error":{"status":502,"code":"server_error","message":"late failure"} + } + }); + let failure = classify_event_failure(&payload).unwrap(); + assert_eq!(failure.client_status(), 502); + assert!(failure.retryable()); + assert_eq!( + classify_stream_event(&payload), + CodexStreamEventKind::TerminalFailure + ); + } + + #[test] + fn completed_event_with_nested_null_error_is_success() { + let payload = serde_json::json!({ + "type":"response.completed", + "error": null, + "response": {"status":"completed", "error":null} + }); + assert!(classify_event_failure(&payload).is_none()); + assert_eq!( + classify_stream_event(&payload), + CodexStreamEventKind::TerminalSuccess + ); + } + + #[test] + fn native_fatal_codes_are_not_retried_without_numeric_status() { + for (code, status, error_type) in [ + ("context_length_exceeded", 413, "request_too_large"), + ("cyber_policy", 400, "invalid_request_error"), + ("invalid_prompt", 400, "invalid_request_error"), + ("bio_policy", 400, "invalid_request_error"), + ("insufficient_quota", 429, "rate_limit_error"), + ("usage_not_included", 403, "permission_error"), + ] { + let failure = classify_event_failure(&serde_json::json!({ + "type":"response.failed", + "response":{"status":"failed","error":{"code":code,"message":"rejected"}} + })) + .unwrap(); + assert!(!failure.retryable(), "{code}"); + assert_eq!(failure.client_status(), status, "{code}"); + assert_eq!(failure.client_error_type(), error_type, "{code}"); + } + } + + #[test] + fn native_overload_codes_remain_retryable_without_numeric_status() { + for code in ["server_is_overloaded", "slow_down"] { + let failure = classify_event_failure(&serde_json::json!({ + "type":"response.failed", + "response":{"status":"failed","error":{"code":code,"message":"busy"}} + })) + .unwrap(); + assert!(failure.retryable(), "{code}"); + assert_eq!(failure.client_status(), 529, "{code}"); + assert_eq!(failure.client_error_type(), "overloaded_error", "{code}"); + } + } + + #[test] + fn context_window_message_overrides_generic_upstream_400() { + let failure = classify_event_failure(&serde_json::json!({ + "type":"error", + "status":400, + "error":{"type":"invalid_request_error","message":"input exceeds context window"} + })) + .unwrap(); + assert!(!failure.retryable()); + assert_eq!(failure.explicit_status, Some(400)); + assert_eq!(failure.client_status(), 413); + assert_eq!(failure.client_error_type(), "request_too_large"); + } } diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index 54aaad7a..c09f9302 100644 --- a/src/providers/codex/mod.rs +++ b/src/providers/codex/mod.rs @@ -407,6 +407,11 @@ impl CodexProvider { attempt += 1; sleep(delay.wait_ms).await; }; + if let Some(failure) = events::first_event_failure(&upstream.body) { + abort_compaction_attempt(ctx.session_id.as_deref(), compaction_attempt); + abort_continuation_for_owner(&request_continuation); + return map_codex_event_failure_to_response(&failure); + } log.info( "codex_upstream_response_received", Some(serde_json::Map::from_iter([ @@ -885,39 +890,24 @@ async fn live_stream_response_once( { Ok(result) => result, Err(message) => { - if retryable_live_start_payload(&payload, &message) { - let lower_message = message.to_ascii_lowercase(); - let status = websocket::event_error_status(&payload).unwrap_or_else(|| { - let error = payload.get("error").or_else(|| { - payload.get("response").and_then(|value| value.get("error")) - }); - let overloaded = error.is_some_and(|error| { - error.get("code").and_then(|value| value.as_str()) - == Some("overloaded_error") - || error.get("type").and_then(|value| value.as_str()) - == Some("overloaded_error") - }); - if payload.get("type").and_then(|value| value.as_str()) - == Some("codex.rate_limits") - || lower_message.contains("rate limit") - { - 429 - } else if overloaded || lower_message.contains("overloaded") { - 529 - } else { - 503 - } - }); - return provider_retry( - &upstream_events, - client::CodexError { - status, - message: message.clone(), - detail: Some(message), - retry_after: retry_after_from_live_payload(&payload), - origin: client::CodexErrorOrigin::WebSocket, - }, + if let Some(failure) = events::classify_event_failure(&payload) { + if failure.retryable() { + return provider_retry( + &upstream_events, + codex_event_failure_error( + &failure, + client::CodexErrorOrigin::WebSocket, + ), + ); + } + abort_request_state( + ctx.session_id.as_deref(), + &request_continuation, + compaction.attempt, ); + return LiveStreamStart::Response(map_codex_event_failure_to_response( + &failure, + )); } abort_request_state( ctx.session_id.as_deref(), @@ -1014,7 +1004,7 @@ fn translate_live_stream_payload( traffic: Option<&crate::traffic::TrafficCapture>, ) -> Result<(Vec, bool), String> { let chunk = translator.accept(payload, traffic)?; - let terminal = is_codex_terminal_event(payload) || translator.is_finished(); + let terminal = events::event_is_terminal(payload) || translator.is_finished(); Ok((chunk, terminal)) } @@ -1129,9 +1119,16 @@ fn remaining_live_stream_response( &request_continuation, compaction.attempt, ); + let failure = events::classify_event_failure(&payload); + let error_type = failure + .as_ref() + .map_or("api_error", events::CodexEventFailure::client_error_type); + let error_message = failure + .as_ref() + .map_or(message.as_str(), |failure| failure.message.as_str()); let chunk = translator.error_chunk( - &message, - "api_error", + error_message, + error_type, ctx.traffic.as_deref(), ); if !chunk.is_empty() { @@ -1171,12 +1168,14 @@ fn remaining_live_stream_response( &request_continuation, compaction.attempt, ); - let chunk = - translator.finish_after_closed_completed_tool_call(ctx.traffic.as_deref()); - if !chunk.is_empty() { - record_live_stream_progress(&ctx, &chunk); - let _ = tx.send(Ok(Bytes::from(chunk))).await; - return; + if err.origin == client::CodexErrorOrigin::WebSocket { + let chunk = translator + .finish_after_closed_completed_tool_call(ctx.traffic.as_deref()); + if !chunk.is_empty() { + record_live_stream_progress(&ctx, &chunk); + let _ = tx.send(Ok(Bytes::from(chunk))).await; + return; + } } let error_type = codex_stream_error_type(&err); let chunk = translator.error_chunk( @@ -1282,23 +1281,8 @@ fn is_empty_codex_success_completion(upstream_sse: &[u8]) -> bool { saw_success_terminal } -fn is_codex_terminal_event(payload: &serde_json::Value) -> bool { - matches!( - payload.get("type").and_then(|v| v.as_str()), - Some("response.completed") - | Some("response.incomplete") - | Some("response.done") - | Some("response.failed") - | Some("response.error") - | Some("error") - ) -} - fn is_codex_success_terminal_event(payload: &serde_json::Value) -> bool { - matches!( - payload.get("type").and_then(|v| v.as_str()), - Some("response.completed") | Some("response.done") - ) + events::event_is_success_terminal(payload) } fn retryable_live_start_codex_error(err: &client::CodexError) -> bool { @@ -1355,12 +1339,17 @@ fn retryable_live_message(message: &str) -> bool { .any(|needle| lower.contains(needle)) } -fn retryable_live_start_payload(payload: &serde_json::Value, _message: &str) -> bool { - events::classify_event_failure(payload).is_some_and(|failure| failure.retryable()) -} - -fn retry_after_from_live_payload(payload: &serde_json::Value) -> Option { - events::classify_event_failure(payload).and_then(|failure| failure.retry_after) +fn codex_event_failure_error( + failure: &events::CodexEventFailure, + origin: client::CodexErrorOrigin, +) -> client::CodexError { + client::CodexError { + status: failure.status, + message: failure.message.clone(), + detail: Some(failure.message.clone()), + retry_after: failure.retry_after.clone(), + origin, + } } fn codex_stream_error_type(err: &client::CodexError) -> &'static str { @@ -1480,6 +1469,16 @@ fn map_codex_failure_to_response(message: &str) -> Response { } } +fn map_codex_event_failure_to_response(failure: &events::CodexEventFailure) -> Response { + let status = StatusCode::from_u16(failure.client_status()).unwrap_or(StatusCode::BAD_GATEWAY); + let response = json_error(status, failure.client_error_type(), &failure.message); + if let Some(retry_after) = failure.retry_after.as_deref() { + ([(http::header::RETRY_AFTER, retry_after)], response).into_response() + } else { + response + } +} + fn is_context_window_overflow(message: &str) -> bool { message.to_ascii_lowercase().contains("context window") } @@ -2129,28 +2128,28 @@ mod tests { } #[test] - fn live_start_payload_retry_detection_covers_rate_limit_and_overload() { - assert!(retryable_live_start_payload( - &serde_json::json!({ + fn live_start_payload_retry_detection_uses_event_failure_classification() { + assert!( + events::classify_event_failure(&serde_json::json!({ "type": "codex.rate_limits", "rate_limits": {"limit_reached": true} - }), - "rate limit reached", - )); - assert!(retryable_live_start_payload( - &serde_json::json!({ + })) + .is_none() + ); + assert!( + events::classify_event_failure(&serde_json::json!({ "type": "response.failed", "response": {"error": {"type": "overloaded_error", "message": "overloaded"}} - }), - "overloaded", - )); - assert!(!retryable_live_start_payload( - &serde_json::json!({ + })) + .is_some_and(|failure| failure.retryable()) + ); + assert!( + !events::classify_event_failure(&serde_json::json!({ "type": "response.failed", - "response": {"error": {"message": "bad request"}} - }), - "bad request", - )); + "response": {"error": {"status": 400, "code": "invalid_prompt", "message": "bad request"}} + })) + .is_some_and(|failure| failure.retryable()) + ); } async fn run_live_failure_case( @@ -2405,11 +2404,15 @@ mod tests { let status = run_live_failure_case( "live-retry-exhaustion-cleanup", serde_json::json!({ - "type": "codex.rate_limits", - "rate_limits": { - "allowed": false, - "limit_reached": true, - "primary": {"reset_after_seconds": 0} + "type": "response.failed", + "response": { + "status": "failed", + "error": { + "status": 429, + "code": "rate_limit_exceeded", + "message": "rate limit reached", + "retry_after_seconds": 0 + } } }), 11, @@ -2425,11 +2428,15 @@ mod tests { let status = run_live_failure_case( "live-excessive-retry-after-cleanup", serde_json::json!({ - "type": "codex.rate_limits", - "rate_limits": { - "allowed": false, - "limit_reached": true, - "primary": {"reset_after_seconds": 31} + "type": "response.failed", + "response": { + "status": "failed", + "error": { + "status": 429, + "code": "rate_limit_exceeded", + "message": "rate limit reached", + "retry_after_seconds": 31 + } } }), 1, @@ -2448,13 +2455,17 @@ mod tests { "type": "response.failed", "response": { "status": "failed", - "error": {"message": "invalid request"} + "error": { + "status": 400, + "code": "invalid_prompt", + "message": "invalid request" + } } }), 1, ) .await; - assert_eq!(status, StatusCode::BAD_GATEWAY); + assert_eq!(status, StatusCode::BAD_REQUEST); } #[tokio::test] diff --git a/src/providers/codex/translate/live_stream.rs b/src/providers/codex/translate/live_stream.rs index 8a20163f..56b067a7 100644 --- a/src/providers/codex/translate/live_stream.rs +++ b/src/providers/codex/translate/live_stream.rs @@ -1,14 +1,12 @@ use std::collections::HashMap; use crate::anthropic::sse::encode_sse_event; -use crate::providers::codex::events::is_terminal_rate_limit_event; +use crate::providers::codex::events::classify_event_failure; use crate::traffic::TrafficCapture; use super::read_rewrite::sanitize_read_args; use super::reasoning_signature::{PendingReasoning, encode_reasoning_signature}; -use super::reducer::{ - CodexUsage, STOP_END_TURN, STOP_MAX_TOKENS, STOP_TOOL_USE, map_codex_usage_to_anthropic, -}; +use super::reducer::{CodexUsage, STOP_END_TURN, STOP_TOOL_USE, map_codex_usage_to_anthropic}; const BUFFERED_READ_REPAIR_TRAILING_WHITESPACE_BYTES: usize = 1_024; const BUFFERED_TOOL_MAX_ARGS_BYTES: usize = 5_000_000; @@ -112,19 +110,17 @@ impl LiveStreamTranslator { let kind = payload.get("type").and_then(|v| v.as_str()).unwrap_or(""); let mut out = Vec::new(); + if let Some(failure) = classify_event_failure(payload) { + return Err(failure.message); + } + match kind { "codex.rate_limits" => { - if is_terminal_rate_limit_event(payload) { - return Err("rate limit reached".to_string()); - } self.emit_ping(traffic, &mut out); } "keepalive" | "response.created" | "response.in_progress" => { self.emit_ping(traffic, &mut out); } - "response.failed" | "response.error" | "error" => { - return Err(error_message(payload)); - } "response.web_search_call.in_progress" | "response.web_search_call.searching" | "response.web_search_call.completed" => {} @@ -167,7 +163,7 @@ impl LiveStreamTranslator { "response.output_item.done" => { self.output_item_done(payload, traffic, &mut out); } - "response.completed" | "response.incomplete" | "response.done" => { + "response.completed" | "response.done" => { self.finish(payload, traffic, &mut out); } _ => {} @@ -339,7 +335,6 @@ impl LiveStreamTranslator { "function_call" => { self.close_thinking(traffic, out); self.saw_tool_use = true; - self.semantic_output_started = true; let index = self.anthropic_index; self.anthropic_index += 1; let call_id = item @@ -530,6 +525,7 @@ impl LiveStreamTranslator { if delta.is_empty() { return Ok(()); } + self.semantic_output_started = true; let mut repaired_read: Option<(usize, String)> = None; let Some(LiveBlock::Tool { index, @@ -562,6 +558,7 @@ impl LiveStreamTranslator { } else { *emitted_args = true; let index = *index; + self.semantic_output_started = true; self.emit( traffic, out, @@ -579,6 +576,7 @@ impl LiveStreamTranslator { } if let Some((index, repaired)) = repaired_read { self.blocks_by_output_index.remove(&output_index); + self.semantic_output_started = true; self.emit( traffic, out, @@ -713,6 +711,7 @@ impl LiveStreamTranslator { emitted_args, .. } => { + self.semantic_output_started = true; if let Some(final_args) = payload .get("item") .and_then(|item| item.get("arguments")) @@ -896,10 +895,7 @@ impl LiveStreamTranslator { self.emit_web_searches(traffic, out); self.ensure_message_start(traffic, out); let usage = payload.get("response").map(parse_codex_usage); - let incomplete = response_is_incomplete(payload); - let stop_reason = if incomplete { - STOP_MAX_TOKENS - } else if self.saw_tool_use { + let stop_reason = if self.saw_tool_use { STOP_TOOL_USE } else { STOP_END_TURN @@ -1095,21 +1091,6 @@ fn parse_codex_usage(response: &serde_json::Value) -> CodexUsage { } } -fn response_is_incomplete(payload: &serde_json::Value) -> bool { - payload.get("type").and_then(|v| v.as_str()) == Some("response.incomplete") - || payload - .get("response") - .and_then(|r| r.get("status")) - .and_then(|v| v.as_str()) - == Some("incomplete") - || payload - .get("response") - .and_then(|r| r.get("incomplete_details")) - .and_then(|d| d.get("reason")) - .and_then(|v| v.as_str()) - .is_some() -} - fn repair_whitespace_stalled_read_args( name: &str, args: &str, @@ -1178,22 +1159,6 @@ fn is_valid_read_args(value: &serde_json::Value) -> bool { true } -fn error_message(payload: &serde_json::Value) -> String { - payload - .get("response") - .and_then(|r| r.get("error")) - .and_then(|e| e.get("message")) - .and_then(|v| v.as_str()) - .or_else(|| { - payload - .get("error") - .and_then(|e| e.get("message")) - .and_then(|v| v.as_str()) - }) - .unwrap_or("Upstream error") - .to_string() -} - #[cfg(test)] mod tests { use super::*; @@ -1330,7 +1295,7 @@ mod tests { } #[test] - fn tool_thinking_and_web_search_events_are_semantic() { + fn structural_tool_start_is_not_semantic_until_arguments_arrive() { let mut tool = LiveStreamTranslator::new("msg_tool", "gpt-5.5"); tool.accept( &json!({ @@ -1341,6 +1306,16 @@ mod tests { None, ) .unwrap(); + assert!(!tool.has_semantic_output()); + tool.accept( + &json!({ + "type": "response.function_call_arguments.delta", + "output_index": 0, + "delta": "{}" + }), + None, + ) + .unwrap(); assert!(tool.has_semantic_output()); let mut thinking = LiveStreamTranslator::new("msg_thinking", "gpt-5.5"); @@ -1455,7 +1430,7 @@ mod tests { } #[test] - fn finishes_after_closed_completed_tool_call() { + fn websocket_compat_can_finish_after_closed_completed_tool_call() { let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5"); let mut out = Vec::new(); for event in [ @@ -1546,9 +1521,9 @@ mod tests { } #[test] - fn rate_limit_event_returns_error() { + fn rate_limit_event_is_progress_telemetry() { let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5"); - let err = translator + let out = translator .accept( &json!({ "type": "codex.rate_limits", @@ -1556,8 +1531,10 @@ mod tests { }), None, ) - .unwrap_err(); - assert_eq!(err, "rate limit reached"); + .unwrap(); + let out = String::from_utf8(out).unwrap(); + assert!(out.contains("event: ping")); + assert!(!out.contains("event: error")); } #[test] diff --git a/src/providers/codex/translate/reducer.rs b/src/providers/codex/translate/reducer.rs index 83beec6d..4212d441 100644 --- a/src/providers/codex/translate/reducer.rs +++ b/src/providers/codex/translate/reducer.rs @@ -1,5 +1,5 @@ use crate::anthropic::sse::parse_sse_events; -use crate::providers::codex::events::is_terminal_rate_limit_event; +use crate::providers::codex::events::{CodexFailureKind, classify_event_failure}; use super::read_rewrite::sanitize_read_args; use super::reasoning_signature::{PendingReasoning, ReasoningReplay, encode_reasoning_signature}; @@ -329,20 +329,28 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream event_count += 1; last_event_type = Some(t.clone()); + if let Some(failure) = classify_event_failure(&p) { + let kind = match failure.kind { + CodexFailureKind::RateLimit => UpstreamErrorKind::RateLimit, + CodexFailureKind::Overloaded => UpstreamErrorKind::Overloaded, + CodexFailureKind::Transient => UpstreamErrorKind::Transient, + CodexFailureKind::Permanent => UpstreamErrorKind::Failed, + }; + let retry_after_seconds = failure + .retry_after + .as_deref() + .and_then(|value| value.parse::().ok()) + .filter(|value| value.is_finite() && *value >= 0.0) + .map(|value| value.ceil() as u64); + return Err(UpstreamStreamError { + kind, + message: failure.message, + retry_after_seconds, + diagnostics: None, + }); + } + if t == "codex.rate_limits" { - if is_terminal_rate_limit_event(&p) { - let retry_after = p - .get("rate_limits") - .and_then(|r| r.get("primary")) - .and_then(|r| r.get("reset_after_seconds")) - .and_then(|v| v.as_f64()); - return Err(UpstreamStreamError { - kind: UpstreamErrorKind::RateLimit, - message: "rate limit reached".to_string(), - retry_after_seconds: retry_after.map(|f| f as u64), - diagnostics: None, - }); - } out.push(ReducerEvent::Progress); continue; } @@ -360,28 +368,6 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream continue; } - if t == "response.failed" || t == "response.error" || t == "error" { - let msg = p - .get("response") - .and_then(|r| r.get("error")) - .and_then(|e| e.get("message")) - .and_then(|v| v.as_str()) - .or_else(|| { - p.get("error") - .and_then(|e| e.get("message")) - .and_then(|v| v.as_str()) - }) - .unwrap_or("Upstream error"); - let kind = upstream_failure_kind(&p, msg); - let retry_after = retry_after_from_payload(&p); - return Err(UpstreamStreamError { - kind, - message: msg.to_string(), - retry_after_seconds: retry_after, - diagnostics: None, - }); - } - if t == "response.output_item.added" { let item = match p.get("item") { Some(v) => v, @@ -1008,82 +994,6 @@ fn server_tool_use_id_from_codex_web_search_id(id: &str) -> String { format!("srvtoolu_{suffix}") } -fn upstream_failure_kind(payload: &serde_json::Value, message: &str) -> UpstreamErrorKind { - let status = payload - .get("status") - .or_else(|| payload.get("status_code")) - .and_then(|v| v.as_u64()); - let code = payload - .get("response") - .and_then(|r| r.get("error")) - .and_then(|e| e.get("code")) - .or_else(|| payload.get("error").and_then(|e| e.get("code"))) - .and_then(|v| v.as_str()); - let err_type = payload - .get("response") - .and_then(|r| r.get("error")) - .and_then(|e| e.get("type")) - .or_else(|| payload.get("error").and_then(|e| e.get("type"))) - .and_then(|v| v.as_str()); - let lower_msg = message.to_lowercase(); - - if status == Some(529) - || code == Some("overloaded_error") - || err_type == Some("overloaded_error") - || lower_msg.contains("overloaded") - { - return UpstreamErrorKind::Overloaded; - } - - if (status.is_some_and(|s| (500..600).contains(&s))) - || code == Some("server_error") - || code == Some("internal_server_error") - || code == Some("internal_error") - || err_type == Some("server_error") - || err_type == Some("internal_server_error") - || err_type == Some("internal_error") - || is_retryable_transport_message(&lower_msg) - { - return UpstreamErrorKind::Transient; - } - - UpstreamErrorKind::Failed -} - -fn retry_after_from_payload(payload: &serde_json::Value) -> Option { - let raw = payload - .get("response") - .and_then(|r| r.get("error")) - .and_then(|e| e.get("retry_after_seconds")) - .or_else(|| { - payload - .get("error") - .and_then(|e| e.get("retry_after_seconds")) - }) - .or_else(|| payload.get("retry_after_seconds")) - .or_else(|| payload.get("headers").and_then(|h| h.get("retry-after"))) - .or_else(|| payload.get("headers").and_then(|h| h.get("Retry-After"))); - let value = match raw { - Some(v) if v.is_number() => v.as_f64(), - Some(v) if v.is_string() => v.as_str().and_then(|s| s.parse::().ok()), - _ => None, - }; - value.map(|f| f as u64) -} - -fn is_retryable_transport_message(msg: &str) -> bool { - msg.contains("you can retry your request") - || msg.contains("socket connection was closed unexpectedly") - || msg.contains("connection closed unexpectedly") - || msg.contains("connection reset") - || msg.contains("operation timed out") - || msg.contains("econnreset") - || msg.contains("epipe") - || msg.contains("etimedout") - || msg.contains("und_err_socket") - || msg.contains("fetch failed") -} - pub fn map_codex_usage_to_anthropic( u: &Option, web_search_requests: Option, @@ -1239,14 +1149,21 @@ mod tests { } #[test] - fn reduce_rate_limit_throws() { - let upstream = sse( - "codex.rate_limits", - json!({"rate_limits":{"limit_reached":true,"primary":{"reset_after_seconds":30}}}), + fn reduce_rate_limit_snapshot_is_progress_telemetry() { + let upstream = format!( + "{}{}", + sse( + "codex.rate_limits", + json!({"rate_limits":{"limit_reached":true,"primary":{"reset_after_seconds":30}}}), + ), + sse( + "response.completed", + json!({"response":{"id":"resp_1","status":"completed","usage":{}}}), + ) ); - let result = reduce_upstream_bytes(upstream.as_bytes()); - assert!(result.is_err()); - assert_eq!(result.unwrap_err().kind, UpstreamErrorKind::RateLimit); + let out = reduce_upstream_bytes(upstream.as_bytes()).unwrap(); + assert!(matches!(out.first(), Some(ReducerEvent::Progress))); + assert!(matches!(out.last(), Some(ReducerEvent::Finish { .. }))); } #[test] @@ -1392,24 +1309,14 @@ mod tests { } #[test] - fn reduce_incomplete_is_max_tokens() { + fn reduce_incomplete_is_upstream_error() { let upstream = sse( "response.incomplete", json!({"response":{"id":"resp_1","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"usage":{}}}), ); - let out = reduce_upstream_bytes(upstream.as_bytes()).unwrap(); - let last = out.last().unwrap(); - if let ReducerEvent::Finish { - stop_reason, - continuation_eligible, - .. - } = last - { - assert_eq!(*stop_reason, "max_tokens"); - assert!(!continuation_eligible); - } else { - panic!("expected Finish"); - } + let err = reduce_upstream_bytes(upstream.as_bytes()).unwrap_err(); + assert_eq!(err.kind, UpstreamErrorKind::Transient); + assert!(err.message.contains("max_output_tokens")); } #[test] diff --git a/src/providers/codex/websocket.rs b/src/providers/codex/websocket.rs index f0fa6f24..9e0d33fd 100644 --- a/src/providers/codex/websocket.rs +++ b/src/providers/codex/websocket.rs @@ -57,14 +57,6 @@ const WEBSOCKET_CONNECT_FORBIDDEN_COOLDOWN: Duration = Duration::from_secs(3); const WEBSOCKET_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30); const WEBSOCKET_KEEPALIVE_SEND_TIMEOUT: Duration = Duration::from_secs(10); -// Terminal WebSocket event types that signal the request is done -const TERMINAL_EVENTS: &[&str] = &[ - "response.completed", - "response.incomplete", - "response.failed", - "error", -]; - pub type CodexWebSocketEventReceiver = tokio::sync::mpsc::Receiver>; @@ -687,10 +679,7 @@ fn encode_sse(text: &str) -> Vec { // --------------------------------------------------------------------------- pub(super) fn is_terminal_event(payload: &serde_json::Value) -> bool { - match payload.get("type").and_then(|v| v.as_str()) { - Some(t) => TERMINAL_EVENTS.contains(&t), - None => false, - } + super::events::event_is_terminal(payload) } fn is_response_event(payload: &serde_json::Value) -> bool { @@ -3144,11 +3133,20 @@ mod tests { let completed = serde_json::json!({"type": "response.completed"}); assert!(is_terminal_event(&completed)); + let done = serde_json::json!({"type": "response.done"}); + assert!(is_terminal_event(&done)); + let delta = serde_json::json!({"type": "response.output_text.delta"}); assert!(!is_terminal_event(&delta)); let error = serde_json::json!({"type": "error", "error": {"message": "fail"}}); assert!(is_terminal_event(&error)); + + let response_error = serde_json::json!({ + "type": "response.error", + "response": {"error": {"message": "fail"}} + }); + assert!(is_terminal_event(&response_error)); } #[test] diff --git a/tests/codex_agent_continuation.rs b/tests/codex_agent_continuation.rs index 5a99c1a8..49c301a4 100644 --- a/tests/codex_agent_continuation.rs +++ b/tests/codex_agent_continuation.rs @@ -236,8 +236,16 @@ impl PendingRequest { self.outcome .send(MockOutcome::RawEvent { event: json!({ - "type": "codex.rate_limits", - "rate_limits": {"allowed": false, "limit_reached": true} + "type": "response.failed", + "response": { + "status": "failed", + "error": { + "status": 429, + "code": "rate_limit_exceeded", + "message": "rate limit exceeded", + "retry_after_seconds": 0 + } + } }), acknowledged, }) diff --git a/tests/smoke_cutover.rs b/tests/smoke_cutover.rs index 9f53bc5f..12f1622f 100644 --- a/tests/smoke_cutover.rs +++ b/tests/smoke_cutover.rs @@ -219,6 +219,91 @@ async fn spawn_truncated_http_upstream(body: &'static [u8]) -> String { format!("http://{addr}") } +async fn spawn_retrying_truncated_http_upstream( + first_body: Vec, + success_body: Vec, + attempts: Arc, +) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + tokio::spawn(async move { + for attempt in 0..2 { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + let mut request = [0_u8; 8192]; + let _ = stream.read(&mut request).await; + attempts.fetch_add(1, Ordering::SeqCst); + let body = if attempt == 0 { + first_body.as_slice() + } else { + success_body.as_slice() + }; + let content_length = if attempt == 0 { + body.len() + 4096 + } else { + body.len() + }; + let headers = format!( + "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ncontent-length: {content_length}\r\nconnection: close\r\n\r\n" + ); + let _ = stream.write_all(headers.as_bytes()).await; + let _ = stream.write_all(body).await; + let _ = stream.shutdown().await; + } + }); + + format!("http://{addr}") +} + +#[allow(clippy::await_holding_lock)] +async fn assert_codex_http_retries_structural_body_error(first_body: Vec) { + let _guard = env_lock(); + clear_all_continuations_for_tests(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let attempts = Arc::new(AtomicUsize::new(0)); + let success_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_retry\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_retry\"}}\n\n", + "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"retry succeeded\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"message\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_retry\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n" + ) + .as_bytes() + .to_vec(); + let upstream = + spawn_retrying_truncated_http_upstream(first_body, success_body, attempts.clone()).await; + + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream); + let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "http"); + let response = call_messages_body(json!({ + "model": "gpt-5.5", + "max_tokens": 64, + "stream": true, + "messages": [{"role":"user","content":"hello"}] + })) + .await; + + assert_eq!(response.status(), StatusCode::OK); + let body = tokio::time::timeout( + Duration::from_secs(2), + axum::body::to_bytes(response.into_body(), usize::MAX), + ) + .await + .expect("retried stream must finish") + .unwrap(); + let text = String::from_utf8_lossy(&body); + assert_eq!(attempts.load(Ordering::SeqCst), 2, "stream body: {text}"); + assert!(text.contains("retry succeeded"), "stream body: {text}"); + assert!(!text.contains("event: error"), "stream body: {text}"); + assert_eq!(text.matches("event: message_start").count(), 1); + assert_eq!(text.matches("event: message_stop").count(), 1); +} + #[allow(clippy::await_holding_lock)] async fn assert_codex_http_presemantic_retry(first_response: Vec) { let _guard = env_lock(); @@ -599,6 +684,23 @@ async fn spawn_websocket_close_then_retry_upstream(captured: Arc Date: Sun, 9 Aug 2026 00:32:30 +0500 Subject: [PATCH 2/2] fix(opencode): preserve max-token completions --- src/providers/codex/events.rs | 82 +++++++- src/providers/codex/translate/accumulate.rs | 44 +++- src/providers/codex/translate/live_stream.rs | 99 ++++++++- src/providers/codex/translate/mod.rs | 6 + src/providers/codex/translate/reducer.rs | 206 +++++++++++++++++-- src/providers/opencode/mod.rs | 103 +++++++++- src/providers/opencode/responses.rs | 17 +- 7 files changed, 522 insertions(+), 35 deletions(-) diff --git a/src/providers/codex/events.rs b/src/providers/codex/events.rs index 39f0def5..56b220ee 100644 --- a/src/providers/codex/events.rs +++ b/src/providers/codex/events.rs @@ -108,13 +108,35 @@ pub(crate) fn event_error(payload: &Value) -> Option<&Value> { }) } +pub(crate) fn response_is_incomplete_terminal(payload: &Value) -> bool { + let event_type = payload.get("type").and_then(Value::as_str); + let incomplete_details = payload.pointer("/response/incomplete_details"); + matches!( + event_type, + Some("response.completed" | "response.incomplete" | "response.done") + ) && (event_type == Some("response.incomplete") + || payload.pointer("/response/status").and_then(Value::as_str) == Some("incomplete") + || incomplete_details.is_some_and(|details| !details.is_null())) +} + +pub(crate) fn is_standard_max_output_tokens_incomplete(payload: &Value) -> bool { + let status = payload.pointer("/response/status"); + payload.get("type").and_then(Value::as_str) == Some("response.incomplete") + && (status.is_none() || status.and_then(Value::as_str) == Some("incomplete")) + && payload + .pointer("/response/incomplete_details/reason") + .and_then(Value::as_str) + == Some("max_output_tokens") + && event_error(payload).is_none() +} + pub(crate) fn classify_event_failure(payload: &Value) -> Option { let event_type = payload.get("type").and_then(Value::as_str)?; let response_status = payload.pointer("/response/status").and_then(Value::as_str); if event_type == "codex.rate_limits" { return None; } - if event_type == "response.incomplete" || response_status == Some("incomplete") { + if response_is_incomplete_terminal(payload) { let reason = payload .pointer("/response/incomplete_details/reason") .and_then(Value::as_str) @@ -489,4 +511,62 @@ mod tests { assert_eq!(failure.client_status(), 413); assert_eq!(failure.client_error_type(), "request_too_large"); } + + #[test] + fn incomplete_policy_only_accepts_consistent_max_output_terminal() { + let allowed = serde_json::json!({ + "type":"response.incomplete", + "response": { + "status":"incomplete", + "error":null, + "incomplete_details":{"reason":"max_output_tokens"} + } + }); + assert!(response_is_incomplete_terminal(&allowed)); + assert!(is_standard_max_output_tokens_incomplete(&allowed)); + let allowed_without_status = serde_json::json!({ + "type":"response.incomplete", + "response": { + "incomplete_details":{"reason":"max_output_tokens"} + } + }); + assert!(is_standard_max_output_tokens_incomplete( + &allowed_without_status + )); + + for rejected in [ + serde_json::json!({ + "type":"response.incomplete", + "response":{"status":"incomplete","incomplete_details":{"reason":"content_filter"}} + }), + serde_json::json!({ + "type":"response.incomplete", + "response":{"status":"incomplete","incomplete_details":{}} + }), + serde_json::json!({ + "type":"response.completed", + "response":{"status":"completed","incomplete_details":{"reason":"max_output_tokens"}} + }), + serde_json::json!({ + "type":"response.incomplete", + "response":{"status":null,"incomplete_details":{"reason":"max_output_tokens"}} + }), + serde_json::json!({ + "type":"response.incomplete", + "response":{"status":123,"incomplete_details":{"reason":"max_output_tokens"}} + }), + serde_json::json!({ + "type":"response.incomplete", + "response":{"status":"completed","incomplete_details":{"reason":"max_output_tokens"}} + }), + serde_json::json!({ + "type":"response.completed", + "response":{"status":"completed","incomplete_details":{}} + }), + ] { + assert!(response_is_incomplete_terminal(&rejected)); + assert!(!is_standard_max_output_tokens_incomplete(&rejected)); + assert!(classify_event_failure(&rejected).is_some()); + } + } } diff --git a/src/providers/codex/translate/accumulate.rs b/src/providers/codex/translate/accumulate.rs index f7c3f34b..7157460c 100644 --- a/src/providers/codex/translate/accumulate.rs +++ b/src/providers/codex/translate/accumulate.rs @@ -2,9 +2,10 @@ use serde_json::Value; use crate::traffic::TrafficCapture; +use super::IncompleteResponsePolicy; use super::reducer::{ AnthropicUsage, ReducerEvent, UpstreamStreamError, map_codex_usage_to_anthropic, - reduce_upstream_bytes, + reduce_upstream_bytes_with_policy, }; use super::web_search_compat::{WebSearchCompatContent, build_web_search_compat_blocks}; @@ -13,7 +14,13 @@ pub fn accumulate_response( message_id: &str, model: &str, ) -> Result { - accumulate_response_with_traffic(upstream, message_id, model, None) + accumulate_response_with_policy_and_traffic( + upstream, + message_id, + model, + IncompleteResponsePolicy::Error, + None, + ) } pub fn accumulate_response_with_traffic( @@ -22,7 +29,38 @@ pub fn accumulate_response_with_traffic( model: &str, traffic: Option<&TrafficCapture>, ) -> Result { - let events = match reduce_upstream_bytes(upstream) { + accumulate_response_with_policy_and_traffic( + upstream, + message_id, + model, + IncompleteResponsePolicy::Error, + traffic, + ) +} + +pub(crate) fn accumulate_response_with_policy( + upstream: &[u8], + message_id: &str, + model: &str, + incomplete_response_policy: IncompleteResponsePolicy, +) -> Result { + accumulate_response_with_policy_and_traffic( + upstream, + message_id, + model, + incomplete_response_policy, + None, + ) +} + +fn accumulate_response_with_policy_and_traffic( + upstream: &[u8], + message_id: &str, + model: &str, + incomplete_response_policy: IncompleteResponsePolicy, + traffic: Option<&TrafficCapture>, +) -> Result { + let events = match reduce_upstream_bytes_with_policy(upstream, incomplete_response_policy) { Ok(events) => events, Err(err) => { write_reducer_error_capture(traffic, &err); diff --git a/src/providers/codex/translate/live_stream.rs b/src/providers/codex/translate/live_stream.rs index 56b067a7..d8e868f9 100644 --- a/src/providers/codex/translate/live_stream.rs +++ b/src/providers/codex/translate/live_stream.rs @@ -1,12 +1,17 @@ use std::collections::HashMap; use crate::anthropic::sse::encode_sse_event; -use crate::providers::codex::events::classify_event_failure; +use crate::providers::codex::events::{ + classify_event_failure, is_standard_max_output_tokens_incomplete, +}; use crate::traffic::TrafficCapture; +use super::IncompleteResponsePolicy; use super::read_rewrite::sanitize_read_args; use super::reasoning_signature::{PendingReasoning, encode_reasoning_signature}; -use super::reducer::{CodexUsage, STOP_END_TURN, STOP_TOOL_USE, map_codex_usage_to_anthropic}; +use super::reducer::{ + CodexUsage, STOP_END_TURN, STOP_MAX_TOKENS, STOP_TOOL_USE, map_codex_usage_to_anthropic, +}; const BUFFERED_READ_REPAIR_TRAILING_WHITESPACE_BYTES: usize = 1_024; const BUFFERED_TOOL_MAX_ARGS_BYTES: usize = 5_000_000; @@ -65,6 +70,7 @@ pub struct LiveStreamTranslator { // Seeds Claude Code's live subagent counter until the provider returns // authoritative usage in the terminal message_delta. estimated_input_tokens: u64, + incomplete_response_policy: IncompleteResponsePolicy, finished: bool, } @@ -94,10 +100,19 @@ impl LiveStreamTranslator { deferred_text: Vec::new(), semantic_output_started: false, estimated_input_tokens, + incomplete_response_policy: IncompleteResponsePolicy::Error, finished: false, } } + pub(crate) fn with_incomplete_response_policy( + mut self, + policy: IncompleteResponsePolicy, + ) -> Self { + self.incomplete_response_policy = policy; + self + } + pub fn accept( &mut self, payload: &serde_json::Value, @@ -110,7 +125,10 @@ impl LiveStreamTranslator { let kind = payload.get("type").and_then(|v| v.as_str()).unwrap_or(""); let mut out = Vec::new(); - if let Some(failure) = classify_event_failure(payload) { + let allowed_incomplete = self.incomplete_response_policy + == IncompleteResponsePolicy::AllowMaxOutputTokens + && is_standard_max_output_tokens_incomplete(payload); + if !allowed_incomplete && let Some(failure) = classify_event_failure(payload) { return Err(failure.message); } @@ -163,7 +181,7 @@ impl LiveStreamTranslator { "response.output_item.done" => { self.output_item_done(payload, traffic, &mut out); } - "response.completed" | "response.done" => { + "response.completed" | "response.incomplete" | "response.done" => { self.finish(payload, traffic, &mut out); } _ => {} @@ -895,7 +913,12 @@ impl LiveStreamTranslator { self.emit_web_searches(traffic, out); self.ensure_message_start(traffic, out); let usage = payload.get("response").map(parse_codex_usage); - let stop_reason = if self.saw_tool_use { + let stop_reason = if self.incomplete_response_policy + == IncompleteResponsePolicy::AllowMaxOutputTokens + && is_standard_max_output_tokens_incomplete(payload) + { + STOP_MAX_TOKENS + } else if self.saw_tool_use { STOP_TOOL_USE } else { STOP_END_TURN @@ -1294,6 +1317,72 @@ mod tests { assert!(!translator.has_semantic_output()); } + #[test] + fn strict_live_translator_rejects_incomplete_response() { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5"); + let error = translator + .accept( + &json!({ + "type":"response.incomplete", + "response": { + "status":"incomplete", + "incomplete_details":{"reason":"max_output_tokens"} + } + }), + None, + ) + .unwrap_err(); + assert!(error.contains("max_output_tokens")); + } + + #[test] + fn standard_responses_policy_maps_max_output_tokens_to_message_stop() { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.6-luna") + .with_incomplete_response_policy(IncompleteResponsePolicy::AllowMaxOutputTokens); + let out = translator + .accept( + &json!({ + "type":"response.incomplete", + "response": { + "status":"incomplete", + "error":null, + "incomplete_details":{"reason":"max_output_tokens"}, + "usage":{"input_tokens":2,"output_tokens":8} + } + }), + None, + ) + .unwrap(); + let out = String::from_utf8(out).unwrap(); + assert!(out.contains(r#""stop_reason":"max_tokens""#)); + assert!(out.contains(r#""output_tokens":8"#)); + assert!(out.contains("event: message_stop")); + assert!(translator.is_finished()); + } + + #[test] + fn standard_responses_policy_rejects_other_incomplete_reasons() { + for reason in ["content_filter", "unknown"] { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.6-luna") + .with_incomplete_response_policy(IncompleteResponsePolicy::AllowMaxOutputTokens); + assert!( + translator + .accept( + &json!({ + "type":"response.incomplete", + "response": { + "status":"incomplete", + "incomplete_details":{"reason":reason} + } + }), + None, + ) + .is_err(), + "{reason}" + ); + } + } + #[test] fn structural_tool_start_is_not_semantic_until_arguments_arrive() { let mut tool = LiveStreamTranslator::new("msg_tool", "gpt-5.5"); diff --git a/src/providers/codex/translate/mod.rs b/src/providers/codex/translate/mod.rs index cdf6f423..08ec3490 100644 --- a/src/providers/codex/translate/mod.rs +++ b/src/providers/codex/translate/mod.rs @@ -7,3 +7,9 @@ pub mod reducer; pub mod request; pub mod stream; pub mod web_search_compat; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum IncompleteResponsePolicy { + Error, + AllowMaxOutputTokens, +} diff --git a/src/providers/codex/translate/reducer.rs b/src/providers/codex/translate/reducer.rs index 4212d441..b79102b2 100644 --- a/src/providers/codex/translate/reducer.rs +++ b/src/providers/codex/translate/reducer.rs @@ -1,6 +1,10 @@ use crate::anthropic::sse::parse_sse_events; -use crate::providers::codex::events::{CodexFailureKind, classify_event_failure}; +use crate::providers::codex::events::{ + CodexFailureKind, classify_event_failure, is_standard_max_output_tokens_incomplete, + response_is_incomplete_terminal, +}; +use super::IncompleteResponsePolicy; use super::read_rewrite::sanitize_read_args; use super::reasoning_signature::{PendingReasoning, ReasoningReplay, encode_reasoning_signature}; use super::request::ResponsesInputItem; @@ -245,6 +249,13 @@ pub fn finish_metadata_from_upstream( } pub fn reduce_upstream_bytes(input: &[u8]) -> Result, UpstreamStreamError> { + reduce_upstream_bytes_with_policy(input, IncompleteResponsePolicy::Error) +} + +pub(crate) fn reduce_upstream_bytes_with_policy( + input: &[u8], + incomplete_response_policy: IncompleteResponsePolicy, +) -> Result, UpstreamStreamError> { let sse_events = parse_sse_events(input); let mut out = Vec::new(); @@ -329,7 +340,32 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream event_count += 1; last_event_type = Some(t.clone()); - if let Some(failure) = classify_event_failure(&p) { + if _saw_terminal { + let message = if t == "response.completed" + || t == "response.incomplete" + || t == "response.done" + { + "upstream stream contained multiple terminal Codex response events" + } else { + "upstream stream contained a JSON event after the terminal Codex response event" + }; + return Err(UpstreamStreamError { + kind: UpstreamErrorKind::Transient, + message: message.to_string(), + retry_after_seconds: None, + diagnostics: Some(UpstreamStreamDiagnostics { + event_count, + last_event_type, + saw_terminal_event: true, + open_blocks: describe_open_blocks(&blocks_by_output_index), + }), + }); + } + + let allowed_incomplete = incomplete_response_policy + == IncompleteResponsePolicy::AllowMaxOutputTokens + && is_standard_max_output_tokens_incomplete(&p); + if !allowed_incomplete && let Some(failure) = classify_event_failure(&p) { let kind = match failure.kind { CodexFailureKind::RateLimit => UpstreamErrorKind::RateLimit, CodexFailureKind::Overloaded => UpstreamErrorKind::Overloaded, @@ -761,9 +797,7 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream .and_then(|v| v.as_str()) .map(|s| s.to_string()); final_usage = p.get("response").map(parse_codex_usage); - if response_is_incomplete(&p, &t) { - incomplete = true; - } + incomplete = response_is_incomplete_terminal(&p); continuation_eligible = (t == "response.completed" || t == "response.done") && !incomplete; continue; @@ -875,21 +909,6 @@ fn parse_codex_usage(response: &serde_json::Value) -> CodexUsage { } } -fn response_is_incomplete(payload: &serde_json::Value, event_type: &str) -> bool { - event_type == "response.incomplete" - || payload - .get("response") - .and_then(|r| r.get("status")) - .and_then(|v| v.as_str()) - == Some("incomplete") - || payload - .get("response") - .and_then(|r| r.get("incomplete_details")) - .and_then(|d| d.get("reason")) - .and_then(|v| v.as_str()) - .is_some() -} - fn should_buffer_tool_args(name: &str) -> bool { name == "Read" } @@ -1319,6 +1338,153 @@ mod tests { assert!(err.message.contains("max_output_tokens")); } + #[test] + fn standard_responses_policy_maps_max_output_tokens_to_finish() { + let upstream = sse( + "response.incomplete", + json!({"response":{"id":"resp_1","status":"incomplete","error":null,"incomplete_details":{"reason":"max_output_tokens"},"usage":{"input_tokens":2,"output_tokens":8}}}), + ); + let out = reduce_upstream_bytes_with_policy( + upstream.as_bytes(), + IncompleteResponsePolicy::AllowMaxOutputTokens, + ) + .unwrap(); + let Some(ReducerEvent::Finish { + stop_reason, + terminal_type, + continuation_eligible, + usage, + .. + }) = out.last() + else { + panic!("expected Finish"); + }; + assert_eq!(*stop_reason, STOP_MAX_TOKENS); + assert_eq!(terminal_type, TERM_INCOMPLETE); + assert!(!continuation_eligible); + assert_eq!( + usage.as_ref().and_then(|usage| usage.output_tokens), + Some(8) + ); + } + + #[test] + fn standard_responses_policy_rejects_a_second_terminal_event() { + let upstream = format!( + "{}{}", + sse( + "response.incomplete", + json!({"response":{"id":"resp_1","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"usage":{}}}), + ), + sse( + "response.completed", + json!({"response":{"id":"resp_1","status":"completed","usage":{}}}), + ), + ); + let err = reduce_upstream_bytes_with_policy( + upstream.as_bytes(), + IncompleteResponsePolicy::AllowMaxOutputTokens, + ) + .unwrap_err(); + assert_eq!(err.kind, UpstreamErrorKind::Transient); + assert!(err.message.contains("multiple terminal")); + } + + #[test] + fn standard_responses_policy_rejects_content_after_terminal() { + let trailing_events = [ + ( + "response.output_text.delta", + json!({"output_index":0,"delta":"late"}), + ), + ( + "response.output_item.added", + json!({ + "output_index":0, + "item":{"type":"message","id":"msg_late"} + }), + ), + ]; + + for (event_type, payload) in trailing_events { + let upstream = format!( + "{}{}", + sse( + "response.incomplete", + json!({"response":{"id":"resp_1","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"usage":{}}}), + ), + sse(event_type, payload), + ); + let err = reduce_upstream_bytes_with_policy( + upstream.as_bytes(), + IncompleteResponsePolicy::AllowMaxOutputTokens, + ) + .unwrap_err(); + + assert_eq!(err.kind, UpstreamErrorKind::Transient); + assert!(err.message.contains("after the terminal")); + let diagnostics = err.diagnostics.expect("expected diagnostics"); + assert_eq!(diagnostics.event_count, 2); + assert_eq!(diagnostics.last_event_type.as_deref(), Some(event_type)); + assert!(diagnostics.saw_terminal_event); + } + } + + #[test] + fn completed_terminal_followed_by_done_marker_remains_valid() { + let upstream = format!( + "{}data: [DONE]\n\n", + sse( + "response.completed", + json!({"response":{"id":"resp_1","status":"completed","usage":{}}}), + ), + ); + + let out = reduce_upstream_bytes(upstream.as_bytes()).unwrap(); + let Some(ReducerEvent::Finish { stop_reason, .. }) = out.last() else { + panic!("expected Finish"); + }; + assert_eq!(*stop_reason, STOP_END_TURN); + } + + #[test] + fn standard_responses_max_tokens_takes_priority_over_tool_use() { + let upstream = format!( + "{}{}{}{}", + sse( + "response.output_item.added", + json!({ + "output_index":0, + "item":{"type":"function_call","call_id":"call_1","name":"Read"} + }), + ), + sse( + "response.function_call_arguments.delta", + json!({"output_index":0,"delta":"{}"}), + ), + sse( + "response.output_item.done", + json!({ + "output_index":0, + "item":{"type":"function_call","call_id":"call_1","name":"Read","arguments":"{}"} + }), + ), + sse( + "response.incomplete", + json!({"response":{"id":"resp_1","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"usage":{}}}), + ), + ); + let out = reduce_upstream_bytes_with_policy( + upstream.as_bytes(), + IncompleteResponsePolicy::AllowMaxOutputTokens, + ) + .unwrap(); + let Some(ReducerEvent::Finish { stop_reason, .. }) = out.last() else { + panic!("expected Finish"); + }; + assert_eq!(*stop_reason, STOP_MAX_TOKENS); + } + #[test] fn reduce_completed_with_null_incomplete_details_is_end_turn() { let upstream = sse( diff --git a/src/providers/opencode/mod.rs b/src/providers/opencode/mod.rs index f9fc869b..cdc1b56d 100644 --- a/src/providers/opencode/mod.rs +++ b/src/providers/opencode/mod.rs @@ -22,10 +22,7 @@ use crate::provider::{ CliHandlers, Generation, GenerationBody, Provider, ProviderError, ProviderErrorKind, RequestContext, }; -use crate::providers::{ - codex::translate::accumulate::accumulate_response as accumulate_responses_response, - kimi::count_tokens, -}; +use crate::providers::kimi::count_tokens; use self::client::{OpenCodeClient, OpenCodeError}; use self::model::EndpointKind; @@ -149,7 +146,7 @@ impl OpenCodeProvider { Err(error) => return map_error(error), }; capture_buffered_upstream(&ctx, &bytes, "sse"); - match accumulate_responses_response(&bytes, &message_id, requested) { + match responses::accumulate_response(&bytes, &message_id, requested) { Ok(value) => value, Err(error) => return invalid_upstream_response(error), } @@ -553,6 +550,29 @@ mod tests { .into_response(); } if uri.path().ends_with("/responses") { + if body["max_output_tokens"] == 8 { + return ( + [(http::header::CONTENT_TYPE, "text/event-stream")], + concat!( + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_limited\"}}\n\n", + "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"truncated response\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"message\"}}\n\n", + "data: {\"type\":\"response.incomplete\",\"response\":{\"id\":\"resp_limited\",\"status\":\"incomplete\",\"error\":null,\"incomplete_details\":{\"reason\":\"max_output_tokens\"},\"usage\":{\"input_tokens\":6,\"output_tokens\":8}}}\n\n", + "data: [DONE]\n\n" + ), + ) + .into_response(); + } + if body["max_output_tokens"] == 9 { + return ( + [(http::header::CONTENT_TYPE, "text/event-stream")], + concat!( + "data: {\"type\":\"response.incomplete\",\"response\":{\"id\":\"resp_filtered\",\"status\":\"incomplete\",\"error\":null,\"incomplete_details\":{\"reason\":\"content_filter\"},\"usage\":{\"input_tokens\":6,\"output_tokens\":0}}}\n\n", + "data: [DONE]\n\n" + ), + ) + .into_response(); + } return ( [(http::header::CONTENT_TYPE, "text/event-stream")], concat!( @@ -687,6 +707,79 @@ mod tests { server.abort(); } + #[tokio::test] + async fn responses_max_output_tokens_is_a_normal_buffered_completion() { + let (provider, server) = mock_provider().await; + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "opencode-go/gpt-5.6-luna", + "stream": false, + "max_tokens": 8, + "messages": [{"role":"user","content":"write a long response"}] + })) + .unwrap(); + let response = provider.handle_messages(body, context()).await; + assert_eq!(response.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let value: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(value["content"][0]["text"], "truncated response"); + assert_eq!(value["stop_reason"], "max_tokens"); + assert_eq!(value["usage"]["output_tokens"], 8); + server.abort(); + } + + #[tokio::test] + async fn responses_max_output_tokens_is_a_normal_streaming_completion() { + let (provider, server) = mock_provider().await; + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "opencode-go/gpt-5.6-luna", + "max_tokens": 8, + "messages": [{"role":"user","content":"write a long response"}] + })) + .unwrap(); + let generation = provider + .generate_anthropic_stream(body, context()) + .await + .unwrap(); + let GenerationBody::LiveSse(body) = generation.body else { + panic!("OpenCode Go generation must remain live"); + }; + let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap(); + let output = String::from_utf8(bytes.to_vec()).unwrap(); + assert!(output.contains("truncated response")); + assert!(output.contains(r#""stop_reason":"max_tokens""#)); + assert!(output.contains("event: message_stop")); + assert!(!output.contains("event: error")); + server.abort(); + } + + #[tokio::test] + async fn responses_content_filter_is_not_misreported_as_max_tokens() { + let (provider, server) = mock_provider().await; + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "opencode-go/gpt-5.6-luna", + "stream": false, + "max_tokens": 9, + "messages": [{"role":"user","content":"hello"}] + })) + .unwrap(); + let response = provider.handle_messages(body, context()).await; + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let value: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(value["error"]["type"], "api_error"); + assert!( + value["error"]["message"] + .as_str() + .unwrap() + .contains("content_filter") + ); + server.abort(); + } + #[tokio::test] async fn streaming_contract_covers_all_three_upstream_protocols() { let (provider, server) = mock_provider().await; diff --git a/src/providers/opencode/responses.rs b/src/providers/opencode/responses.rs index 3cf84eae..e7f82e83 100644 --- a/src/providers/opencode/responses.rs +++ b/src/providers/opencode/responses.rs @@ -8,6 +8,7 @@ use futures_util::StreamExt; use crate::anthropic::schema::MessagesRequest; use crate::monitor::{MonitorHandle, usage_from_anthropic_sse}; use crate::providers::codex::translate::{ + IncompleteResponsePolicy, accumulate::accumulate_response_with_policy, live_stream::LiveStreamTranslator, request::translate_openai_compatible_request, }; use crate::providers::grok::translate::stream::SseDecoder; @@ -28,6 +29,19 @@ pub fn prepare_request( Ok(value) } +pub fn accumulate_response( + upstream: &[u8], + message_id: &str, + model: &str, +) -> anyhow::Result { + accumulate_response_with_policy( + upstream, + message_id, + model, + IncompleteResponsePolicy::AllowMaxOutputTokens, + ) +} + pub fn stream_body( upstream: OpenCodeResponse, message_id: String, @@ -39,7 +53,8 @@ pub fn stream_body( let state = ResponsesStreamState { upstream: upstream.into_stream(), decoder: SseDecoder::default(), - translator: LiveStreamTranslator::new(message_id, model), + translator: LiveStreamTranslator::new(message_id, model) + .with_incomplete_response_policy(IncompleteResponsePolicy::AllowMaxOutputTokens), terminal: false, error_sent: false, monitor,