diff --git a/crates/tinyinference/src/providers/openai/test.rs b/crates/tinyinference/src/providers/openai/test.rs index 921efeb..822bdf1 100644 --- a/crates/tinyinference/src/providers/openai/test.rs +++ b/crates/tinyinference/src/providers/openai/test.rs @@ -2222,3 +2222,41 @@ fn degrade_for_400_unions_with_existing_baseline_degrade() { }) ); } + +/// The Codex OAuth backend ends a streamed Responses call with a +/// `response.completed` whose `output` array is **empty** and delivers the actual +/// content in earlier `response.output_item.done` events. Reading the terminal +/// event alone therefore yields a well-formed response with no text — which is +/// exactly what surfaced to users as "The model returned an empty response". +#[test] +fn responses_sse_fold_grafts_streamed_output_items_onto_the_completed_response() { + let body = concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"status\":\"in_progress\",\"output\":[]}}\n\n", + "event: response.output_item.done\n", + "data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"hello\"}]}}\n\n", + "event: response.completed\n", + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"output\":[]}}\n\n", + ); + + let value = super::transport::responses_sse_final_value(body).expect("a final response"); + assert_eq!( + value.get("status").and_then(|status| status.as_str()), + Some("completed"), + "the terminal event wins over the earlier in-progress one" + ); + let response = super::responses::parse_responses_response(value); + assert_eq!(response.text(), "hello"); +} + +/// A body that never reaches a terminal event still yields the last response it +/// saw, rather than failing the whole call. +#[test] +fn responses_sse_fold_falls_back_to_the_last_seen_response() { + let body = "data: {\"type\":\"response.created\",\"response\":{\"status\":\"in_progress\",\"output\":[]}}\n"; + let value = super::transport::responses_sse_final_value(body).expect("a fallback response"); + assert_eq!( + value.get("status").and_then(|status| status.as_str()), + Some("in_progress") + ); +} diff --git a/crates/tinyinference/src/providers/openai/transport.rs b/crates/tinyinference/src/providers/openai/transport.rs index 492b55d..9342756 100644 --- a/crates/tinyinference/src/providers/openai/transport.rs +++ b/crates/tinyinference/src/providers/openai/transport.rs @@ -111,6 +111,13 @@ pub struct OpenAiModel { keep_alive: Option, json_schema_strict: AtomicBool, native_tools_on_wire: AtomicBool, + /// Whether the Responses endpoint requires `stream: true` (the ChatGPT Codex + /// OAuth backend does: it answers a unary POST with HTTP 400 + /// `{"detail":"Stream must be set to true"}` and only ever speaks SSE). + /// Latched on the first such 400 so exactly one call pays the probe; every + /// later request sends `stream: true` up front and the SSE body is folded + /// back into one `ModelResponse`. + responses_requires_stream: AtomicBool, } impl std::fmt::Debug for OpenAiModel { @@ -364,6 +371,7 @@ impl OpenAiModel { keep_alive: None, json_schema_strict: AtomicBool::new(true), native_tools_on_wire: AtomicBool::new(true), + responses_requires_stream: AtomicBool::new(false), } } @@ -1248,7 +1256,10 @@ impl OpenAiModel { model: model.clone(), input, instructions, - stream: None, + stream: self + .responses_requires_stream + .load(Ordering::Relaxed) + .then_some(true), store: Some(false), max_output_tokens, tools, @@ -1295,13 +1306,58 @@ impl OpenAiModel { self.send_responses(&retry, request.timeout_ms, &url) .await? } + // The ChatGPT Codex OAuth backend refuses a unary Responses call + // outright (`{"detail":"Stream must be set to true"}`) — it only + // speaks SSE. Latch that fact so later calls send `stream: true` + // up front, and retry this one immediately; the SSE body is folded + // back into a single `ModelResponse` below, so the caller still + // sees an ordinary unary result. + Err(Error::Provider(err)) + if err.status == Some(400) + && body.stream.is_none() + && err + .message + .to_ascii_lowercase() + .contains("stream must be set") => + { + self.responses_requires_stream + .store(true, Ordering::Relaxed); + let retry = responses::ResponsesRequest { + stream: Some(true), + ..body + }; + self.send_responses(&retry, request.timeout_ms, &url) + .await? + } Err(e) => return Err(e), }; + let is_event_stream = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(|content_type| { + content_type + .to_ascii_lowercase() + .contains("text/event-stream") + }) + .unwrap_or(false); let text = response .text() .await .map_err(|e| Error::Model(format!("openai responses body read failed: {e}")))?; - let value: Value = serde_json::from_str(&text)?; + // Content-type is a hint, not a contract here: the Codex backend has been + // observed answering a streamed Responses call without an + // `text/event-stream` content-type, and a plain JSON body parses on the + // first branch anyway. So try JSON first and fold SSE whenever that + // fails (or whenever the header did say SSE). + let value: Value = match (is_event_stream, serde_json::from_str::(&text)) { + (false, Ok(value)) => value, + _ => responses_sse_final_value(&text).ok_or_else(|| { + Error::Model( + "openai responses stream carried no `response.completed` event".to_string(), + ) + })?, + }; Ok(responses::parse_responses_response(value)) } @@ -1811,3 +1867,69 @@ impl ChatModel for OpenAiModel { Ok(Box::pin(stream)) } } + +/// Fold a Responses-API SSE body into the single final `response` object. +/// +/// The Codex backend streams `data:`-prefixed JSON events and carries the +/// complete result on `response.completed`. Earlier events (`response.created`, +/// per-delta events) are partial, so the completed one is preferred; a body that +/// ends without it falls back to the last event that carried a `response` +/// object, then to the last event that itself looks like a response (`output` +/// present), so a backend that streams a bare final object still parses. +pub(super) fn responses_sse_final_value(body: &str) -> Option { + let mut fallback: Option = None; + let mut final_response: Option = None; + // Codex streams each finished output item as its own `response.output_item.done` + // event and then sends a `response.completed` whose `output` array is EMPTY — + // so the terminal event alone carries usage and status but no text. Collect the + // items as they stream and graft them onto the final response. + let mut items: Vec = Vec::new(); + for line in body.lines() { + let payload = match line.strip_prefix("data:") { + Some(payload) => payload.trim(), + None => continue, + }; + if payload.is_empty() || payload == "[DONE]" { + continue; + } + let Ok(event) = serde_json::from_str::(payload) else { + continue; + }; + let kind = event + .get("type") + .and_then(Value::as_str) + .unwrap_or_default(); + match kind { + "response.output_item.done" => { + if let Some(item) = event.get("item") { + items.push(item.clone()); + } + } + "response.completed" | "response.incomplete" => { + if let Some(response) = event.get("response") { + final_response = Some(response.clone()); + } + } + _ => { + if let Some(response) = event.get("response") { + fallback = Some(response.clone()); + } else if event.get("output").is_some() { + fallback = Some(event); + } + } + } + } + let mut response = final_response.or(fallback)?; + let output_is_empty = response + .get("output") + .and_then(Value::as_array) + .map(|output| output.is_empty()) + .unwrap_or(true); + if let Some(object) = response + .as_object_mut() + .filter(|_| output_is_empty && !items.is_empty()) + { + object.insert("output".to_string(), Value::Array(items)); + } + Some(response) +}