Skip to content

Commit 1ccd48a

Browse files
authored
fix(openai): latch the stream-only constraint and widen its detection (#85)
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.
1 parent 1862845 commit 1ccd48a

3 files changed

Lines changed: 251 additions & 28 deletions

File tree

‎src/harness/providers/openai/mod.rs‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,15 @@
2626
//! or automatically as a single retry when a 400 body implicates the shape. See
2727
//! the module `README.md` "Local-server compatibility" section.
2828
//!
29+
//! Some OpenAI-compatible proxies go further and refuse unary calls entirely,
30+
//! answering `stream: false` with an HTTP 400/422 such as
31+
//! `{"detail":"Stream must be set to true"}`. [`ChatModel::invoke`] recognises
32+
//! that family of rejections (`is_stream_required_error` in `transport`), folds
33+
//! the SSE stream into a single [`ModelResponse`] instead, and **latches** the
34+
//! constraint on the instance so only the first call pays the rejected round
35+
//! trip. Declare it up front with
36+
//! [`OpenAiModel::with_requires_streaming`] to skip even that one.
37+
//!
2938
//! # Example
3039
//!
3140
//! ```no_run
@@ -90,7 +99,7 @@ use sse::*;
9099
#[cfg(test)]
91100
use transport::{
92101
Degrade, auth_headers, degrade_for_400, effective_temperature, glob_match,
93-
merge_provider_options, merge_system_into_user, request_timeout,
102+
is_stream_required_error, merge_provider_options, merge_system_into_user, request_timeout,
94103
};
95104

96105
#[cfg(test)]

‎src/harness/providers/openai/test.rs‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -678,6 +678,111 @@ fn requires_streaming_flag_skips_non_streaming_attempt() {
678678
);
679679
}
680680

