Share the stream-required discovery across OpenAiModel instances by endpoint

src/harness/providers/openai/test.rs

@@ -17,6 +17,13 @@ use crate::harness::providers::{ProviderKind, ProviderSpec};
 use crate::harness::tool::ToolSchema;
 
 /// Builds a model with a fixed key/model so translation output is deterministic.
+///
+/// NOTE: this uses the default base URL. The stream-required latch now records
+/// discovery in a process-global registry keyed by `base_url`, so a test that
+/// calls `latch_stream_required()` on a `model()` would flip
+/// `requires_streaming()` to `true` for every other test on the default base URL
+/// (e.g. `requires_streaming_flag_skips_non_streaming_attempt`). A test that
+/// latches MUST first override the base URL with a unique `.with_base_url(...)`.
 fn model() -> OpenAiModel {
     OpenAiModel::new("test-key").with_model("gpt-4.1-mini")
 }
@@ -708,7 +715,9 @@ fn requires_streaming_flag_skips_non_streaming_attempt() {
 // agent turn forever (511 events / 2 users on the linked Sentry issue).
 #[test]
 fn stream_required_constraint_latches_after_discovery() {
-    let m = model();
+    // A unique base_url keeps this test's runtime discovery out of the
+    // process-global endpoint registry shared with the DEFAULT_BASE_URL tests.
+    let m = model().with_base_url("https://stream-latch-discovery.invalid/v1");
     assert!(!m.requires_streaming(), "must start un-latched");
 
     m.latch_stream_required();
@@ -728,7 +737,8 @@ 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 shared: std::sync::Arc<OpenAiModel> =
+        std::sync::Arc::new(model().with_base_url("https://stream-latch-shared-handle.invalid/v1"));
     let clone = std::sync::Arc::clone(&shared);
     assert!(!clone.requires_streaming());
 
@@ -739,6 +749,94 @@ fn stream_required_latch_survives_through_a_shared_handle() {
     );
 }
 
+#[test]
+fn redact_base_url_for_log_strips_credentials_and_query() {
+    // A proxy endpoint that carries a secret in userinfo or a query param must
+    // never reach a log line (the latch discovery logs base_url).
+    let redacted = super::transport::redact_base_url_for_log(
+        "https://user:sup3rsecret@proxy.example.com:8443/v1?api_key=abc123#frag",
+    );
+    // Exact match proves userinfo, query AND fragment are all gone (not just
+    // the specific secret tokens) while scheme, host, port and path survive.
+    assert_eq!(
+        redacted, "https://proxy.example.com:8443/v1",
+        "credentials, query and fragment must be stripped; host/port/path kept: {redacted}"
+    );
+
+    // A credential-free URL is unchanged.
+    assert_eq!(
+        super::transport::redact_base_url_for_log("https://api.openai.com/v1"),
+        "https://api.openai.com/v1"
+    );
+
+    // A value that does not parse as a URL is replaced wholesale, never logged.
+    assert_eq!(
+        super::transport::redact_base_url_for_log("::not a url::"),
+        "<unparseable base_url redacted>"
+    );
+}
+
+// openhuman#5497: the per-instance latch is not enough on its own. Hosts build a
+// fresh `OpenAiModel` per workload (chat, summariser, titler, …) all aimed at
+// the same streaming-only endpoint, so a purely per-instance latch lets the
+// guaranteed-400 probe reappear for every workload on every turn. The discovery
+// is therefore shared process-wide, keyed by endpoint.
+#[test]
+fn stream_required_discovery_is_shared_across_instances_by_endpoint() {
+    let base = "https://shared-stream-discovery.invalid/v1";
+
+    let discoverer = OpenAiModel::new("k").with_base_url(base);
+    assert!(!discoverer.requires_streaming(), "must start un-latched");
+    discoverer.latch_stream_required();
+    assert!(discoverer.requires_streaming());
+
+    // A brand-new instance (different key + model, same endpoint) inherits the
+    // constraint without ever issuing its own doomed non-streaming probe.
+    let sibling = OpenAiModel::new("other-key")
+        .with_model("some-other-model")
+        .with_base_url(base);
+    assert!(
+        sibling.requires_streaming(),
+        "a fresh instance for the same endpoint must adopt the shared discovery"
+    );
+
+    // Scoped per endpoint — an unrelated base_url is unaffected.
+    let elsewhere = OpenAiModel::new("k").with_base_url("https://unrelated-endpoint.invalid/v1");
+    assert!(
+        !elsewhere.requires_streaming(),
+        "the shared record must be scoped per endpoint, not global-for-all"
+    );
+}
+
+// An explicit `with_requires_streaming(false)` is an opt-out and must stay one
+// even after a sibling latches the same endpoint — otherwise an aggregating
+// proxy (mixed upstreams behind one base_url) could force a deliberately-opted-
+// out model onto the streaming path with no way back (openhuman#5497 review).
+#[test]
+fn explicit_requires_streaming_opt_out_wins_over_sibling_discovery() {
+    let base = "https://escape-hatch-endpoint.invalid/v1";
+
+    // A sibling discovers the endpoint is streaming-only and records it globally.
+    let discoverer = OpenAiModel::new("k").with_base_url(base);
+    discoverer.latch_stream_required();
+    assert!(discoverer.requires_streaming());
+
+    // A model explicitly opted out on the SAME endpoint keeps the unary path.
+    let opted_out = OpenAiModel::new("k")
+        .with_base_url(base)
+        .with_requires_streaming(false);
+    assert!(
+        !opted_out.requires_streaming(),
+        "an explicit opt-out must win over a sibling's endpoint-wide discovery"
+    );
+
+    // An explicit opt-in is likewise authoritative and needs no discovery.
+    let opted_in = OpenAiModel::new("k")
+        .with_base_url("https://opt-in-endpoint.invalid/v1")
+        .with_requires_streaming(true);
+    assert!(opted_in.requires_streaming());
+}
+
 // 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.

src/harness/providers/openai/transport.rs

@@ -7,7 +7,9 @@
 use super::responses;
 use super::*;
 
+use std::collections::HashSet;
 use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{LazyLock, RwLock};
 
 use crate::harness::model::StreamAccumulator;
 
@@ -140,6 +142,18 @@ pub struct OpenAiModel {
     /// guaranteed-400 round trip before falling back. See
     /// [`Self::latch_stream_required`].
     stream_required: AtomicBool,
+    /// Whether [`stream_required`](Self::stream_required) was set explicitly by a
+    /// caller via [`with_requires_streaming`](Self::with_requires_streaming), as
+    /// opposed to left at its default or discovered at run time.
+    ///
+    /// When `true`, [`requires_streaming`](Self::requires_streaming) returns the
+    /// instance's own value and does **not** consult the process-wide
+    /// endpoint registry — an explicit opt-out (`with_requires_streaming(false)`)
+    /// therefore wins over a sibling's endpoint-wide discovery. Without this, an
+    /// aggregating proxy that fronts mixed upstreams on one `base_url` would
+    /// force a model the host deliberately opted out (e.g. a titler routed to a
+    /// non-streaming upstream) onto the streaming path with no way back.
+    stream_required_explicit: AtomicBool,
     /// Whether a JSON-Schema `response_format` is sent with OpenAI **strict**
     /// structured output (`"strict": true`).
     ///
@@ -449,6 +463,7 @@ impl OpenAiModel {
             reasoning_tags: Some(ReasoningTagExtraction::default()),
             reasoning_tags_overridden: false,
             stream_required: AtomicBool::new(false),
+            stream_required_explicit: AtomicBool::new(false),
             json_schema_strict: AtomicBool::new(true),
             native_tools_on_wire: AtomicBool::new(true),
             cache_accounting: CacheTokenAccounting::default(),
@@ -457,23 +472,52 @@ impl OpenAiModel {
         }
     }
 
-    /// When the provider requires `stream: true` for every request, routes
-    /// unary [`invoke`](ChatModel::invoke) through the streaming path internally.
+    /// Declares up front whether the provider requires `stream: true` for every
+    /// request. When enabled, unary [`invoke`](ChatModel::invoke) is routed
+    /// through the streaming path internally, skipping the one exploratory
+    /// request that run-time discovery costs.
     ///
-    /// 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.
+    /// This is an **explicit, per-instance override**: it is authoritative for
+    /// this instance and is never overridden by a sibling's endpoint-wide
+    /// discovery (see [`Self::requires_streaming`]). So `with_requires_streaming(false)`
+    /// is a genuine opt-out — a model the host knows does not need streaming
+    /// (e.g. a titler routed to a non-streaming upstream behind an aggregating
+    /// proxy) stays on the unary path even if another model sharing the same
+    /// `base_url` has latched the streaming constraint.
     pub fn with_requires_streaming(self, enabled: bool) -> Self {
         self.stream_required.store(enabled, Ordering::Relaxed);
+        self.stream_required_explicit.store(true, Ordering::Relaxed);
         self
     }
 
-    /// 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.
+    /// Returns whether this instance requires streaming for all unary calls.
+    ///
+    /// Precedence:
+    /// 1. An **explicit** [`with_requires_streaming`](Self::with_requires_streaming)
+    ///    is authoritative and short-circuits the shared registry, so an opt-out
+    ///    is never clobbered by another model on the same endpoint.
+    /// 2. Otherwise a value **learned at run time** on this instance (see
+    ///    [`Self::latch_stream_required`]) wins.
+    /// 3. Otherwise a **sibling's** endpoint-wide discovery is adopted (and
+    ///    cached locally to keep the shared read off the hot path).
     pub fn requires_streaming(&self) -> bool {
-        self.stream_required.load(Ordering::Relaxed)
+        // (1) An explicit setting wins outright — including an explicit `false`,
+        // which must remain a working opt-out.
+        if self.stream_required_explicit.load(Ordering::Relaxed) {
+            return self.stream_required.load(Ordering::Relaxed);
+        }
+        // (2) This instance's own value (default or run-time-latched).
+        if self.stream_required.load(Ordering::Relaxed) {
+            return true;
+        }
+        // (3) A sibling pointed at the same endpoint may have already discovered
+        // the constraint. Adopt it so this fresh instance skips the doomed
+        // non-streaming probe, and cache it locally.
+        if endpoint_requires_streaming(&self.base_url) {
+            self.stream_required.store(true, Ordering::Relaxed);
+            return true;
+        }
+        false
     }
 
     /// Remembers that this endpoint only accepts `stream: true`.
@@ -485,11 +529,17 @@ impl OpenAiModel {
     /// 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) {
+        self.stream_required.store(true, Ordering::Relaxed);
+        // Share the discovery process-wide so sibling instances (a fresh model
+        // per workload) skip the probe instead of re-paying the 400. Log on the
+        // first record per endpoint, so a streaming-only proxy is announced
+        // exactly once per process rather than once per instance.
+        if remember_endpoint_requires_streaming(&self.base_url) {
             tracing::info!(
                 provider = %self.provider,
                 model = %self.model,
-                "[openai] provider requires stream:true; latching for subsequent calls"
+                base_url = %redact_base_url_for_log(&self.base_url),
+                "[openai] provider requires stream:true; latching endpoint-wide for subsequent calls"
             );
         }
     }
@@ -2307,6 +2357,65 @@ pub(super) struct Degrade {
 /// set does not widen the false-positive surface meaningfully.
 const STREAM_REQUIRED_STATUSES: [u16; 2] = [400, 422];
 
+/// Process-global set of base URLs known to only accept `stream: true`.
+///
+/// The per-instance [`OpenAiModel::stream_required`] latch stops a *single*
+/// model from re-probing, but hosts build a **fresh `OpenAiModel` per workload**
+/// (chat, summariser, titler, …), each pointed at the same endpoint. Without a
+/// shared record every new instance re-discovers the constraint and pays another
+/// guaranteed-400 round trip — so a streaming-only proxy keeps emitting the 400
+/// on every turn even though one instance already learned better
+/// (openhuman#5497, the multi-instance shape of openhuman#5165).
+///
+/// Keyed by `base_url` because the constraint is a property of the endpoint /
+/// proxy, not of any one model id or instance. Written **only** by runtime
+/// discovery ([`OpenAiModel::latch_stream_required`]); an explicit
+/// [`with_requires_streaming`](OpenAiModel::with_requires_streaming) stays
+/// per-instance so it can still be cleared per instance. Entries are never
+/// removed — a streaming-only endpoint does not stop being one within a process.
+static STREAM_REQUIRED_ENDPOINTS: LazyLock<RwLock<HashSet<String>>> =
+    LazyLock::new(|| RwLock::new(HashSet::new()));
+
+/// Redacts a base URL for logging: drops any embedded credentials (userinfo)
+/// and query/fragment, keeping only `scheme://host[:port]/path`. Some
+/// OpenAI-compatible proxies carry an API key in the URL (userinfo or a query
+/// param), so the raw `base_url` must never reach a log line. A value that does
+/// not parse as a URL is replaced wholesale rather than logged, so a malformed
+/// endpoint cannot leak either.
+pub(super) fn redact_base_url_for_log(base_url: &str) -> String {
+    match reqwest::Url::parse(base_url) {
+        Ok(mut url) => {
+            let _ = url.set_username("");
+            let _ = url.set_password(None);
+            url.set_query(None);
+            url.set_fragment(None);
+            url.to_string()
+        }
+        Err(_) => "<unparseable base_url redacted>".to_string(),
+    }
+}
+
+/// Whether `base_url` has already been discovered to require streaming.
+fn endpoint_requires_streaming(base_url: &str) -> bool {
+    STREAM_REQUIRED_ENDPOINTS
+        .read()
+        .is_ok_and(|set| set.contains(base_url))
+}
+
+/// Record that `base_url` only accepts `stream: true`. Returns `true` only for
+/// the call that first inserted it, so discovery logs exactly once per endpoint
+/// per process even under a concurrent first-probe race. Fails open (no record)
+/// if the lock is poisoned — callers still hold their own per-instance latch.
+fn remember_endpoint_requires_streaming(base_url: &str) -> bool {
+    if endpoint_requires_streaming(base_url) {
+        return false;
+    }
+    STREAM_REQUIRED_ENDPOINTS
+        .write()
+        .map(|mut set| set.insert(base_url.to_string()))
+        .unwrap_or(false)
+}
+
 /// Recognises "this endpoint only accepts `stream: true`" from a provider error.
 ///
 /// Some OpenAI-compatible proxies refuse unary calls outright. The wording is