feat(stella-model): streaming→non-streaming fallback on hung or empty streams, with a first-byte deadline (#2686)

AGENTS.md

@@ -294,7 +294,7 @@ Append; do not renumber. `scripts/check-invariants.sh` enforces both halves.
    *after* the stable prefix (see `crates/stella-cli/src/agent.rs::build_system_prompt`
    and `crates/stella-cli/src/memory.rs` for the L-E8 discipline).
 8. **Provider feature parity is declared, not assumed.** Providers diverge
-   in sneaky ways, and this is guarded on **two axes** today in
+   in sneaky ways, and this is guarded on **three axes** today in
    `crates/stella-model/src/provider_parity.rs`:
    - **`CachePosture`** — how the prompt cache is engaged/observed
      (Anthropic's cache is explicit opt-in; DeepSeek spells its cache-hit
@@ -307,10 +307,16 @@ Append; do not renumber. `scripts/check-invariants.sh` enforces both halves.
      (`reasoning[_]effort`) honor a pinned effort; the shared adapter drops
      it for `Unsupported` providers (bedrock/deepseek/local) — and a pinned
      effort against one surfaces a one-line boot notice, never a silent drop.
+   - **`StreamFallbackPosture`** — how a provider recovers when its
+     streaming path is broken (a stream hung before its first byte, or a
+     200 with an empty stream). The shared chat-completions adapter arms a
+     bounded per-session latch and re-issues the retried attempt as a unary
+     request (#2686); the other streaming dialects declare the gap; Bedrock
+     is already unary.
 
    Each provider id declares a posture on **every** axis and, for a
-   controllable/opt-in/implicit posture, names the **witness test** proving
-   it on the wire. Tests enforce each matrix from both sides: `stella-cli`'s
+   controllable/opt-in/implicit/fallback posture, names the **witness test**
+   proving it on the wire. Tests enforce each matrix from both sides: `stella-cli`'s
    config tests fail if a seeded provider lacks a row on either axis, and
    `stella-model`'s parity tests fail if a row's witness test no longer
    exists. Adding a provider — or a new divergent feature axis — means
@@ -547,7 +553,7 @@ a plan needs and the part that rarely changes:
 |---|---|
 | `stella-cli` | `src/command_deck.rs`, `src/agent.rs`, `src/agent/tests.rs`, `src/fleet_cmd.rs` |
 | `stella-core` | `src/driver/tests.rs`, `src/driver.rs`, `src/bus.rs` |
-| `stella-model` | `src/openai.rs`, `src/zai/tests.rs`, `src/anthropic/tests.rs`, `src/zai.rs` |
+| `stella-model` | `src/openai.rs`, `src/zai/tests.rs`, `src/anthropic/tests.rs` |
 | `stella-pipeline` | `src/pipeline.rs`, `src/pipeline/tests.rs` |
 | `stella-store` | `src/tests.rs`, `src/lib.rs`, `src/usage.rs` |
 | `stella-tools` | `src/registry.rs`, `src/scripts.rs` |

crates/stella-cli/src/config/tests.rs

@@ -75,6 +75,24 @@ fn every_seeded_provider_declares_a_reasoning_posture() {
     }
 }
 
+/// The stream-fallback sibling (#2686): every seeded provider must declare
+/// how it recovers when its streaming path is broken — a unary fallback
+/// (with the witness proving the retried attempt goes out non-streaming), a
+/// declared streaming-only gap, or an adapter that is already unary. Same
+/// completeness contract as the other two axes.
+#[test]
+fn every_seeded_provider_declares_a_stream_fallback_posture() {
+    for provider in PROVIDERS.iter().chain(std::iter::once(&LOCAL_PROVIDER)) {
+        assert!(
+            stella_model::provider_parity::stream_fallback_posture(provider.id).is_some(),
+            "provider `{}` has no StreamFallbackPosture row in \
+             stella-model/src/provider_parity.rs — add it (with a witness test for a \
+             UnaryFallback row, or a note for a no-fallback posture) in this PR",
+            provider.id
+        );
+    }
+}
+
 #[test]
 fn alias_env_var_resolves_when_the_primary_is_unset() {
     // Synthetic provider with unique var names so parallel tests can't

crates/stella-model/README.md

@@ -85,7 +85,6 @@ the file.
 
 - [`src/anthropic/tests.rs`](src/anthropic/tests.rs)
 - [`src/openai.rs`](src/openai.rs)
-- [`src/zai.rs`](src/zai.rs)
 - [`src/zai/tests.rs`](src/zai/tests.rs)
 
 A ceiling can move only via `make file-size-update`, which lands as a
@@ -99,14 +98,15 @@ escape hatch for an irreducible line (a module declaration in an oversized
 |---|---|
 | [`src/lib.rs`](src/lib.rs) | The crate map (read this first) and the public re-exports: `Provider`, `Catalog`, `ApiKey`, the cache-economics helpers. |
 | [`src/provider.rs`](src/provider.rs) | Two-line re-export of the port. Open it to be reminded where the trait actually lives. |
-| [`src/zai.rs`](src/zai.rs) (+ [`src/zai/tests.rs`](src/zai/tests.rs), [`src/zai/tests/error_classify.rs`](src/zai/tests/error_classify.rs)) | The shared OpenAI Chat Completions adapter. One adapter serving Z.ai/GLM, xAI, DeepSeek, OpenRouter, `local`, and settings-defined gateways; per-identity behavior (OpenRouter's root `cache_control` + sticky `session_id`, GLM's `thinking`, xAI's `reasoning_effort`) is gated on `self.id` inside it. |
+| [`src/zai.rs`](src/zai.rs) (+ [`src/zai/stream.rs`](src/zai/stream.rs), [`src/zai/unary.rs`](src/zai/unary.rs), [`src/zai/tests.rs`](src/zai/tests.rs), [`src/zai/tests/error_classify.rs`](src/zai/tests/error_classify.rs)) | The shared OpenAI Chat Completions adapter. One adapter serving Z.ai/GLM, xAI, DeepSeek, OpenRouter, `local`, and settings-defined gateways; per-identity behavior (OpenRouter's root `cache_control` + sticky `session_id`, GLM's `thinking`, xAI's `reasoning_effort`) is gated on `self.id` inside it. `stream.rs` is the SSE aggregation half, `unary.rs` the non-streaming fallback a broken streaming path latches onto (`src/stream_recovery.rs`, #2686). |
 | [`src/anthropic.rs`](src/anthropic.rs) (+ [`src/anthropic/tests.rs`](src/anthropic/tests.rs)), [`src/openai.rs`](src/openai.rs), [`src/gemini.rs`](src/gemini.rs), [`src/vertex.rs`](src/vertex.rs), [`src/bedrock.rs`](src/bedrock.rs) | One adapter per structurally distinct wire dialect: Messages API, Responses API, `generateContent` (direct and Vertex's project-scoped enterprise path, sharing wire types and the stream aggregator), Bedrock Converse. All follow the same shape — `new(ApiKey, model)` capturing catalog pricing, `with_base_url`, `http::client()`, `SseDecoder`, `http::classify_http_status`, `impl Provider` (plus `complete_observed`, the mid-stream tool-call announcement the engine speculates on — every streaming adapter implements it; Bedrock, which is unary, implements it as one terminal `text_delta` and announces no tool calls). Open the one whose vendor you are debugging. |
 | [`src/catalog.rs`](src/catalog.rs) | `Catalog`, `CatalogEntry`, `Pricing`, `ToolDialect`, and the compile-time seed. The only sanctioned slug → model resolution. |
 | [`src/credential.rs`](src/credential.rs) (+ [`src/credential/aux.rs`](src/credential/aux.rs)) | `ApiKey` (the non-`Display` secret wrapper), the flag → env → file → prompt chain, `CredentialsFile` (keys *and* the `[credential_fields.<provider>]` companions), the multi-variable resolvers `VertexAddressing` / `BedrockCredentials`, and `AuxCredentials` — the redacting, zeroizing set a host uses to carry the values a provider needs beyond one key (Bedrock's secret access key, session token, and region). |
-| [`src/provider_parity.rs`](src/provider_parity.rs) | `CachePosture` / `ReasoningPosture` — the per-provider matrix. A new provider id lands here or tests fail. |
+| [`src/provider_parity.rs`](src/provider_parity.rs) | `CachePosture` / `ReasoningPosture` / `StreamFallbackPosture` — the per-provider matrix. A new provider id lands here or tests fail. |
 | [`src/cache_economics.rs`](src/cache_economics.rs) | Cache savings arithmetic (`Pricing::cache_savings_usd`) and `diagnose_cache`, which reads the parity matrix to tell an opt-in bug from prefix instability. |
 | [`src/sse.rs`](src/sse.rs) | Dependency-free SSE line parser + incremental UTF-8 decoder every streaming adapter feeds. |
-| [`src/http.rs`](src/http.rs) | Crate-private plumbing: the timeout-bounded `reqwest` clients, `classify_http_status`, and the two shared stream-failure errors. Change error retryability here, not in an adapter. |
+| [`src/http.rs`](src/http.rs) | Crate-private plumbing: the timeout-bounded `reqwest` clients, the first-byte/idle stream-read bounds, `classify_http_status`, and the two shared stream-failure errors. Change error retryability here, not in an adapter. |
+| [`src/stream_recovery.rs`](src/stream_recovery.rs) | Crate-private: the streaming→non-streaming fallback latch (`StreamRecovery`) and the fault classification (`StreamFault`) any adapter with a unary fallback path shares (#2686). |
 | [`src/attachment.rs`](src/attachment.rs) | Crate-private. Turns `Attachment`s into dialect-neutral `WirePart`s once, so each adapter only maps parts onto its own JSON. |
 | [`src/modelsdev.rs`](src/modelsdev.rs), [`src/provider_listing.rs`](src/provider_listing.rs) | Fetch-and-parse only: the models.dev master list and provider-native `/models` discovery. Deciding what to store belongs to `stella-cli`. Best-effort by contract — a dead or shape-drifted endpoint returns an `Err(String)` the caller reports and moves past, so one provider can never fail a refresh of the others. |
 

crates/stella-model/src/http.rs

@@ -24,6 +24,21 @@ pub(crate) const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
 /// the turn forever.
 pub(crate) const STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(120);
 
+/// How long to wait for the **first** chunk of a stream's body before
+/// treating the stream itself as broken. Distinct from — and shorter than —
+/// [`STREAM_IDLE_TIMEOUT`], because the two waits mean different things: a
+/// gap *between* fragments is a model thinking, while a response that has
+/// sent its headers and then not one body byte is the signature of a proxy
+/// buffering the SSE body — a path fault the same request completes fine
+/// over a non-streaming call (issue #2686). 90 seconds matches the zappy
+/// comparator's stream watchdog and sits well above any observed
+/// time-to-first-token (which grows with prompt-processing time on large
+/// cache writes, not with generation length), while discovering a buffering
+/// proxy 30s sooner than the idle bound and ~13x sooner than the engine's
+/// 816s model deadline. A trip is fallback-eligible: see
+/// `crate::stream_recovery`.
+pub(crate) const FIRST_BYTE_TIMEOUT: Duration = Duration::from_secs(90);
+
 /// A `reqwest::Client` with [`CONNECT_TIMEOUT`] applied, plus a per-read
 /// stall bound of [`STREAM_IDLE_TIMEOUT`]. The read timeout closes the gap
 /// [`next_with_timeout`] cannot see: the wait between a successful connect
@@ -414,25 +429,55 @@ pub(crate) fn classify_http_status(
     }
 }
 
+/// One bounded stream read, with the four outcomes kept distinct. The
+/// distinction [`next_with_timeout`] collapses — a stall versus a transport
+/// fault — is exactly the bit the streaming→non-streaming fallback needs: a
+/// stall before the first byte is a broken *streaming path* (fallback
+/// material), while a reset says nothing about streaming specifically.
+pub(crate) enum StreamRead<T> {
+    /// The next item arrived within the bound.
+    Item(T),
+    /// Clean end of stream.
+    End,
+    /// Nothing arrived within the bound — the stream is stalled.
+    Idle,
+    /// The transport failed mid-read.
+    Failed(String),
+}
+
+/// Await the next stream item, bounded by `idle`, reporting the outcome as a
+/// [`StreamRead`]. `idle` is a parameter (rather than reading
+/// [`STREAM_IDLE_TIMEOUT`] directly) purely so the timeout path is
+/// unit-testable in milliseconds.
+pub(crate) async fn next_stream_read<S, T>(stream: &mut S, idle: Duration) -> StreamRead<T>
+where
+    S: Stream<Item = reqwest::Result<T>> + Unpin,
+{
+    match tokio::time::timeout(idle, stream.next()).await {
+        Ok(Some(Ok(item))) => StreamRead::Item(item),
+        Ok(Some(Err(e))) => StreamRead::Failed(e.to_string()),
+        Ok(None) => StreamRead::End,
+        Err(_elapsed) => StreamRead::Idle,
+    }
+}
+
 /// Await the next stream item, bounded by `idle`. Maps a stalled stream (no
 /// item within `idle`) and any transport error to a **retryable**
 /// `ProviderError::Transport`, and a clean end-of-stream to `Ok(None)`.
-///
-/// `idle` is a parameter (rather than reading [`STREAM_IDLE_TIMEOUT`]
-/// directly) purely so the timeout path is unit-testable in milliseconds;
-/// adapters always pass [`STREAM_IDLE_TIMEOUT`].
+/// A thin classification over [`next_stream_read`] for the adapters that
+/// don't need to tell the two failure shapes apart.
 pub(crate) async fn next_with_timeout<S, T>(
     stream: &mut S,
     idle: Duration,
 ) -> Result<Option<T>, ProviderError>
 where
     S: Stream<Item = reqwest::Result<T>> + Unpin,
 {
-    match tokio::time::timeout(idle, stream.next()).await {
-        Ok(Some(Ok(item))) => Ok(Some(item)),
-        Ok(Some(Err(e))) => Err(ProviderError::transport(e.to_string())),
-        Ok(None) => Ok(None),
-        Err(_elapsed) => Err(ProviderError::transport(format!(
+    match next_stream_read(stream, idle).await {
+        StreamRead::Item(item) => Ok(Some(item)),
+        StreamRead::End => Ok(None),
+        StreamRead::Failed(message) => Err(ProviderError::transport(message)),
+        StreamRead::Idle => Err(ProviderError::transport(format!(
             "stream idle timeout: no data for {}s",
             idle.as_secs()
         ))),

crates/stella-model/src/lib.rs

@@ -34,6 +34,9 @@
 //!   provider-native `/models` discovery, both fetch-and-parse only.
 //! - [`sse`] — the shared, dependency-free SSE + incremental UTF-8 decoder
 //!   every streaming adapter feeds.
+//! - `stream_recovery` (crate-private) — the streaming→non-streaming
+//!   fallback latch armed when a stream hangs before its first byte or
+//!   comes back empty (#2686).
 pub mod anthropic;
 pub(crate) mod attachment;
 pub mod bedrock;
@@ -49,6 +52,7 @@ pub mod provider;
 pub mod provider_listing;
 pub mod provider_parity;
 pub mod sse;
+pub(crate) mod stream_recovery;
 pub mod vertex;
 pub mod zai;