681+
// openhuman#5165: the streaming-only constraint used to be re-discovered on
682+
// every unary call, because nothing recorded it. `ChatModel::invoke` takes
683+
// `&self`, so the latch has to be interior-mutable — a plain `bool` field could
684+
// only ever be set by the builder, which nothing calls in production. Without
685+
// the latch a streaming-only proxy costs one guaranteed-400 round trip per
686+
// agent turn forever (511 events / 2 users on the linked Sentry issue).
687+
#[test]
688+
fn stream_required_constraint_latches_after_discovery() {
689+
let m = model();
690+
assert!(!m.requires_streaming(), "must start un-latched");
691+
692+
m.latch_stream_required();
693+
assert!(
694+
m.requires_streaming(),
695+
"a discovered constraint must be remembered so later calls skip \
696+
the doomed non-streaming attempt"
697+
);
698+
699+
// Idempotent: latching again is a no-op (the log fires only on transition).
700+
m.latch_stream_required();
701+
assert!(m.requires_streaming());
702+
}
703+
704+
#[test]
705+
fn stream_required_latch_survives_through_a_shared_handle() {
706+
// Production holds models as `Arc<dyn ChatModel>`, so the latch must be
707+
// observable through a shared reference — that is the whole reason it is an
708+
// `AtomicBool` and not a `bool`.
709+
let shared: std::sync::Arc<OpenAiModel> = std::sync::Arc::new(model());
710+
let clone = std::sync::Arc::clone(&shared);
711+
assert!(!clone.requires_streaming());
712+
713+
shared.latch_stream_required();
714+
assert!(
715+
clone.requires_streaming(),
716+
"the latch must be visible to every holder of the shared model"
717+
);
718+
}
719+
720+
// The trigger used to be a single case-sensitive `contains("Stream must be set
721+
// to true")`. OpenAI-compatible proxies do not standardise this wording, so any
722+
// other phrasing hard-failed the run instead of falling back to streaming.
723+
#[test]
724+
fn is_stream_required_error_matches_the_known_proxy_wordings() {
725+
let m = model();
726+
727+
for body in [
728+
// The exact body from the openhuman#5165 Sentry report (FastAPI-style
729+
// `detail` envelope — note `parse_error_body` cannot promote it to
730+
// `message`, so the raw-body scan is what catches nesting).
731+
r#"{"detail":"Stream must be set to true"}"#,
732+
// Lower-case, and quoted field name.
733+
r#"{"detail":"stream must be set to true"}"#,
734+
r#"{"error":{"message":"'stream' must be true for this endpoint"}}"#,
735+
// Other phrasings seen from OpenAI-compatible gateways.
736+
r#"{"error":{"message":"streaming is required for this model"}}"#,
737+
r#"{"error":{"message":"Only streaming responses are supported"}}"#,
738+
r#"{"error":{"message":"this deployment requires streaming"}}"#,
739+
r#"{"error":{"message":"stream=true is required"}}"#,
740+
// Nested under a non-standard key alongside a generic top-level message.
741+
r#"{"message":"Bad Request","errors":[{"reason":"stream must be true"}]}"#,
742+
] {
743+
let err = m.parse_error_body(400, body);
744+
assert!(
745+
is_stream_required_error(&err),
746+
"must detect the streaming requirement in: {body}"
747+
);
748+
}
749+
750+
// FastAPI validates request fields with 422, so accept that status too.
751+
let unprocessable = m.parse_error_body(422, r#"{"detail":"Stream must be set to true"}"#);
752+
assert!(
753+
is_stream_required_error(&unprocessable),
754+
"422 with the streaming wording must also trigger the fallback"
755+
);
756+
}
757+
758+
#[test]
759+
fn is_stream_required_error_ignores_unrelated_failures() {
760+
let m = model();
761+
762+
// Wrong status: the same wording on a 500 is a server fault, not a
763+
// request-shape constraint, and must stay retryable rather than be
764+
// rerouted into the streaming path.
765+
let server_error = m.parse_error_body(500, r#"{"detail":"Stream must be set to true"}"#);
766+
assert!(!is_stream_required_error(&server_error));
767+
768+
// Right status, unrelated wording — must not hijack the other 400 handlers
769+
// (`degrade_for_400`, the `/responses` max_output_tokens retry).
770+
for body in [
771+
r#"{"error":{"message":"Invalid value for 'tool_choice'"}}"#,
772+
r#"{"error":{"message":"max_output_tokens is not supported"}}"#,
773+
// Mentions stream but states no requirement.
774+
r#"{"error":{"message":"stream_options is not supported by this model"}}"#,
775+
// States a requirement but about a different field.
776+
r#"{"error":{"message":"'store' must be true"}}"#,
777+
] {
778+
let err = m.parse_error_body(400, body);
779+
assert!(
780+
!is_stream_required_error(&err),
781+
"must NOT treat this as a streaming requirement: {body}"
782+
);
783+
}
784+
}
785+
681786
#[test]
682787
fn reasoning_tag_extraction_defaults_off_for_hosted_openai_only() {
683788
// Hosted OpenAI never emits inline `<think>` reasoning; unconditional

‎src/harness/providers/openai/transport.rs‎

Lines changed: 136 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,8 @@
77
use super::responses;
88
use super::*;
99

10+
use std::sync::atomic::{AtomicBool, Ordering};
11+
1012
use crate::harness::model::StreamAccumulator;
1113

1214
/// How the provider expects the API credential to be sent on each request.
@@ -113,11 +115,22 @@ pub struct OpenAiModel {
113115
/// inline `<think>` reasoning, and unconditional extraction would silently
114116
/// strip legitimate content that mentions a literal `<think>` tag.
115117
reasoning_tags_overridden: bool,
116-
/// When `true`, the unary [`ChatModel::invoke`] path sends `stream: true`
117-
/// on the wire and folds the server-sent-events stream into a single
118+
/// When set, the unary [`ChatModel::invoke`] path sends `stream: true` on
119+
/// the wire and folds the server-sent-events stream into a single
118120
/// [`ModelResponse`] internally — for providers that reject a
119-
/// `stream: false` body with an HTTP 400. `false` by default.
120-
requires_streaming: bool,
121+
/// `stream: false` body (`{"detail":"Stream must be set to true"}` and
122+
/// friends). Starts `false`; seeded by
123+
/// [`with_requires_streaming`](Self::with_requires_streaming) and
124+
/// **latched at run time** the first time the provider rejects a
125+
/// non-streaming request.
126+
///
127+
/// It is an [`AtomicBool`] rather than a plain `bool` because
128+
/// [`ChatModel::invoke`] takes `&self` (models are shared behind
129+
/// `Arc<dyn ChatModel>`): without interior mutability the discovery could
130+
/// not be remembered, so every single call would keep paying a
131+
/// guaranteed-400 round trip before falling back. See
132+
/// [`Self::latch_stream_required`].
133+
stream_required: AtomicBool,
121134
}
122135

123136
/// The auth headers `(name, value)` for a given [`AuthStyle`] + credential.
@@ -345,21 +358,45 @@ impl OpenAiModel {
345358
// `<think>` and must not strip literal mentions of the tag.
346359
reasoning_tags: Some(ReasoningTagExtraction::default()),
347360
reasoning_tags_overridden: false,
348-
requires_streaming: false,
361+
stream_required: AtomicBool::new(false),
349362
}
350363
}
351364

352365
/// When the provider requires `stream: true` for every request, routes
353366
/// unary [`invoke`](ChatModel::invoke) through the streaming path internally.
354-
pub fn with_requires_streaming(mut self, enabled: bool) -> Self {
355-
self.requires_streaming = enabled;
367+
///
368+
/// Optional: the transport also discovers this on its own (see
369+
/// [`Self::latch_stream_required`]). Set it explicitly to skip the one
370+
/// exploratory request that discovery costs.
371+
pub fn with_requires_streaming(self, enabled: bool) -> Self {
372+
self.stream_required.store(enabled, Ordering::Relaxed);
356373
self
357374
}
358375

359-
/// Returns whether the provider requires streaming for all calls.
360-
#[cfg(test)]
361-
pub(super) fn requires_streaming(&self) -> bool {
362-
self.requires_streaming
376+
/// Returns whether this instance currently requires streaming for all
377+
/// unary calls — either declared up front via
378+
/// [`with_requires_streaming`](Self::with_requires_streaming) or learned
379+
/// from a provider rejection.
380+
pub fn requires_streaming(&self) -> bool {
381+
self.stream_required.load(Ordering::Relaxed)
382+
}
383+
384+
/// Remembers that this endpoint only accepts `stream: true`.
385+
///
386+
/// Called once the provider has told us so with an HTTP 400/422 (see
387+
/// [`is_stream_required_error`]). Without this latch the transport
388+
/// re-discovers the constraint on **every** unary call, so a streaming-only
389+
/// proxy costs one guaranteed-400 round trip per agent turn — the shape
390+
/// behind the 511 events / 2 users on openhuman#5165. Logs on the
391+
/// transition only, so the discovery is visible exactly once per process.
392+
pub(super) fn latch_stream_required(&self) {
393+
if !self.stream_required.swap(true, Ordering::Relaxed) {
394+
tracing::info!(
395+
provider = %self.provider,
396+
model = %self.model,
397+
"[openai] provider requires stream:true; latching for subsequent calls"
398+
);
399+
}
363400
}
364401

365402
/// Routes calls to the OpenAI **Responses API** (`/v1/responses`) instead of
@@ -1390,6 +1427,73 @@ pub(super) struct Degrade {
13901427
pub json_object: bool,
13911428
}
13921429

1430+
/// Statuses an OpenAI-compatible proxy uses to reject a non-streaming request.
1431+
///
1432+
/// `400` is what the reference implementation returns. `422` is included because
1433+
/// the wording seen in the wild arrives in a FastAPI `{"detail": …}` envelope,
1434+
/// and FastAPI's own rejection status is `422` — a proxy that validates `stream`
1435+
/// as a request field rather than raising an explicit `HTTPException(400)` lands
1436+
/// there. The stream-specific wording still has to match, so widening the status
1437+
/// set does not widen the false-positive surface meaningfully.
1438+
const STREAM_REQUIRED_STATUSES: [u16; 2] = [400, 422];
1439+
1440+
/// Recognises "this endpoint only accepts `stream: true`" from a provider error.
1441+
///
1442+
/// Some OpenAI-compatible proxies refuse unary calls outright. The wording is
1443+
/// not standardised, so match a family of phrasings case-insensitively rather
1444+
/// than one literal string:
1445+
///
1446+
/// - `{"detail":"Stream must be set to true"}` — the openhuman#5165 report
1447+
/// - `stream must be true` / `'stream' must be set to true`
1448+
/// - `streaming is required` / `only streaming is supported`
1449+
/// - `stream=true is required`
1450+
///
1451+
/// Both the decoded `message` **and** the raw body are searched, because
1452+
/// [`OpenAiModel::parse_error_body`] only promotes `error.message` / `message`
1453+
/// to `ProviderError::message`. A proxy that nests the reason under any other
1454+
/// key (`detail`, `errors[0].reason`, …) alongside a generic top-level
1455+
/// `message` would otherwise be invisible to the check.
1456+
///
1457+
/// The `stream` mention is required in addition to the "must be true" phrasing
1458+
/// so unrelated 400s ("`max_output_tokens` must be true"-shaped nonsense,
1459+
/// `tool_choice` rejections) never route a call into the streaming path.
1460+
///
1461+
/// Pure, so the detection policy is unit-testable without a network call.
1462+
pub(super) fn is_stream_required_error(error: &ProviderError) -> bool {
1463+
if !error
1464+
.status
1465+
.is_some_and(|status| STREAM_REQUIRED_STATUSES.contains(&status))
1466+
{
1467+
return false;
1468+
}
1469+
if mentions_stream_required(&error.message) {
1470+
return true;
1471+
}
1472+
error
1473+
.raw
1474+
.as_ref()
1475+
.is_some_and(|raw| mentions_stream_required(&raw.to_string()))
1476+
}
1477+
1478+
/// The wording half of [`is_stream_required_error`], split out so both the
1479+
/// decoded message and the raw body run through identical rules.
1480+
fn mentions_stream_required(haystack: &str) -> bool {
1481+
let lower = haystack.to_ascii_lowercase();
1482+
if !lower.contains("stream") {
1483+
return false;
1484+
}
1485+
const PHRASES: [&str; 7] = [
1486+
"must be set to true",
1487+
"must be true",
1488+
"streaming is required",
1489+
"requires streaming",
1490+
"only streaming",
1491+
"stream=true",
1492+
"stream: true is required",
1493+
];
1494+
PHRASES.iter().any(|phrase| lower.contains(phrase))
1495+
}
1496+
13931497
/// Computes the additional degradation to apply after an HTTP 400, or `None`
13941498
/// when the failure is not an auto-degradable request-shape rejection.
13951499
///
@@ -1514,10 +1618,12 @@ impl<State: Send + Sync> ChatModel<State> for OpenAiModel {
15141618
/// Invokes the OpenAI Chat Completions endpoint and maps the response into a
15151619
/// [`ModelResponse`].
15161620
///
1517-
/// When the provider requires streaming
1518-
/// ([`Self::requires_streaming`] or a 400 with `"Stream must be set to
1519-
/// true"`), the call is degraded to the streaming path internally and the
1520-
/// SSE stream is folded into a single response.
1621+
/// When the provider requires streaming — declared via
1622+
/// [`OpenAiModel::with_requires_streaming`], or discovered from a rejection
1623+
/// matching [`is_stream_required_error`] — the call is degraded to the
1624+
/// streaming path internally and the SSE stream is folded into a single
1625+
/// response. A discovered constraint is latched on the instance, so only the
1626+
/// first call pays the rejected round trip.
15211627
///
15221628
/// # Errors
15231629
///
@@ -1529,13 +1635,15 @@ impl<State: Send + Sync> ChatModel<State> for OpenAiModel {
15291635
return self.invoke_responses(&request).await;
15301636
}
15311637

1532-
// Short-circuit: providers that require `stream: true` on every call
1533-
// skip the non-streaming attempt entirely to avoid a guaranteed 400.
1534-
if self.requires_streaming {
1535-
tracing::info!(
1536-
provider = self.provider,
1638+
// Short-circuit: providers known to require `stream: true` on every
1639+
// call skip the non-streaming attempt entirely to avoid a guaranteed
1640+
// 400. "Known" covers both an explicit `with_requires_streaming(true)`
1641+
// and a constraint latched from an earlier rejection on this instance.
1642+
if self.requires_streaming() {
1643+
tracing::debug!(
1644+
provider = %self.provider,
15371645
model = %self.model,
1538-
"[openai] requires_streaming is set; folding stream into unary response"
1646+
"[openai] stream:true required; folding stream into unary response"
15391647
);
15401648
return invoke_with_streaming(self, request).await;
15411649
}
@@ -1545,15 +1653,16 @@ impl<State: Send + Sync> ChatModel<State> for OpenAiModel {
15451653
.await
15461654
{
15471655
Ok(response) => response,
1548-
Err(TinyAgentsError::Provider(err))
1549-
if err.status == Some(400)
1550-
&& err.message.contains("Stream must be set to true") =>
1551-
{
1552-
// The provider rejects `stream: false`. Fall back to the
1553-
// streaming path and fold the SSE stream into a response.
1656+
Err(TinyAgentsError::Provider(err)) if is_stream_required_error(&err) => {
1657+
// The provider rejects `stream: false`. Remember it so the next
1658+
// call goes straight to streaming instead of re-paying this
1659+
// failed round trip, then fall back to the streaming path and
1660+
// fold the SSE stream into a response.
1661+
self.latch_stream_required();
15541662
tracing::info!(
15551663
provider = %err.provider,
15561664
model = %err.model.as_deref().unwrap_or("?"),
1665+
status = ?err.status,
15571666
"[openai] provider rejected non-streaming request; \
15581667
falling back to streaming path"
15591668
);

0 commit comments

Comments
 (0)