diff --git a/src/openhuman/inference/provider/compatible.rs b/src/openhuman/inference/provider/compatible.rs index cadeff6c4..5133bfacc 100644 --- a/src/openhuman/inference/provider/compatible.rs +++ b/src/openhuman/inference/provider/compatible.rs @@ -30,8 +30,9 @@ use futures_util::{stream, StreamExt}; use compatible_dump::{dump_prompt_if_enabled, dump_response_if_enabled, reserve_dump_seq}; use compatible_parse::{ - build_responses_prompt, extract_responses_text, normalize_function_arguments, - parse_chat_response_body, parse_responses_response_body, parse_tool_calls_from_content_json, + aggregate_responses_sse_body, build_responses_prompt, extract_responses_text, + normalize_function_arguments, parse_chat_response_body, parse_responses_response_body, + parse_tool_calls_from_content_json, }; use compatible_stream::sse_bytes_to_chunks; use compatible_types::{ @@ -347,11 +348,45 @@ impl OpenAiCompatibleProvider { ); } + // #3201: the Codex/ChatGPT OAuth Responses endpoint + // (`https://chatgpt.com/backend-api/codex/responses`) rejects + // `stream: false` outright with `{"detail":"Stream must be set to + // true"}`. PR #3192 fixed the sibling `store: false` requirement; + // this branch lifts the same constraint for the stream flag and + // parses the resulting SSE body inline so the existing non-streaming + // call signature is preserved. Other Responses-API providers (real + // OpenAI, custom OpenAI-compatible) keep the single-envelope path — + // they accept `stream: false` and the SSE branch would be wasted + // work for them. + // + // Detection is keyed on the `/backend-api/codex` path segment, not + // the `chatgpt.com` host: the same path segment is what + // `OpenAiCodexRouting` substitutes when a user is signed in via + // OAuth (see `OPENAI_CODEX_BACKEND_BASE_URL`), and it's specific + // enough that no other OpenAI-compatible provider URL uses it. + // + // Parse the URL and inspect path segments rather than scanning the + // whole `base_url` so a proxy URL whose query string or fragment + // contains the literal `/backend-api/codex` (e.g. + // `.../v1?upstream=/backend-api/codex`) doesn't get falsely + // promoted into the SSE branch. + let is_codex_oauth_responses = reqwest::Url::parse(&self.base_url) + .ok() + .and_then(|url| { + let segments: Vec<&str> = url.path_segments()?.collect(); + Some( + segments + .windows(2) + .any(|window| window == ["backend-api", "codex"]), + ) + }) + .unwrap_or(false); + let request = ResponsesRequest { model: model.to_string(), input, instructions, - stream: Some(false), + stream: Some(is_codex_oauth_responses), store: Some(false), }; @@ -417,6 +452,12 @@ impl OpenAiCompatibleProvider { } let body = response.text().await?; + if is_codex_oauth_responses { + // SSE branch — `stream: true` always produces a Server-Sent + // Event body, even on the non-streaming wrapper. Aggregate it + // back into the same `String` shape the caller expects. + return aggregate_responses_sse_body(&self.name, &body); + } let responses = parse_responses_response_body(&self.name, &body)?; extract_responses_text(responses) diff --git a/src/openhuman/inference/provider/compatible_parse.rs b/src/openhuman/inference/provider/compatible_parse.rs index 089fca0f1..f3a41196b 100644 --- a/src/openhuman/inference/provider/compatible_parse.rs +++ b/src/openhuman/inference/provider/compatible_parse.rs @@ -286,3 +286,124 @@ pub(crate) fn extract_responses_text(response: ResponsesResponse) -> Option anyhow::Result { + let mut accumulated = String::new(); + let mut terminal_text: Option = None; + + for raw_line in body.split('\n') { + let line = raw_line.trim_end_matches('\r'); + let Some(data) = line.strip_prefix("data:") else { + continue; + }; + let data = data.trim(); + if data.is_empty() || data == "[DONE]" { + continue; + } + + let value: serde_json::Value = match serde_json::from_str(data) { + Ok(v) => v, + Err(error) => { + // Skip individual unparseable events rather than failing the + // whole turn — providers occasionally emit comments/keepalives + // shaped like `data: {ping}` that aren't strict JSON. + log::debug!( + "[providers][{provider_name}] Responses SSE: skipping unparseable event ({error})" + ); + continue; + } + }; + + let event_type = value.get("type").and_then(serde_json::Value::as_str); + match event_type { + Some("response.output_text.delta") => { + if let Some(delta) = value.get("delta").and_then(serde_json::Value::as_str) { + accumulated.push_str(delta); + } + } + Some("response.completed") => { + // Use the same "non-empty-after-trim" policy as + // `extract_responses_text` / `first_nonempty` so a + // whitespace-only terminal `output_text` doesn't override + // a non-empty accumulated delta stream and collapse a + // valid streamed reply to blank output. + terminal_text = value + .get("response") + .and_then(|response| response.get("output_text")) + .and_then(serde_json::Value::as_str) + .and_then(|text| first_nonempty(Some(text))); + } + // Treat error-shaped events as a hard failure so the caller + // surfaces the upstream reason instead of an empty completion. + Some("response.failed") | Some("response.error") | Some("error") => { + let snippet = compact_sanitized_body_snippet(data); + anyhow::bail!( + "{provider_name} Responses API stream reported a failure event: {snippet}" + ); + } + _ => {} + } + } + + // Prefer the terminal `response.output_text` when it carries a non-empty + // string — some providers batch full text in `response.completed` and + // skip per-token deltas, and others repeat what we accumulated. Either + // way the terminal text is the authoritative version on the wire. + // (`first_nonempty` in the match arm above already filtered whitespace.) + if let Some(text) = terminal_text { + return Ok(text); + } + if !accumulated.is_empty() { + return Ok(accumulated); + } + + let snippet = compact_sanitized_body_snippet(body); + anyhow::bail!( + "{provider_name} Responses API SSE stream produced no text events; body={snippet}" + ) +} diff --git a/src/openhuman/inference/provider/compatible_tests.rs b/src/openhuman/inference/provider/compatible_tests.rs index 1566fe934..5231c470a 100644 --- a/src/openhuman/inference/provider/compatible_tests.rs +++ b/src/openhuman/inference/provider/compatible_tests.rs @@ -277,6 +277,106 @@ fn parse_responses_response_body_reports_sanitized_snippet() { assert!(!msg.contains("sk-another-secret")); } +// ── aggregate_responses_sse_body (#3201) ───────────────────────────────────── + +/// Per-delta accumulation: the Codex/ChatGPT OAuth stream is a sequence of +/// `response.output_text.delta` events whose `delta` fields concatenate into +/// the final assistant text. +#[test] +fn aggregate_responses_sse_body_concatenates_text_deltas() { + let body = "data: {\"type\":\"response.output_text.delta\",\"delta\":\"hello \"}\n\n\ + data: {\"type\":\"response.output_text.delta\",\"delta\":\"world\"}\n\n\ + data: [DONE]\n"; + let text = + super::compatible_parse::aggregate_responses_sse_body("custom", body).expect("aggregate"); + assert_eq!(text, "hello world"); +} + +/// Some providers (and the Codex endpoint when the model batches its +/// reply) skip per-token deltas and emit the full text in +/// `response.completed.response.output_text`. The aggregator must fall +/// back to that terminal field when no deltas accumulated. +#[test] +fn aggregate_responses_sse_body_prefers_terminal_output_text_when_present() { + let body = "data: {\"type\":\"response.created\",\"response\":{}}\n\n\ + data: {\"type\":\"response.completed\",\"response\":{\"output_text\":\"batched final text\"}}\n\n\ + data: [DONE]\n"; + let text = + super::compatible_parse::aggregate_responses_sse_body("custom", body).expect("aggregate"); + assert_eq!(text, "batched final text"); +} + +/// #3201 CodeRabbit nit: a whitespace-only terminal `output_text` must +/// behave like the field is absent, so accumulated deltas survive instead +/// of being silently collapsed into blank output. Mirrors +/// `extract_responses_text`'s `first_nonempty(...)` policy. +#[test] +fn aggregate_responses_sse_body_ignores_whitespace_only_terminal_output_text() { + let body = "data: {\"type\":\"response.output_text.delta\",\"delta\":\"good \"}\n\n\ + data: {\"type\":\"response.output_text.delta\",\"delta\":\"reply\"}\n\n\ + data: {\"type\":\"response.completed\",\"response\":{\"output_text\":\" \\n\\t\"}}\n\n\ + data: [DONE]\n"; + let text = + super::compatible_parse::aggregate_responses_sse_body("custom", body).expect("aggregate"); + assert_eq!(text, "good reply"); +} + +/// Carriage-return line endings (CRLF, common in HTTP/1.1 SSE) parse the +/// same as LF-only — the trimming is just `\r` stripping. +#[test] +fn aggregate_responses_sse_body_tolerates_crlf_line_endings() { + let body = "data: {\"type\":\"response.output_text.delta\",\"delta\":\"crlf\"}\r\n\r\n\ + data: [DONE]\r\n"; + let text = + super::compatible_parse::aggregate_responses_sse_body("custom", body).expect("aggregate"); + assert_eq!(text, "crlf"); +} + +/// `response.failed` / `response.error` / `error` event shapes are +/// terminal failures — bubble them up so the caller surfaces the upstream +/// reason instead of returning empty text. +#[test] +fn aggregate_responses_sse_body_surfaces_failure_events() { + let body = "data: {\"type\":\"response.created\",\"response\":{}}\n\n\ + data: {\"type\":\"response.failed\",\"error\":\"upstream model unavailable\"}\n\n"; + let err = super::compatible_parse::aggregate_responses_sse_body("custom", body) + .expect_err("failure event should propagate"); + assert!( + err.to_string() + .contains("custom Responses API stream reported a failure event"), + "unexpected error: {err}" + ); +} + +/// A stream that produced no usable text events returns a sanitised +/// "no text events" error so the caller sees something actionable +/// instead of an empty string. +#[test] +fn aggregate_responses_sse_body_errors_when_no_text_events_present() { + let body = "data: {\"type\":\"response.created\",\"response\":{}}\n\n\ + data: [DONE]\n"; + let err = super::compatible_parse::aggregate_responses_sse_body("custom", body) + .expect_err("empty stream should fail"); + assert!( + err.to_string() + .contains("custom Responses API SSE stream produced no text events"), + "unexpected error: {err}" + ); +} + +/// Malformed individual events (provider keepalive comments, etc.) must +/// not abort the whole turn — they're skipped and the good deltas still +/// aggregate. +#[test] +fn aggregate_responses_sse_body_skips_unparseable_events() { + let body = "data: {malformed-keepalive\n\n\ + data: {\"type\":\"response.output_text.delta\",\"delta\":\"good\"}\n\n\ + data: [DONE]\n"; + let text = + super::compatible_parse::aggregate_responses_sse_body("custom", body).expect("aggregate"); + assert_eq!(text, "good"); +} + #[test] fn x_api_key_auth_style() { let p = OpenAiCompatibleProvider::new( @@ -396,6 +496,13 @@ fn extra_query_params_are_applied_to_codex_urls() { ); } +/// #3201: the Codex/ChatGPT OAuth Responses endpoint rejects +/// `stream: false` with `{"detail":"Stream must be set to true"}` and +/// only emits SSE bodies. The non-streaming `chat_via_responses` wrapper +/// must therefore (a) flip the `stream` flag for `/backend-api/codex` +/// URLs and (b) aggregate the SSE body back into the same `String` +/// the caller expects. PR #3192 fixed the sibling `store: false` +/// requirement; this test pins both wire-shape requirements together. #[tokio::test] async fn responses_api_primary_posts_directly_to_responses() { let server = MockServer::start().await; @@ -407,13 +514,16 @@ async fn responses_api_primary_posts_directly_to_responses() { "role": "user", "content": [{"type": "input_text", "text": "hello"}] }], - "stream": false, + "stream": true, "store": false }))) - .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ - "output_text": "hello from responses", - "output": [] - }))) + .respond_with(ResponseTemplate::new(200).set_body_string( + "data: {\"type\":\"response.output_text.delta\",\"delta\":\"hello \"}\n\n\ + data: {\"type\":\"response.output_text.delta\",\"delta\":\"from \"}\n\n\ + data: {\"type\":\"response.output_text.delta\",\"delta\":\"responses\"}\n\n\ + data: {\"type\":\"response.completed\",\"response\":{\"output_text\":\"hello from responses\"}}\n\n\ + data: [DONE]\n\n", + )) .mount(&server) .await;