fix(inference): stream + SSE-aggregate Codex OAuth Responses calls (#3201) (#3288)

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)

src/openhuman/inference/provider/compatible_parse.rs

@@ -286,3 +286,124 @@ pub(crate) fn extract_responses_text(response: ResponsesResponse) -> Option<Stri
 
     None
 }
+
+/// Aggregate an OpenAI Responses-API **SSE** body into the final assistant
+/// text (#3201).
+///
+/// The Codex/ChatGPT OAuth Responses endpoint
+/// (`https://chatgpt.com/backend-api/codex/responses`) rejects requests with
+/// `stream: false` (`Stream must be set to true`), so callers that target it
+/// must send `stream: true` and parse the resulting Server-Sent Event
+/// stream instead of the single JSON envelope `parse_responses_response_body`
+/// handles.
+///
+/// SSE shape (simplified — only the parts we depend on):
+///
+/// ```text
+/// event: response.output_text.delta
+/// data: {"type":"response.output_text.delta","delta":"Hello"}
+///
+/// event: response.output_text.delta
+/// data: {"type":"response.output_text.delta","delta":" world"}
+///
+/// event: response.completed
+/// data: {"type":"response.completed","response":{"output_text":"Hello world", ...}}
+///
+/// data: [DONE]
+/// ```
+///
+/// Strategy:
+///
+/// - Walk every `data: …` line (the `event:` line is informational; we route
+///   off the `type` field inside the JSON payload for resilience to the
+///   sentinel-style endings some providers emit).
+/// - `[DONE]` and empty data lines terminate the loop cleanly.
+/// - `response.output_text.delta` → push `delta` onto the accumulator.
+/// - `response.completed` → if we have a non-empty terminal
+///   `response.output_text`, prefer it (covers providers that batch the full
+///   text in the completion event and omit deltas).
+/// - Unrecognized `type` values are ignored — the spec is open-ended (tool
+///   calls, reasoning summaries, …) and we only need the assistant text here.
+///
+/// Returns the joined text on success. The error path returns the
+/// snippet-sanitised body just like [`parse_responses_response_body`] so a
+/// genuinely malformed stream is debuggable without leaking arbitrary chunk
+/// payloads.
+pub(crate) fn aggregate_responses_sse_body(
+    provider_name: &str,
+    body: &str,
+) -> anyhow::Result<String> {
+    let mut accumulated = String::new();
+    let mut terminal_text: Option<String> = 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}"
+    )
+}

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;