From 41a3498b8ac341b1bfa3116e3d130f7177561e80 Mon Sep 17 00:00:00 2001 From: shanu Date: Wed, 12 Aug 2026 13:41:58 +0530 Subject: [PATCH 1/4] Share the stream-required discovery across instances by endpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-instance stream_required latch stops a single OpenAiModel from re-probing a streaming-only endpoint, but hosts build a fresh OpenAiModel per workload (chat, summariser, titler, …) all aimed at the same endpoint. So every new instance re-discovered the constraint and paid another guaranteed-400 ("Stream must be set to true") round trip — the 400 kept reappearing on every workload, every turn, even after one instance had already learned better. Record the discovery in a process-global set keyed by base_url (the constraint is a property of the endpoint, not of any one instance or model id). requires_streaming() now adopts a sibling's discovery and caches it locally; latch_stream_required() records it once per endpoint and logs once per process. Explicit with_requires_streaming() stays per-instance so it can still be cleared. After the first cold probe the endpoint is never re-probed. Refs openhuman#5497 (multi-instance shape of openhuman#5165). --- src/harness/providers/openai/test.rs | 39 +++++++++++++- src/harness/providers/openai/transport.rs | 65 +++++++++++++++++++++-- 2 files changed, 99 insertions(+), 5 deletions(-) diff --git a/src/harness/providers/openai/test.rs b/src/harness/providers/openai/test.rs index 6f347f6..bb38c42 100644 --- a/src/harness/providers/openai/test.rs +++ b/src/harness/providers/openai/test.rs @@ -708,7 +708,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 +730,8 @@ 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 shared: std::sync::Arc = + 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 +742,38 @@ fn stream_required_latch_survives_through_a_shared_handle() { ); } +// 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" + ); +} + // 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. diff --git a/src/harness/providers/openai/transport.rs b/src/harness/providers/openai/transport.rs index 789787d..466b371 100644 --- a/src/harness/providers/openai/transport.rs +++ b/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; @@ -473,7 +475,18 @@ impl OpenAiModel { /// [`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) + if self.stream_required.load(Ordering::Relaxed) { + return true; + } + // A sibling instance pointed at the same endpoint may have already + // discovered the constraint. Adopt that so this fresh instance skips the + // doomed non-streaming probe too, then cache it locally to avoid the + // shared-lock read on subsequent calls. + 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 +498,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 = %self.base_url, + "[openai] provider requires stream:true; latching endpoint-wide for subsequent calls" ); } } @@ -2307,6 +2326,46 @@ 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>> = + LazyLock::new(|| RwLock::new(HashSet::new())); + +/// 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 From ac8a78f426d9b84e0a84793599d0a985c00e5756 Mon Sep 17 00:00:00 2001 From: shanu Date: Wed, 12 Aug 2026 15:47:36 +0530 Subject: [PATCH 2/4] Redact base_url before logging it in the stream-required latch The endpoint-wide latch discovery logs base_url. An OpenAI-compatible proxy can carry a credential in the URL (userinfo or a query param), so logging the raw value risks leaking a secret. Route it through redact_base_url_for_log, which strips userinfo + query + fragment (keeping scheme://host[:port]/path) and replaces any value that does not parse as a URL. Add a unit test covering a credential-bearing URL, a clean URL, and an unparseable value. --- src/harness/providers/openai/test.rs | 37 +++++++++++++++++++++++ src/harness/providers/openai/transport.rs | 21 ++++++++++++- 2 files changed, 57 insertions(+), 1 deletion(-) diff --git a/src/harness/providers/openai/test.rs b/src/harness/providers/openai/test.rs index bb38c42..6883b62 100644 --- a/src/harness/providers/openai/test.rs +++ b/src/harness/providers/openai/test.rs @@ -747,6 +747,43 @@ fn stream_required_latch_survives_through_a_shared_handle() { // 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 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", + ); + assert!( + !redacted.contains("sup3rsecret"), + "password must be stripped: {redacted}" + ); + assert!( + !redacted.contains("api_key") && !redacted.contains("abc123"), + "query must be stripped: {redacted}" + ); + assert!( + !redacted.contains("user@") && !redacted.contains("user:"), + "username must be stripped: {redacted}" + ); + assert!( + redacted.starts_with("https://proxy.example.com:8443/v1"), + "scheme, host, port and path are preserved: {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::"), + "" + ); +} + #[test] fn stream_required_discovery_is_shared_across_instances_by_endpoint() { let base = "https://shared-stream-discovery.invalid/v1"; diff --git a/src/harness/providers/openai/transport.rs b/src/harness/providers/openai/transport.rs index 466b371..0072ee8 100644 --- a/src/harness/providers/openai/transport.rs +++ b/src/harness/providers/openai/transport.rs @@ -507,7 +507,7 @@ impl OpenAiModel { tracing::info!( provider = %self.provider, model = %self.model, - base_url = %self.base_url, + base_url = %redact_base_url_for_log(&self.base_url), "[openai] provider requires stream:true; latching endpoint-wide for subsequent calls" ); } @@ -2345,6 +2345,25 @@ const STREAM_REQUIRED_STATUSES: [u16; 2] = [400, 422]; static STREAM_REQUIRED_ENDPOINTS: LazyLock>> = 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(_) => "".to_string(), + } +} + /// Whether `base_url` has already been discovered to require streaming. fn endpoint_requires_streaming(base_url: &str) -> bool { STREAM_REQUIRED_ENDPOINTS From 1fa837624ef0d563e2377fe0731a5024518fd5de Mon Sep 17 00:00:00 2001 From: shanu Date: Wed, 12 Aug 2026 17:52:14 +0530 Subject: [PATCH 3/4] Assert the exact redacted URL in the redaction test starts_with plus negative token checks did not prove the query and fragment were removed (a URL with a different query/fragment would pass). Replace with an exact-equality assertion against the fully redacted value, which proves userinfo, query and fragment are all gone while scheme, host, port and path survive. --- src/harness/providers/openai/test.rs | 20 +++++--------------- 1 file changed, 5 insertions(+), 15 deletions(-) diff --git a/src/harness/providers/openai/test.rs b/src/harness/providers/openai/test.rs index 6883b62..e8145c0 100644 --- a/src/harness/providers/openai/test.rs +++ b/src/harness/providers/openai/test.rs @@ -754,21 +754,11 @@ fn redact_base_url_for_log_strips_credentials_and_query() { let redacted = super::transport::redact_base_url_for_log( "https://user:sup3rsecret@proxy.example.com:8443/v1?api_key=abc123#frag", ); - assert!( - !redacted.contains("sup3rsecret"), - "password must be stripped: {redacted}" - ); - assert!( - !redacted.contains("api_key") && !redacted.contains("abc123"), - "query must be stripped: {redacted}" - ); - assert!( - !redacted.contains("user@") && !redacted.contains("user:"), - "username must be stripped: {redacted}" - ); - assert!( - redacted.starts_with("https://proxy.example.com:8443/v1"), - "scheme, host, port and path are preserved: {redacted}" + // 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. From ff12169e0230b34661c1f8271ca6c7196a82a316 Mon Sep 17 00:00:00 2001 From: shanu Date: Wed, 12 Aug 2026 20:15:40 +0530 Subject: [PATCH 4/4] Make an explicit with_requires_streaming an opt-out that beats shared discovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of #98: once discovery went process-global, with_requires_streaming(false) stopped being a real opt-out — requires_streaming() would return true for an instance that explicitly set false the moment any sibling latched the same base_url, and invoke() has no path back once it is true. That bites an aggregating proxy fronting mixed upstreams on one base_url (a titler routed to a non-streaming upstream forced onto the streaming path). Track whether the value was set explicitly (stream_required_explicit); when it was, requires_streaming() returns the instance's own value and does not consult the shared registry, so an explicit opt-out (or opt-in) always wins. Add a test covering opt-out-beats-sibling-discovery and opt-in-needs-no-discovery, and fix the doc comments to describe the real precedence. Also address review nits: move the misplaced openhuman#5497 comment onto the sharing test it describes, and warn on fn model() that latching the default base URL leaks into other tests via the global registry. --- src/harness/providers/openai/test.rs | 46 ++++++++++++++++-- src/harness/providers/openai/transport.rs | 57 +++++++++++++++++------ 2 files changed, 85 insertions(+), 18 deletions(-) diff --git a/src/harness/providers/openai/test.rs b/src/harness/providers/openai/test.rs index e8145c0..a4f402b 100644 --- a/src/harness/providers/openai/test.rs +++ b/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") } @@ -742,11 +749,6 @@ fn stream_required_latch_survives_through_a_shared_handle() { ); } -// 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 redact_base_url_for_log_strips_credentials_and_query() { // A proxy endpoint that carries a secret in userinfo or a query param must @@ -774,6 +776,11 @@ fn redact_base_url_for_log_strips_credentials_and_query() { ); } +// 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"; @@ -801,6 +808,35 @@ fn stream_required_discovery_is_shared_across_instances_by_endpoint() { ); } +// 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. diff --git a/src/harness/providers/openai/transport.rs b/src/harness/providers/openai/transport.rs index 0072ee8..00ef30d 100644 --- a/src/harness/providers/openai/transport.rs +++ b/src/harness/providers/openai/transport.rs @@ -142,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`). /// @@ -451,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(), @@ -459,29 +472,47 @@ 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 { + // (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; } - // A sibling instance pointed at the same endpoint may have already - // discovered the constraint. Adopt that so this fresh instance skips the - // doomed non-streaming probe too, then cache it locally to avoid the - // shared-lock read on subsequent calls. + // (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;