From 2ae0d29a7b7ace1f431ed7d2b616bd57b83df605 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Thu, 30 Jul 2026 18:43:33 +0530 Subject: [PATCH] fix(openai): latch the stream-only constraint and widen its detection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OpenAI-compatible proxies exist that refuse unary calls outright, answering a `stream: false` body with an HTTP 400 `{"detail":"Stream must be set to true"}`. `ChatModel::invoke` already falls back to the streaming path on that rejection, but the fallback has two gaps that keep it from actually converging. **The constraint is never remembered.** `requires_streaming` is a plain `bool` that only `with_requires_streaming` can set, and nothing calls that builder — `invoke` takes `&self`, so the auto-detected discovery had nowhere to go. Every single unary call therefore re-pays a guaranteed-400 round trip before falling back: one wasted request per agent turn, forever. Make it an `AtomicBool` and latch it on first discovery, so only the first call pays and the rest go straight to the streaming path. `latch_stream_required` logs on the transition only, so the discovery is visible exactly once per process. **The detection is one case-sensitive literal.** The trigger was `err.message.contains("Stream must be set to true")`, so any other phrasing — lower-case, `'stream' must be true`, `streaming is required`, `only streaming is supported`, `stream=true is required` — still hard-failed the harness run. Replace it with `is_stream_required_error`, which matches that family case-insensitively and searches the raw body as well as the decoded message: `parse_error_body` only promotes `error.message` / `message`, so a reason nested under any other key (`detail`, `errors[].reason`) was invisible. `422` joins `400` because the wording arrives in a FastAPI `{"detail": …}` envelope and FastAPI's own field-validation status is 422. A `stream` mention is required alongside the "must be true" phrasing so unrelated 400s never get rerouted — the `degrade_for_400` `tool_choice` / `response_format` handlers and the `/responses` `max_output_tokens` retry keep their bodies to themselves. `requires_streaming()` is now `pub` (it was `#[cfg(test)]`-only) so hosts can observe a learned constraint. Tests: latch set/idempotence, latch visibility through a shared `Arc` handle, an 8-body wording matrix including the exact reported body and one nested under a non-standard key, and a negative matrix (wrong status, `tool_choice`, `max_output_tokens`, `stream_options is not supported`, `'store' must be true`). Known gap, deliberately out of scope: the `/responses` path (`responses_api_primary`) returns before this check and sends `stream: None`, so a streaming-only Responses backend still surfaces the raw 400. Folding Responses SSE is a separate piece of work — the transport notes it as a follow-up. Refs openhuman#5165. --- src/harness/providers/openai/mod.rs | 11 +- src/harness/providers/openai/test.rs | 105 ++++++++++++++ src/harness/providers/openai/transport.rs | 163 ++++++++++++++++++---- 3 files changed, 251 insertions(+), 28 deletions(-) diff --git a/src/harness/providers/openai/mod.rs b/src/harness/providers/openai/mod.rs index 700ccae..1328d4b 100644 --- a/src/harness/providers/openai/mod.rs +++ b/src/harness/providers/openai/mod.rs @@ -26,6 +26,15 @@ //! or automatically as a single retry when a 400 body implicates the shape. See //! the module `README.md` "Local-server compatibility" section. //! +//! Some OpenAI-compatible proxies go further and refuse unary calls entirely, +//! answering `stream: false` with an HTTP 400/422 such as +//! `{"detail":"Stream must be set to true"}`. [`ChatModel::invoke`] recognises +//! that family of rejections (`is_stream_required_error` in `transport`), folds +//! the SSE stream into a single [`ModelResponse`] instead, and **latches** the +//! constraint on the instance so only the first call pays the rejected round +//! trip. Declare it up front with +//! [`OpenAiModel::with_requires_streaming`] to skip even that one. +//! //! # Example //! //! ```no_run @@ -90,7 +99,7 @@ use sse::*; #[cfg(test)] use transport::{ Degrade, auth_headers, degrade_for_400, effective_temperature, glob_match, - merge_provider_options, merge_system_into_user, request_timeout, + is_stream_required_error, merge_provider_options, merge_system_into_user, request_timeout, }; #[cfg(test)] diff --git a/src/harness/providers/openai/test.rs b/src/harness/providers/openai/test.rs index bbe1980..2ecf0d7 100644 --- a/src/harness/providers/openai/test.rs +++ b/src/harness/providers/openai/test.rs @@ -678,6 +678,111 @@ fn requires_streaming_flag_skips_non_streaming_attempt() { ); } +// openhuman#5165: the streaming-only constraint used to be re-discovered on +// every unary call, because nothing recorded it. `ChatModel::invoke` takes +// `&self`, so the latch has to be interior-mutable — a plain `bool` field could +// only ever be set by the builder, which nothing calls in production. Without +// the latch a streaming-only proxy costs one guaranteed-400 round trip per +// agent turn forever (511 events / 2 users on the linked Sentry issue). +#[test] +fn stream_required_constraint_latches_after_discovery() { + let m = model(); + assert!(!m.requires_streaming(), "must start un-latched"); + + m.latch_stream_required(); + assert!( + m.requires_streaming(), + "a discovered constraint must be remembered so later calls skip \ + the doomed non-streaming attempt" + ); + + // Idempotent: latching again is a no-op (the log fires only on transition). + m.latch_stream_required(); + assert!(m.requires_streaming()); +} + +#[test] +fn stream_required_latch_survives_through_a_shared_handle() { + // Production holds models as `Arc`, so the latch must be + // observable through a shared reference — that is the whole reason it is an + // `AtomicBool` and not a `bool`. + let shared: std::sync::Arc = std::sync::Arc::new(model()); + let clone = std::sync::Arc::clone(&shared); + assert!(!clone.requires_streaming()); + + shared.latch_stream_required(); + assert!( + clone.requires_streaming(), + "the latch must be visible to every holder of the shared model" + ); +} + +// The trigger used to be a single case-sensitive `contains("Stream must be set +// to true")`. OpenAI-compatible proxies do not standardise this wording, so any +// other phrasing hard-failed the run instead of falling back to streaming. +#[test] +fn is_stream_required_error_matches_the_known_proxy_wordings() { + let m = model(); + + for body in [ + // The exact body from the openhuman#5165 Sentry report (FastAPI-style + // `detail` envelope — note `parse_error_body` cannot promote it to + // `message`, so the raw-body scan is what catches nesting). + r#"{"detail":"Stream must be set to true"}"#, + // Lower-case, and quoted field name. + r#"{"detail":"stream must be set to true"}"#, + r#"{"error":{"message":"'stream' must be true for this endpoint"}}"#, + // Other phrasings seen from OpenAI-compatible gateways. + r#"{"error":{"message":"streaming is required for this model"}}"#, + r#"{"error":{"message":"Only streaming responses are supported"}}"#, + r#"{"error":{"message":"this deployment requires streaming"}}"#, + r#"{"error":{"message":"stream=true is required"}}"#, + // Nested under a non-standard key alongside a generic top-level message. + r#"{"message":"Bad Request","errors":[{"reason":"stream must be true"}]}"#, + ] { + let err = m.parse_error_body(400, body); + assert!( + is_stream_required_error(&err), + "must detect the streaming requirement in: {body}" + ); + } + + // FastAPI validates request fields with 422, so accept that status too. + let unprocessable = m.parse_error_body(422, r#"{"detail":"Stream must be set to true"}"#); + assert!( + is_stream_required_error(&unprocessable), + "422 with the streaming wording must also trigger the fallback" + ); +} + +#[test] +fn is_stream_required_error_ignores_unrelated_failures() { + let m = model(); + + // Wrong status: the same wording on a 500 is a server fault, not a + // request-shape constraint, and must stay retryable rather than be + // rerouted into the streaming path. + let server_error = m.parse_error_body(500, r#"{"detail":"Stream must be set to true"}"#); + assert!(!is_stream_required_error(&server_error)); + + // Right status, unrelated wording — must not hijack the other 400 handlers + // (`degrade_for_400`, the `/responses` max_output_tokens retry). + for body in [ + r#"{"error":{"message":"Invalid value for 'tool_choice'"}}"#, + r#"{"error":{"message":"max_output_tokens is not supported"}}"#, + // Mentions stream but states no requirement. + r#"{"error":{"message":"stream_options is not supported by this model"}}"#, + // States a requirement but about a different field. + r#"{"error":{"message":"'store' must be true"}}"#, + ] { + let err = m.parse_error_body(400, body); + assert!( + !is_stream_required_error(&err), + "must NOT treat this as a streaming requirement: {body}" + ); + } +} + #[test] fn reasoning_tag_extraction_defaults_off_for_hosted_openai_only() { // Hosted OpenAI never emits inline `` reasoning; unconditional diff --git a/src/harness/providers/openai/transport.rs b/src/harness/providers/openai/transport.rs index 1272ef3..d5fe94c 100644 --- a/src/harness/providers/openai/transport.rs +++ b/src/harness/providers/openai/transport.rs @@ -7,6 +7,8 @@ use super::responses; use super::*; +use std::sync::atomic::{AtomicBool, Ordering}; + use crate::harness::model::StreamAccumulator; /// How the provider expects the API credential to be sent on each request. @@ -111,11 +113,22 @@ pub struct OpenAiModel { /// inline `` reasoning, and unconditional extraction would silently /// strip legitimate content that mentions a literal `` tag. reasoning_tags_overridden: bool, - /// When `true`, the unary [`ChatModel::invoke`] path sends `stream: true` - /// on the wire and folds the server-sent-events stream into a single + /// When set, the unary [`ChatModel::invoke`] path sends `stream: true` on + /// the wire and folds the server-sent-events stream into a single /// [`ModelResponse`] internally — for providers that reject a - /// `stream: false` body with an HTTP 400. `false` by default. - requires_streaming: bool, + /// `stream: false` body (`{"detail":"Stream must be set to true"}` and + /// friends). Starts `false`; seeded by + /// [`with_requires_streaming`](Self::with_requires_streaming) and + /// **latched at run time** the first time the provider rejects a + /// non-streaming request. + /// + /// It is an [`AtomicBool`] rather than a plain `bool` because + /// [`ChatModel::invoke`] takes `&self` (models are shared behind + /// `Arc`): without interior mutability the discovery could + /// not be remembered, so every single call would keep paying a + /// guaranteed-400 round trip before falling back. See + /// [`Self::latch_stream_required`]. + stream_required: AtomicBool, } /// The auth headers `(name, value)` for a given [`AuthStyle`] + credential. @@ -342,21 +355,45 @@ impl OpenAiModel { // `` and must not strip literal mentions of the tag. reasoning_tags: Some(ReasoningTagExtraction::default()), reasoning_tags_overridden: false, - requires_streaming: false, + stream_required: AtomicBool::new(false), } } /// When the provider requires `stream: true` for every request, routes /// unary [`invoke`](ChatModel::invoke) through the streaming path internally. - pub fn with_requires_streaming(mut self, enabled: bool) -> Self { - self.requires_streaming = enabled; + /// + /// Optional: the transport also discovers this on its own (see + /// [`Self::latch_stream_required`]). Set it explicitly to skip the one + /// exploratory request that discovery costs. + pub fn with_requires_streaming(self, enabled: bool) -> Self { + self.stream_required.store(enabled, Ordering::Relaxed); self } - /// Returns whether the provider requires streaming for all calls. - #[cfg(test)] - pub(super) fn requires_streaming(&self) -> bool { - self.requires_streaming + /// Returns whether this instance currently requires streaming for all + /// unary calls — either declared up front via + /// [`with_requires_streaming`](Self::with_requires_streaming) or learned + /// from a provider rejection. + pub fn requires_streaming(&self) -> bool { + self.stream_required.load(Ordering::Relaxed) + } + + /// Remembers that this endpoint only accepts `stream: true`. + /// + /// Called once the provider has told us so with an HTTP 400/422 (see + /// [`is_stream_required_error`]). Without this latch the transport + /// re-discovers the constraint on **every** unary call, so a streaming-only + /// proxy costs one guaranteed-400 round trip per agent turn — the shape + /// behind the 511 events / 2 users on openhuman#5165. Logs on the + /// transition only, so the discovery is visible exactly once per process. + pub(super) fn latch_stream_required(&self) { + if !self.stream_required.swap(true, Ordering::Relaxed) { + tracing::info!( + provider = %self.provider, + model = %self.model, + "[openai] provider requires stream:true; latching for subsequent calls" + ); + } } /// Routes calls to the OpenAI **Responses API** (`/v1/responses`) instead of @@ -1325,6 +1362,73 @@ pub(super) struct Degrade { pub json_object: bool, } +/// Statuses an OpenAI-compatible proxy uses to reject a non-streaming request. +/// +/// `400` is what the reference implementation returns. `422` is included because +/// the wording seen in the wild arrives in a FastAPI `{"detail": …}` envelope, +/// and FastAPI's own rejection status is `422` — a proxy that validates `stream` +/// as a request field rather than raising an explicit `HTTPException(400)` lands +/// there. The stream-specific wording still has to match, so widening the status +/// set does not widen the false-positive surface meaningfully. +const STREAM_REQUIRED_STATUSES: [u16; 2] = [400, 422]; + +/// Recognises "this endpoint only accepts `stream: true`" from a provider error. +/// +/// Some OpenAI-compatible proxies refuse unary calls outright. The wording is +/// not standardised, so match a family of phrasings case-insensitively rather +/// than one literal string: +/// +/// - `{"detail":"Stream must be set to true"}` — the openhuman#5165 report +/// - `stream must be true` / `'stream' must be set to true` +/// - `streaming is required` / `only streaming is supported` +/// - `stream=true is required` +/// +/// Both the decoded `message` **and** the raw body are searched, because +/// [`OpenAiModel::parse_error_body`] only promotes `error.message` / `message` +/// to `ProviderError::message`. A proxy that nests the reason under any other +/// key (`detail`, `errors[0].reason`, …) alongside a generic top-level +/// `message` would otherwise be invisible to the check. +/// +/// The `stream` mention is required in addition to the "must be true" phrasing +/// so unrelated 400s ("`max_output_tokens` must be true"-shaped nonsense, +/// `tool_choice` rejections) never route a call into the streaming path. +/// +/// Pure, so the detection policy is unit-testable without a network call. +pub(super) fn is_stream_required_error(error: &ProviderError) -> bool { + if !error + .status + .is_some_and(|status| STREAM_REQUIRED_STATUSES.contains(&status)) + { + return false; + } + if mentions_stream_required(&error.message) { + return true; + } + error + .raw + .as_ref() + .is_some_and(|raw| mentions_stream_required(&raw.to_string())) +} + +/// The wording half of [`is_stream_required_error`], split out so both the +/// decoded message and the raw body run through identical rules. +fn mentions_stream_required(haystack: &str) -> bool { + let lower = haystack.to_ascii_lowercase(); + if !lower.contains("stream") { + return false; + } + const PHRASES: [&str; 7] = [ + "must be set to true", + "must be true", + "streaming is required", + "requires streaming", + "only streaming", + "stream=true", + "stream: true is required", + ]; + PHRASES.iter().any(|phrase| lower.contains(phrase)) +} + /// Computes the additional degradation to apply after an HTTP 400, or `None` /// when the failure is not an auto-degradable request-shape rejection. /// @@ -1449,10 +1553,12 @@ impl ChatModel for OpenAiModel { /// Invokes the OpenAI Chat Completions endpoint and maps the response into a /// [`ModelResponse`]. /// - /// When the provider requires streaming - /// ([`Self::requires_streaming`] or a 400 with `"Stream must be set to - /// true"`), the call is degraded to the streaming path internally and the - /// SSE stream is folded into a single response. + /// When the provider requires streaming — declared via + /// [`OpenAiModel::with_requires_streaming`], or discovered from a rejection + /// matching [`is_stream_required_error`] — the call is degraded to the + /// streaming path internally and the SSE stream is folded into a single + /// response. A discovered constraint is latched on the instance, so only the + /// first call pays the rejected round trip. /// /// # Errors /// @@ -1464,13 +1570,15 @@ impl ChatModel for OpenAiModel { return self.invoke_responses(&request).await; } - // Short-circuit: providers that require `stream: true` on every call - // skip the non-streaming attempt entirely to avoid a guaranteed 400. - if self.requires_streaming { - tracing::info!( - provider = self.provider, + // Short-circuit: providers known to require `stream: true` on every + // call skip the non-streaming attempt entirely to avoid a guaranteed + // 400. "Known" covers both an explicit `with_requires_streaming(true)` + // and a constraint latched from an earlier rejection on this instance. + if self.requires_streaming() { + tracing::debug!( + provider = %self.provider, model = %self.model, - "[openai] requires_streaming is set; folding stream into unary response" + "[openai] stream:true required; folding stream into unary response" ); return invoke_with_streaming(self, request).await; } @@ -1480,15 +1588,16 @@ impl ChatModel for OpenAiModel { .await { Ok(response) => response, - Err(TinyAgentsError::Provider(err)) - if err.status == Some(400) - && err.message.contains("Stream must be set to true") => - { - // The provider rejects `stream: false`. Fall back to the - // streaming path and fold the SSE stream into a response. + Err(TinyAgentsError::Provider(err)) if is_stream_required_error(&err) => { + // The provider rejects `stream: false`. Remember it so the next + // call goes straight to streaming instead of re-paying this + // failed round trip, then fall back to the streaming path and + // fold the SSE stream into a response. + self.latch_stream_required(); tracing::info!( provider = %err.provider, model = %err.model.as_deref().unwrap_or("?"), + status = ?err.status, "[openai] provider rejected non-streaming request; \ falling back to streaming path" );