From 014b345a8d8b21bb864a41890af75da78940bef8 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sat, 11 Jul 2026 22:44:13 +0000 Subject: [PATCH] harness(openai): stream() falls back to JSON when upstream ignores stream:true MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Some OpenAI-compatible endpoints (and many test mocks) ignore stream:true and return a single non-streaming JSON body instead of an SSE event stream. The strict SSE parser then yields no items and a streaming turn stalls. stream() now checks the response content-type: when it is not text/event-stream, it reads the JSON body, parses it as one completion (applying prompt-guided tool recovery), and surfaces it as Started + one MessageDelta + Completed — matching the tolerant behaviour of hand-rolled clients so a non-streaming upstream still drives a streaming turn. Unblocks OpenHuman's crate-native turn cutover against its JSON-only e2e mocks (issue #4249 Phase 3). fmt + clippy clean. Claude-Session: https://claude.ai/code/session_01U8NDGbt9tKj443VquzycLB --- src/harness/providers/openai/transport.rs | 34 +++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/src/harness/providers/openai/transport.rs b/src/harness/providers/openai/transport.rs index 9e8d71a..9623d16 100644 --- a/src/harness/providers/openai/transport.rs +++ b/src/harness/providers/openai/transport.rs @@ -1157,6 +1157,40 @@ impl ChatModel for OpenAiModel { .post_json(&body, request.timeout_ms, true, "stream request") .await?; + // Leniency: some OpenAI-compatible endpoints (and test mocks) ignore + // `stream: true` and return a single non-streaming JSON body. Detect a + // non-`text/event-stream` content-type and parse it as one completion, + // surfaced as `Started` + one `MessageDelta` + `Completed` — matching the + // tolerant behaviour of hand-rolled OpenAI-compatible clients so a + // non-streaming upstream still drives a streaming turn. + let is_event_stream = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .map(|ct| ct.to_ascii_lowercase().contains("text/event-stream")) + .unwrap_or(false); + if !is_event_stream { + let text = response.text().await.map_err(|e| { + TinyAgentsError::Model(format!("openai non-stream stream-body read failed: {e}")) + })?; + let value: Value = serde_json::from_str(&text)?; + let mut parsed = parse_response(value)?; + if !self.profile.tool_calling && !request.tools.is_empty() { + parsed = prompt_tools::apply_to_response(parsed); + } + let delta = crate::harness::message::MessageDelta { + text: parsed.text(), + reasoning: String::new(), + tool_call: None, + }; + let items = vec![ + ModelStreamItem::Started, + ModelStreamItem::MessageDelta(delta), + ModelStreamItem::Completed(parsed), + ]; + return Ok(Box::pin(futures::stream::iter(items))); + } + // Forward each raw chunk as the `bytes::Bytes` buffer reqwest already // produced (a cheap refcount clone, no per-chunk copy); only the error // type is mapped onto the crate error. `SseState` is crate-internal,