fix(openai): latch the stream-only constraint and widen its detection

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)]

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<dyn ChatModel>`, 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<OpenAiModel> = 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 `<think>` reasoning; unconditional

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 `<think>` reasoning, and unconditional extraction would silently
     /// strip legitimate content that mentions a literal `<think>` 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<dyn ChatModel>`): 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 {
             // `<think>` 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<State: Send + Sync> ChatModel<State> 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<State: Send + Sync> ChatModel<State> 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<State: Send + Sync> ChatModel<State> 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"
                 );