diff --git a/CHANGELOG.md b/CHANGELOG.md
index 10122650..54e11b5d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,133 @@ target.
 
 ## [Unreleased]
 
+## [3.3.1] — 2026-10-10
+
+Streams are logged as they ended. A caller that hangs up once it has read
+the whole answer, as Codex does, finished the request; an error the
+upstream or Bedrock reports partway through a stream is logged with the
+status the same error has as an answer, and counts against the route only
+when that answer would; a Responses WebSocket turn ends at its last event.
+The thinkwatch-core crates move from v0.67.1 to v0.68.0.
+
+### Read before upgrading
+
+- **Request logs and error rates change for streams.** Codex streams that
+  were logged as cancelled (`499`) are logged as finished (`200`) and
+  billed on the reported usage. Streams the upstream failed partway are
+  logged with a `4xx` or `5xx` `status_code` instead of `200`, so error
+  rates that count `status_code` ≥ 400 include them; `client_status`
+  keeps the `200` the caller received. No configuration, database schema,
+  API or Helm value changes.
+
+### Fixed
+
+- **A caller that hangs up on the answer's last event finished the
+  request.** Codex closes the connection as soon as it has read
+  `response.completed`, without waiting for the stream to end, and an
+  upstream can end its stream a while after that event: the ChatGPT Codex
+  backend often does, and so can a relay in front of it. A stream forwarded
+  in its own format whose caller left in that window was logged as
+  cancelled (`499`, `stream_outcome: client_cancelled`), though the caller
+  had read the whole answer. Once the answer's last event has gone out —
+  Responses `response.completed`, `response.incomplete` or
+  `response.failed`, Anthropic `message_stop`, Chat `[DONE]` — a caller
+  that leaves is logged as if the stream had run to its end: `200`, billed
+  on the usage the upstream reported, or, after `response.failed`, the
+  upstream's failure (see the next entry). Converted streams were not
+  affected: their last event is written once the upstream's stream has
+  ended.
+- **An error the upstream reports partway through a stream fails the
+  request.** An upstream that has answered `200` can still report an error
+  in the stream: an Anthropic `error` event (an `overloaded_error` arrives
+  this way), Responses `response.failed`, an `error` in a Chat or Gemini
+  chunk. When the stream then ended normally, the request was logged as a
+  success (`200`) and counted as one by the route's circuit breaker. It is
+  now logged as the error it would have been as an answer, with what the
+  upstream said: `stream_outcome` `upstream_error`, `client_status` `200`
+  (the status the caller's response went out with), and the `status_code`
+  and `error_type` that answer would have had, read off the error's own
+  type or code. A rate limit or a spent quota is `429`
+  (`UpstreamRateLimited`); a context too long or another invalid request
+  `400`, an unknown model `404`, an overload `529` or `503`
+  (`ProviderHttpError`); an error that names no status `502`
+  (`ProviderError`). The route's health and circuit breaker treat it as
+  they treat that answer: a `5xx`, a `429` or a refused credential counts
+  against the route, so a route whose streams keep failing this way opens
+  its breaker, while a request the upstream refuses, such as one past the
+  model's context window, does not. This holds for streams forwarded as
+  sent and converted ones, and for a caller that leaves after the error.
+  Billing is unchanged: such a request is billed on the usage the upstream
+  reported.
+  - Error rates that count `status_code` ≥ 400 now include these requests.
+  - The rows of other streams that failed after the response went out as
+    `200` — broken off, cut by tool-call inspection, refused by the
+    upstream before its first byte — carry `client_status` too.
+- **An exception Bedrock reports partway through a stream is logged as
+  the same exception as an answer.** Bedrock reports a failure in a
+  stream it has started as an exception event: `throttlingException` when
+  it throttles partway through, `validationException`,
+  `internalServerException`, `serviceUnavailableException` and others.
+  Each was logged as the stream breaking off in transit (`502`,
+  `error_type` `transport`) and counted against the route's circuit
+  breaker. Each is now logged with the status the desktop gateway gives
+  that exception and with what Bedrock said: throttling `429`
+  (`UpstreamRateLimited`), an invalid request `400`, a model timeout
+  `504`, the service unavailable `503`, any other exception `500`
+  (`ProviderHttpError`). Access denied is logged as a refused credential
+  (`UpstreamAuthError`), without Bedrock's words, which name the IAM
+  principal, as for a `403` answer. The route's health and circuit breaker treat it
+  as they treat that answer (see the previous entry), so a request Bedrock
+  refuses as invalid no longer counts against the route. What the caller
+  receives is unchanged, and a damaged event-stream frame is still logged
+  as `transport` `502`.
+- **A Responses WebSocket turn ends at its last event.** On
+  `GET /v1/responses` over a WebSocket, a turn lasted until the upstream's
+  stream ended, and the connection took the next turn only then, so an
+  upstream that keeps its stream open after `response.completed` held up
+  every following turn by that long. The next turn now starts at once.
+
+### Changed
+
+- thinkwatch-core crates (tw-bedrock, tw-breaker, tw-dialect, tw-guard)
+  v0.67.1 → v0.68.0. tw-dialect changes what the gateway does, tw-guard
+  only adds to its API, and tw-bedrock and tw-breaker do not change:
+  - **Long usage objects are read** (tw-dialect). The usage reader skipped
+    a usage object longer than 8 KB. The ChatGPT Codex backend's usage
+    carries an `attribution` breakdown that grows with the conversation, so
+    behind an upstream that passes it on, a longer Codex session's requests
+    were billed on an estimate (`usage_estimated`) instead of the reported
+    counts. Usage objects up to 1 MiB are now read. One cut across chunks
+    is read on from where it stopped, nothing nested inside a usage object
+    is counted as a second usage, and a stream that arrives in chunks
+    shorter than the `"usage"` key no longer loses it.
+  - **A `null` error in a stream chunk is not an error** (tw-dialect).
+    Some OpenAI-compatible relays and Gemini-format upstreams write
+    `"error": null` in every streamed chunk, and the Chat and Gemini stream
+    readers took each such chunk for an error the upstream reported. A
+    stream converted to another format ended in an error and lost its
+    text, and for such a stream forwarded as sent, no assembled answer was
+    cached or captured as the response body. A Gemini caller's tool result
+    carrying `"error": null` and no `output` went to an upstream of
+    another format as a failed one. A `null` error is now ignored, and a
+    Chat chunk's error without a message reaches the caller of a converted
+    stream as a fixed sentence instead of the whole chunk, which could hold
+    answer text. The gateway's own check for an error partway through a
+    stream (see Fixed) ignores a `null` error as well.
+  - `tw_dialect::convert::ends_answer` recognises an answer's last event.
+    The gateway uses it to tell a caller that left after the whole answer
+    from one that left partway, and to end a WebSocket turn (see Fixed).
+  - tw-guard adds, for the desktop gateway's security log, the `locate`
+    module, `content::places`, `content::places_text` and `RulePlaces`,
+    `content::Rule::draw`, `Ledger::placeholder_of`,
+    `RuleSet::custom_pattern`, `tools::Rule::find_all`, and the fields
+    `pattern`, `arguments`, `capped`, `places` and `path` on
+    `tools::wall::Verdict`. Secret redaction, the content filter and
+    tool-call inspection find, replace, refuse and cut exactly what they
+    did before. `Verdict::arguments` is a flagged tool call's arguments
+    with redacted values restored; the gateway does not log it, and its
+    tool-call audit entries keep the fields they had.
+
 ## [3.3.0] — 2026-10-09
 
 Requests converted for Claude now use prompt caching, and Codex works
@@ -1569,7 +1696,8 @@ unreleased builds should: stop the gateway, run `db/schema.sql`
 against PostgreSQL, restart against this tag. The schema is
 idempotent end-to-end, so the apply is safe to repeat.
 
-[Unreleased]: https://github.com/ThinkWatchProject/ThinkWatch/compare/v3.3.0...HEAD
+[Unreleased]: https://github.com/ThinkWatchProject/ThinkWatch/compare/v3.3.1...HEAD
+[3.3.1]: https://github.com/ThinkWatchProject/ThinkWatch/releases/tag/v3.3.1
 [3.3.0]: https://github.com/ThinkWatchProject/ThinkWatch/releases/tag/v3.3.0
 [3.2.1]: https://github.com/ThinkWatchProject/ThinkWatch/releases/tag/v3.2.1
 [3.2.0]: https://github.com/ThinkWatchProject/ThinkWatch/releases/tag/v3.2.0
diff --git a/Cargo.lock b/Cargo.lock
index 59bef983..61fbf483 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4189,7 +4189,7 @@ dependencies = [
 
 [[package]]
 name = "think-watch-auth"
-version = "3.3.0"
+version = "3.3.1"
 dependencies = [
  "anyhow",
  "argon2",
@@ -4219,7 +4219,7 @@ dependencies = [
 
 [[package]]
 name = "think-watch-common"
-version = "3.3.0"
+version = "3.3.1"
 dependencies = [
  "aes-gcm",
  "anyhow",
@@ -4262,7 +4262,7 @@ dependencies = [
 
 [[package]]
 name = "think-watch-gateway"
-version = "3.3.0"
+version = "3.3.1"
 dependencies = [
  "anyhow",
  "arc-swap",
@@ -4301,7 +4301,7 @@ dependencies = [
 
 [[package]]
 name = "think-watch-mcp-gateway"
-version = "3.3.0"
+version = "3.3.1"
 dependencies = [
  "anyhow",
  "arc-swap",
@@ -4330,7 +4330,7 @@ dependencies = [
 
 [[package]]
 name = "think-watch-server"
-version = "3.3.0"
+version = "3.3.1"
 dependencies = [
  "anyhow",
  "arc-swap",
@@ -4381,7 +4381,7 @@ dependencies = [
 
 [[package]]
 name = "think-watch-test-support"
-version = "3.3.0"
+version = "3.3.1"
 dependencies = [
  "anyhow",
  "async-stream",
@@ -4837,8 +4837,8 @@ dependencies = [
 
 [[package]]
 name = "tw-bedrock"
-version = "0.67.1"
-source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.67.1#fb32ae9607771e77526d501106b2642498d64482"
+version = "0.68.0"
+source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.68.0#0955f9906c88f9c8fba99a17e3898efa23c952b9"
 dependencies = [
  "aws-credential-types",
  "aws-sigv4",
@@ -4852,16 +4852,16 @@ dependencies = [
 
 [[package]]
 name = "tw-breaker"
-version = "0.67.1"
-source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.67.1#fb32ae9607771e77526d501106b2642498d64482"
+version = "0.68.0"
+source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.68.0#0955f9906c88f9c8fba99a17e3898efa23c952b9"
 dependencies = [
  "serde",
 ]
 
 [[package]]
 name = "tw-dialect"
-version = "0.67.1"
-source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.67.1#fb32ae9607771e77526d501106b2642498d64482"
+version = "0.68.0"
+source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.68.0#0955f9906c88f9c8fba99a17e3898efa23c952b9"
 dependencies = [
  "memchr",
  "serde",
@@ -4870,8 +4870,8 @@ dependencies = [
 
 [[package]]
 name = "tw-guard"
-version = "0.67.1"
-source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.67.1#fb32ae9607771e77526d501106b2642498d64482"
+version = "0.68.0"
+source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.68.0#0955f9906c88f9c8fba99a17e3898efa23c952b9"
 dependencies = [
  "base64 0.22.1",
  "bytes",
diff --git a/Cargo.toml b/Cargo.toml
index eb1e48c9..67ed8a67 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -10,7 +10,7 @@ members = [
 ]
 
 [workspace.package]
-version = "3.3.0"
+version = "3.3.1"
 edition = "2024"
 # The MSRV: the newest `rust-version` among the locked dependencies
 # (sqlx 0.9 and the aws-smithy crates under aws-sigv4). Without it,
@@ -70,10 +70,10 @@ opt-level = 3
 # never re-exported through a local shim. And the reverse: something only
 # this side uses (the at-rest crypto, IMDSv2 credentials, the gateway error)
 # lives here, not in core.
-tw-bedrock = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.67.1" }
-tw-breaker = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.67.1" }
-tw-dialect = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.67.1" }
-tw-guard = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.67.1" }
+tw-bedrock = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.68.0" }
+tw-breaker = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.68.0" }
+tw-dialect = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.68.0" }
+tw-guard = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.68.0" }
 
 # Web framework
 axum = { version = "0.8", features = ["macros", "ws"] }
diff --git a/crates/gateway/src/lifecycle/mod.rs b/crates/gateway/src/lifecycle/mod.rs
index 0b72fb79..8ebda1be 100644
--- a/crates/gateway/src/lifecycle/mod.rs
+++ b/crates/gateway/src/lifecycle/mod.rs
@@ -18,6 +18,7 @@
 //! - `emit_audit` → `prepare_body_capture` + `emit_gateway_log_with_extra`.
 
 use std::pin::Pin;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 
 use crate::error::GatewayError;
@@ -159,7 +160,27 @@ pub(crate) type OpenUpstream = Pin<
 ///
 /// **A dropped stream is a cancelled request.** When the client goes,
 /// hyper drops the body, and with it the sender the tail is waiting on;
-/// the tail then records `ClientCancelled`.
+/// the tail then records `ClientCancelled` — unless the answer's last
+/// frame had already gone out (see [`LastFrame`]). A client that leaves
+/// after that leaves having read the whole answer: Codex closes the
+/// connection as soon as it has `response.completed`, and an upstream can
+/// end its stream a while after that frame. The request finished, and is
+/// recorded as `Natural`.
+///
+/// For the same reason a stream that ends here tells its outcome before
+/// its last bytes go out, not after: a consumer that stops reading at
+/// those bytes, as the WebSocket relay does at a turn's last event, never
+/// polls the stream again.
+///
+/// **An error the upstream reports in its stream fails the request** (see
+/// [`ErrorWatch`]): an Anthropic `error` event, Responses
+/// `response.failed`, an `error` in a Chat or Gemini chunk. The upstream
+/// opened the stream with 200 and the stream may end normally, but the
+/// client got half an answer and an error; recorded as a success, the
+/// route would look like it never fails. It is recorded as the error the
+/// same refusal would have been as an answer, with the upstream's words
+/// (see [`failed_partway`]), whether the stream ran to its end or the
+/// client left after the error.
 #[allow(clippy::too_many_arguments)]
 pub(crate) fn build_chat_pump(
     open: OpenUpstream,
@@ -174,17 +195,13 @@ pub(crate) fn build_chat_pump(
     axum::response::Response,
     Pin<Box<dyn std::future::Future<Output = Invoked<ChatCompletionSurface>> + Send>>,
 ) {
-    struct Readers {
-        sniffer: Option<tw_dialect::usage::Sniffer>,
-        collector: Option<tw_dialect::convert::Collector>,
-    }
-    let readers = Arc::new(Mutex::new(Readers {
-        sniffer: None,
-        collector: None,
-    }));
+    let readers = Arc::new(Mutex::new(Readers::default()));
     let readers_for_tail = Arc::clone(&readers);
 
     let (done_tx, done_rx) = tokio::sync::oneshot::channel::<StreamOutcome>();
+    // Set once the answer's last frame is handed to the client.
+    let delivered = Arc::new(AtomicBool::new(false));
+    let delivered_for_tail = Arc::clone(&delivered);
 
     // Tool calls are inspected on what the client is about to receive —
     // converted, if it was, and with redacted values restored — since that
@@ -198,6 +215,9 @@ pub(crate) fn build_chat_pump(
         provider.to_string(),
     );
 
+    let mut last_frame = LastFrame::new(client, client_sse);
+    let provider_for_tail = provider.to_string();
+    let provider = provider.to_string();
     let body = async_stream::stream! {
         let mut done_tx = Some(done_tx);
 
@@ -209,20 +229,19 @@ pub(crate) fn build_chat_pump(
                 // so a throttled upstream stays 429 on the audit row.
                 let mut out = shaper.process(&error_frame(client, e.status_code(), &e.to_string()));
                 out.extend(shaper.finish());
+                tell(&mut done_tx, StreamOutcome::UpstreamError {
+                    error_type: e.error_tag().to_string(),
+                    message: e.to_string(),
+                    status_code: e.status_code(),
+                });
                 yield Ok::<Bytes, std::convert::Infallible>(Bytes::from(out));
-                if let Some(tx) = done_tx.take() {
-                    let _ = tx.send(StreamOutcome::UpstreamError {
-                        error_type: e.error_tag().to_string(),
-                        message: e.to_string(),
-                        status_code: e.status_code(),
-                    });
-                }
                 return;
             }
         };
         if let Ok(mut r) = readers.lock() {
             r.sniffer = Some(tw_dialect::usage::Sniffer::new());
             r.collector = Some(wire.collect.collector());
+            r.errors = Some(ErrorWatch::new(wire.dialect));
         }
         // The hop that answered can have numbered a value of its own.
         shaper.restore_with(&wire.ledger);
@@ -240,15 +259,14 @@ pub(crate) fn build_chat_pump(
                     Some(t) => t
                         .feed(&raw)
                         .map(Bytes::from)
-                        .map_err(|e| format!("Bedrock ended the stream: {e}")),
+                        .map_err(|e| bedrock_ended(&provider, e)),
                 },
-                Err(e) => Err(format!("The upstream stream broke off: {e}")),
+                Err(e) => Err(broke_off(format!("The upstream stream broke off: {e}"))),
             };
             match item {
                 Ok(chunk) => {
                     if let Ok(mut r) = readers.lock() {
-                        if let Some(s) = r.sniffer.as_mut() { s.feed(&chunk); }
-                        if let Some(c) = r.collector.as_mut() { c.process(&chunk); }
+                        r.feed(&chunk);
                     }
                     let client_bytes = match convert.as_mut() {
                         Some(c) => c.process(&chunk),
@@ -259,38 +277,33 @@ pub(crate) fn build_chat_pump(
                     // still goes out, then the refusal.
                     let stop = inspector.as_mut().and_then(|i| i.check(&out));
                     if let Some((err, safe)) = stop {
-                        yield Ok(Bytes::from(cut(&shaper, convert.as_mut(), client, &out[..safe], &err)));
-                        if let Some(tx) = done_tx.take() {
-                            let _ = tx.send(StreamOutcome::UpstreamError {
-                                error_type: err.error_tag().to_string(),
-                                message: err.to_string(),
-                                status_code: err.status_code(),
-                            });
-                        }
+                        let out = cut(&shaper, convert.as_mut(), client, &out[..safe], &err);
+                        tell(&mut done_tx, StreamOutcome::UpstreamError {
+                            error_type: err.error_tag().to_string(),
+                            message: err.to_string(),
+                            status_code: err.status_code(),
+                        });
+                        yield Ok(Bytes::from(out));
                         return;
                     }
                     if !out.is_empty() {
+                        if last_frame.is_in(&out) {
+                            delivered.store(true, Ordering::Release);
+                        }
                         yield Ok(Bytes::from(out));
                     }
                 }
-                Err(message) => {
+                Err((message, outcome)) => {
                     // Headers are gone; the only way left to say it is in
                     // the stream, in the caller's own format.
-                    tracing::warn!("{message}");
                     let tail = match convert.as_mut() {
                         Some(c) => c.fail(&message),
                         None => error_frame(client, 502, &message),
                     };
                     let mut out = shaper.process(&tail);
                     out.extend(shaper.finish());
+                    tell(&mut done_tx, outcome);
                     yield Ok(Bytes::from(out));
-                    if let Some(tx) = done_tx.take() {
-                        let _ = tx.send(StreamOutcome::UpstreamError {
-                            error_type: "transport".into(),
-                            message,
-                            status_code: 502,
-                        });
-                    }
                     return;
                 }
             }
@@ -302,22 +315,23 @@ pub(crate) fn build_chat_pump(
         // stop), so they are inspected too.
         let stop = inspector.as_mut().and_then(|i| i.check(&out));
         if let Some((err, safe)) = stop {
-            yield Ok(Bytes::from(cut(&shaper, None, client, &out[..safe], &err)));
-            if let Some(tx) = done_tx.take() {
-                let _ = tx.send(StreamOutcome::UpstreamError {
-                    error_type: err.error_tag().to_string(),
-                    message: err.to_string(),
-                    status_code: err.status_code(),
-                });
-            }
+            let out = cut(&shaper, None, client, &out[..safe], &err);
+            tell(&mut done_tx, StreamOutcome::UpstreamError {
+                error_type: err.error_tag().to_string(),
+                message: err.to_string(),
+                status_code: err.status_code(),
+            });
+            yield Ok(Bytes::from(out));
             return;
         }
+        let outcome = match readers.lock().ok().and_then(|mut r| r.upstream_error()) {
+            Some(said) => failed_partway(&provider, &said),
+            None => StreamOutcome::Natural,
+        };
+        tell(&mut done_tx, outcome);
         if !out.is_empty() {
             yield Ok(Bytes::from(out));
         }
-        if let Some(tx) = done_tx.take() {
-            let _ = tx.send(StreamOutcome::Natural);
-        }
     };
 
     // A Gemini caller that did not ask for SSE reads one JSON array.
@@ -338,7 +352,23 @@ pub(crate) fn build_chat_pump(
     let input_estimate = request.input_estimate;
 
     let tail = Box::pin(async move {
-        let outcome = done_rx.await.unwrap_or(StreamOutcome::ClientCancelled);
+        let outcome = match done_rx.await {
+            Ok(outcome) => outcome,
+            // Dropped before the stream ended: the client left.
+            Err(_) => match readers_for_tail
+                .lock()
+                .ok()
+                .and_then(|mut r| r.upstream_error())
+            {
+                // After the upstream had failed, which the client leaving
+                // does not change.
+                Some(said) => failed_partway(&provider_for_tail, &said),
+                // After the answer's last frame went out: the client left
+                // having read all of it.
+                None if delivered_for_tail.load(Ordering::Acquire) => StreamOutcome::Natural,
+                None => StreamOutcome::ClientCancelled,
+            },
+        };
         metrics::counter!(
             "gateway_stream_completion_total",
             "outcome" => outcome.metric_label()
@@ -400,6 +430,333 @@ pub(crate) fn build_chat_pump(
     (response, tail)
 }
 
+/// What the pump reads off the upstream's bytes as they pass, for the
+/// tail. Each reader exists once the upstream answered.
+#[derive(Default)]
+struct Readers {
+    sniffer: Option<tw_dialect::usage::Sniffer>,
+    collector: Option<tw_dialect::convert::Collector>,
+    /// `None` once it found an error.
+    errors: Option<ErrorWatch>,
+    /// The first error the upstream reported in its stream.
+    said: Option<Said>,
+}
+
+impl Readers {
+    fn feed(&mut self, chunk: &[u8]) {
+        if let Some(s) = self.sniffer.as_mut() {
+            s.feed(chunk);
+        }
+        if let Some(c) = self.collector.as_mut() {
+            c.process(chunk);
+        }
+        if let Some(said) = self.errors.as_mut().and_then(|w| w.feed(chunk)) {
+            self.said = Some(said);
+            self.errors = None;
+        }
+    }
+
+    /// The error the upstream reported in its stream, if it did — read to
+    /// the end, for an upstream whose last frame has no blank line after
+    /// it.
+    fn upstream_error(&mut self) -> Option<Said> {
+        if let Some(mut w) = self.errors.take() {
+            self.said = self.said.take().or_else(|| w.flush());
+        }
+        self.said.take()
+    }
+}
+
+/// Watches the upstream's frames for an error it reports in its stream
+/// (see [`said_in`]: an Anthropic `error` event, Responses
+/// `response.failed`, an `error` in a Chat or Gemini chunk). Read on the
+/// upstream's bytes, so a converted stream is watched the same way as one
+/// forwarded as sent.
+struct ErrorWatch {
+    upstream: Dialect,
+    frames: tw_dialect::frame::Decoder,
+}
+
+impl ErrorWatch {
+    fn new(upstream: Dialect) -> Self {
+        Self {
+            upstream,
+            frames: Default::default(),
+        }
+    }
+
+    /// What the upstream said, if this chunk completes an error frame.
+    fn feed(&mut self, chunk: &[u8]) -> Option<Said> {
+        let frames = self.frames.feed(chunk);
+        self.first(&frames)
+    }
+
+    /// The stream ended: a last frame without a blank line after it.
+    fn flush(&mut self) -> Option<Said> {
+        let frames = self.frames.flush();
+        self.first(&frames)
+    }
+
+    fn first(&self, frames: &[tw_dialect::frame::Frame]) -> Option<Said> {
+        frames.iter().find_map(|f| said_in(self.upstream, f))
+    }
+}
+
+/// An error the upstream reported in its stream.
+#[derive(Debug, PartialEq)]
+struct Said {
+    /// In its words, or [`NO_MESSAGE`] when the error carried none.
+    message: String,
+    /// The status the same error has as an answer, when the error names
+    /// one (see [`status_of`]).
+    status: Option<u16>,
+}
+
+/// The message of an error the upstream reported without one.
+const NO_MESSAGE: &str = "The error carried no message.";
+
+/// What the upstream said in `f`, if `f` is an error frame.
+///
+/// `tw_dialect::convert::stream_error` picks the frame: the same reading
+/// the converter gives it, and most frames are passed over without being
+/// parsed. The error itself is read here, where each format keeps it:
+/// Anthropic's `error`; Responses' `response.error` in `response.failed`,
+/// or the `error` event's own fields (or its `error`); the top-level
+/// `error` of a Chat or Gemini chunk. Gemini sends an error as a
+/// one-element array outside SSE; upstreams are asked for SSE, but a
+/// frame like that is read the same.
+///
+/// A Chat or Gemini chunk whose `error` is `null` is not an error: an
+/// OpenAI-compatible relay can send `"error": null` in every chunk. And
+/// the message is never the frame's own text, which can be the model's
+/// answer: an error without a message gets [`NO_MESSAGE`].
+fn said_in(upstream: Dialect, f: &tw_dialect::frame::Frame) -> Option<Said> {
+    use serde_json::Value;
+
+    tw_dialect::convert::stream_error(upstream, f)?;
+    let v = match serde_json::from_str::<Value>(f.data.trim()) {
+        Ok(Value::Array(mut items)) if items.len() == 1 => items.remove(0),
+        Ok(v) => v,
+        Err(_) => Value::Null,
+    };
+    fn present(e: Option<&Value>) -> Option<&Value> {
+        e.filter(|e| !e.is_null())
+    }
+    let error = match upstream {
+        Dialect::Chat | Dialect::Gemini => Some(present(v.get("error"))?),
+        Dialect::Responses => {
+            let kind = v.get("type").and_then(Value::as_str).or(f.event.as_deref());
+            if kind == Some("response.failed") {
+                present(v.pointer("/response/error"))
+            } else {
+                Some(present(v.get("error")).unwrap_or(&v))
+            }
+        }
+        Dialect::Anthropic => Some(present(v.get("error")).unwrap_or(&v)),
+        Dialect::Bedrock => Some(&v),
+    };
+    let message = match error {
+        Some(Value::String(s)) => Some(s.as_str()),
+        Some(e) => e.get("message").and_then(Value::as_str),
+        None => None,
+    }
+    .map(str::trim)
+    .filter(|m| !m.is_empty())
+    .unwrap_or(NO_MESSAGE);
+    Some(Said {
+        message: message.to_string(),
+        status: error.and_then(status_of),
+    })
+}
+
+/// The status an error the upstream reports in its stream has as an
+/// answer: a numeric `code` from 400 to 599 (Gemini's, and some relays'),
+/// or what its `code`, `type` or `status` names (see [`status_named`]).
+/// `None` when it says nothing recognisable.
+fn status_of(e: &serde_json::Value) -> Option<u16> {
+    use serde_json::Value;
+
+    let numeric = match e.get("code") {
+        Some(Value::Number(n)) => n.as_u64().and_then(|n| u16::try_from(n).ok()),
+        Some(Value::String(s)) => s.trim().parse::<u16>().ok(),
+        _ => None,
+    };
+    numeric.filter(|c| (400..600).contains(c)).or_else(|| {
+        ["code", "type", "status"]
+            .iter()
+            .find_map(|key| e.get(*key)?.as_str().and_then(status_named))
+    })
+}
+
+/// The status each format documents for an error it names: Anthropic's
+/// error types, OpenAI's error types and codes, Gemini's statuses — the
+/// statuses the same errors come with as an answer. An overload is 529 as
+/// Anthropic sends it, or 503.
+fn status_named(name: &str) -> Option<u16> {
+    Some(match name {
+        "invalid_request_error"
+        | "context_length_exceeded"
+        | "invalid_prompt"
+        | "INVALID_ARGUMENT"
+        | "FAILED_PRECONDITION" => 400,
+        "authentication_error" | "invalid_api_key" | "UNAUTHENTICATED" => 401,
+        "billing_error" => 402,
+        "permission_error" | "PERMISSION_DENIED" => 403,
+        "not_found_error" | "model_not_found" | "NOT_FOUND" => 404,
+        "request_too_large" => 413,
+        "rate_limit_error"
+        | "rate_limit_exceeded"
+        | "insufficient_quota"
+        | "usage_limit_reached"
+        | "RESOURCE_EXHAUSTED" => 429,
+        "api_error" => 500,
+        "server_is_overloaded" | "slow_down" | "UNAVAILABLE" => 503,
+        "timeout_error" | "DEADLINE_EXCEEDED" => 504,
+        "overloaded_error" => 529,
+        _ => return None,
+    })
+}
+
+/// The outcome of a stream the upstream reported an error in: the error
+/// the same refusal would have been as an answer (see
+/// `transport::status_error`), so the route's health and circuit breaker
+/// treat it as they treat that answer — a 5xx, 408 or 429 counts against
+/// the route, a request the upstream refuses (another 4xx) does not.
+/// An error that names no status is the upstream's failure, 502.
+///
+/// The message carries what the upstream said, 500 characters at most as
+/// a refusal's body is cut — except for a refused credential, whose
+/// words stay out of the row as an answer's do: they can name the account
+/// behind it. The log has them either way.
+fn failed_partway(provider: &str, said: &Said) -> StreamOutcome {
+    tracing::warn!(
+        provider,
+        status = ?said.status,
+        message = %said.message,
+        "upstream reported an error partway through a stream"
+    );
+    let words: String = said.message.chars().take(500).collect();
+    let what = format!("{provider} reported an error partway through the answer: {words}");
+    let e = match said.status {
+        Some(status) => crate::proxy::transport::status_error(status, None, what.clone()),
+        None => GatewayError::ProviderError(what.clone()),
+    };
+    let message = match &e {
+        GatewayError::UpstreamRateLimited { .. } => format!("{e}: {what}"),
+        GatewayError::UpstreamAuthError { .. } => {
+            format!("{e}: {provider} reported it partway through the answer")
+        }
+        _ => e.to_string(),
+    };
+    StreamOutcome::UpstreamError {
+        error_type: e.error_tag().to_string(),
+        message,
+        status_code: e.status_code(),
+    }
+}
+
+/// A stream that broke off in transit: what the client is told, and the
+/// outcome. No usable answer arrived — the upstream's failure, 502.
+fn broke_off(message: String) -> (String, StreamOutcome) {
+    tracing::warn!("{message}");
+    let outcome = StreamOutcome::UpstreamError {
+        error_type: "transport".into(),
+        message: message.clone(),
+        status_code: 502,
+    };
+    (message, outcome)
+}
+
+/// Bedrock's event stream ended in an error: what the client is told —
+/// the same whichever it was — and the outcome.
+///
+/// A damaged frame broke the stream off. An exception Bedrock reported
+/// in the stream (a `throttlingException` partway through, say) is an
+/// error the upstream reported, recorded like the others (see
+/// [`failed_partway`]) with the status [`bedrock_status`] gives it.
+fn bedrock_ended(
+    provider: &str,
+    e: tw_bedrock::eventstream::StreamError,
+) -> (String, StreamOutcome) {
+    use tw_bedrock::eventstream::StreamError;
+
+    let told = format!("Bedrock ended the stream: {e}");
+    match e {
+        StreamError::Malformed(_) => broke_off(told),
+        StreamError::Upstream { kind, message } => {
+            let said = Said {
+                message: format!("{kind}: {message}"),
+                status: Some(bedrock_status(&kind)),
+            };
+            (told, failed_partway(provider, &said))
+        }
+    }
+}
+
+/// The status of an exception Bedrock reports in its event stream, as the
+/// desktop gateway reads one that arrives in a stream: throttling 429,
+/// an invalid request 400, access denied 403, a model timeout 504, the
+/// service unavailable 503, and any other exception (an internal error,
+/// a model stream error) 500. Bedrock names them in camel case
+/// (`throttlingException`); any case is read.
+fn bedrock_status(kind: &str) -> u16 {
+    const STATUSES: [(&str, u16); 5] = [
+        ("throttlingException", 429),
+        ("serviceUnavailableException", 503),
+        ("validationException", 400),
+        ("accessDeniedException", 403),
+        ("modelTimeoutException", 504),
+    ];
+    STATUSES
+        .iter()
+        .find(|(name, _)| name.eq_ignore_ascii_case(kind))
+        .map_or(500, |(_, status)| *status)
+}
+
+/// Tell the tail how the stream ended. Only the first outcome counts.
+fn tell(done_tx: &mut Option<tokio::sync::oneshot::Sender<StreamOutcome>>, outcome: StreamOutcome) {
+    if let Some(tx) = done_tx.take() {
+        let _ = tx.send(outcome);
+    }
+}
+
+/// Watches what the client receives for the answer's last frame
+/// (`tw_dialect::convert::ends_answer`): Responses `response.completed`,
+/// `response.incomplete` or `response.failed`, Anthropic `message_stop`,
+/// Chat `[DONE]`. A Gemini answer has none, SSE or JSON array: it ends
+/// with the stream.
+struct LastFrame {
+    client: Dialect,
+    /// `None` once the frame was seen, and for an answer without one.
+    frames: Option<tw_dialect::frame::Decoder>,
+}
+
+impl LastFrame {
+    fn new(client: Dialect, client_sse: bool) -> Self {
+        let has_one = client_sse && !matches!(client, Dialect::Gemini | Dialect::Bedrock);
+        Self {
+            client,
+            frames: has_one.then(Default::default),
+        }
+    }
+
+    /// Whether `out`, the next bytes the client receives, completes the
+    /// last frame. True once; nothing is read after it.
+    fn is_in(&mut self, out: &[u8]) -> bool {
+        let Some(frames) = self.frames.as_mut() else {
+            return false;
+        };
+        let found = frames
+            .feed(out)
+            .iter()
+            .any(|f| tw_dialect::convert::ends_answer(self.client, f));
+        if found {
+            self.frames = None;
+        }
+        found
+    }
+}
+
 /// Reframe a client-format SSE stream as Gemini's JSON-array stream (see
 /// [`crate::proxy::shaper::JsonArrayFramer`]).
 fn as_json_array(
@@ -585,7 +942,15 @@ impl Surface for ChatCompletionSurface {
         let (prompt_tokens, completion_tokens) = tokens(&usage);
         let (response_body, cost, logged_status, error_detail) = match &invoked.view {
             CapturedView::Streaming { outcome, captured } => {
-                let (status, detail) = outcome.logged_status_and_detail();
+                let (status, mut detail) = outcome.logged_status_and_detail();
+                // The response went out as 200 before the stream failed;
+                // `status` is what the failure would have been as a
+                // response. Both are kept.
+                if let (StreamOutcome::UpstreamError { .. }, Some(serde_json::Value::Object(d))) =
+                    (outcome, detail.as_mut())
+                {
+                    d.insert("client_status".into(), 200.into());
+                }
                 (
                     captured.assembled.as_deref(),
                     captured.cost_usd,
@@ -692,3 +1057,352 @@ fn usage_estimated(view: &CapturedView<ChatCompletionSurface>) -> bool {
         CapturedView::Buffered(ChatCompletionOutcome::ShortCircuit(_)) => false,
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn the_last_frame_is_seen_in_the_chunk_that_completes_it() {
+        let mut w = LastFrame::new(Dialect::Responses, true);
+        // The model's text naming the event is not the event.
+        assert!(!w.is_in(
+            b"event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"delta\":\"response.completed\"}\n\n"
+        ));
+        assert!(!w.is_in(b"event: response.completed\ndata: {\"type\":\"response.com"));
+        assert!(w.is_in(b"pleted\",\"response\":{\"status\":\"completed\"}}\n\n"));
+        // Seen once.
+        assert!(!w.is_in(b"event: response.completed\ndata: {}\n\n"));
+
+        let mut w = LastFrame::new(Dialect::Anthropic, true);
+        assert!(!w.is_in(b"event: message_delta\ndata: {\"type\":\"message_delta\"}\n\n"));
+        assert!(w.is_in(b"event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n"));
+
+        let mut w = LastFrame::new(Dialect::Chat, true);
+        assert!(!w.is_in(b"data: {\"choices\":[]}\n\n"));
+        assert!(w.is_in(b"data: [DONE]\n\n"));
+    }
+
+    #[test]
+    fn an_error_the_upstream_reports_in_its_stream_is_found_in_its_words() {
+        let mut r = Readers {
+            errors: Some(ErrorWatch::new(Dialect::Responses)),
+            ..Default::default()
+        };
+        // The model's text naming an error is not one.
+        r.feed(b"event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"delta\":\"an \\\"error\\\" here\"}\n\n");
+        r.feed(b"event: response.failed\ndata: {\"type\":\"response.failed\",\"response\":{\"status\":\"failed\",");
+        assert!(r.said.is_none());
+        r.feed(b"\"error\":{\"code\":\"server_error\",\"message\":\"boom\"}}}\n\n");
+        assert_eq!(
+            r.upstream_error().map(|s| s.message).as_deref(),
+            Some("boom")
+        );
+
+        // An upstream whose last frame has no blank line after it.
+        let mut r = Readers {
+            errors: Some(ErrorWatch::new(Dialect::Anthropic)),
+            ..Default::default()
+        };
+        r.feed(b"event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"Overloaded\"}}");
+        assert_eq!(
+            r.upstream_error().map(|s| s.message).as_deref(),
+            Some("Overloaded")
+        );
+
+        // A stream without one.
+        let mut r = Readers {
+            errors: Some(ErrorWatch::new(Dialect::Chat)),
+            ..Default::default()
+        };
+        r.feed(b"data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\ndata: [DONE]\n\n");
+        assert_eq!(r.upstream_error(), None);
+    }
+
+    /// What a stream whose upstream sent `frames` is recorded as, with
+    /// whether it counts against the route: `None` when it carried no
+    /// error.
+    fn recorded(upstream: Dialect, frames: &str) -> Option<(String, i64, String, bool)> {
+        let mut r = Readers {
+            errors: Some(ErrorWatch::new(upstream)),
+            ..Default::default()
+        };
+        r.feed(frames.as_bytes());
+        let said = r.upstream_error()?;
+        let StreamOutcome::UpstreamError {
+            error_type,
+            message,
+            status_code,
+        } = failed_partway("up", &said)
+        else {
+            panic!("not a failure");
+        };
+        let counted = crate::proxy::upstream_failed(&error_type, status_code);
+        Some((error_type, status_code, message, counted))
+    }
+
+    fn responses_failed(error: serde_json::Value) -> String {
+        let v = serde_json::json!({
+            "type": "response.failed",
+            "response": {"id": "resp_1", "status": "failed", "error": error},
+        });
+        format!("event: response.failed\ndata: {v}\n\n")
+    }
+
+    /// Throttling stays a 429, and counts against the route like a 429
+    /// answer does: that upstream's quota, which another route does not
+    /// share.
+    #[test]
+    fn a_rate_limit_in_a_stream_is_a_429() {
+        let (error_type, status, message, counted) = recorded(
+            Dialect::Responses,
+            &responses_failed(serde_json::json!({
+                "code": "rate_limit_exceeded",
+                "message": "Rate limit reached for requests",
+            })),
+        )
+        .unwrap();
+        assert_eq!((error_type.as_str(), status), ("UpstreamRateLimited", 429));
+        assert!(counted);
+        assert!(
+            message.contains("Rate limit reached for requests"),
+            "{message}"
+        );
+
+        // Gemini names it by number and by status.
+        let gemini = "data: {\"error\":{\"code\":429,\"message\":\"Resource has been exhausted\",\"status\":\"RESOURCE_EXHAUSTED\"}}\n\n";
+        assert_eq!(recorded(Dialect::Gemini, gemini).unwrap().1, 429);
+        // An Anthropic one by its type.
+        let anthropic = "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"rate_limit_error\",\"message\":\"slow down\"}}\n\n";
+        assert_eq!(recorded(Dialect::Anthropic, anthropic).unwrap().1, 429);
+    }
+
+    /// A request the upstream refuses is the caller's, as when the
+    /// upstream refuses it with an answer: it does not count against the
+    /// route.
+    #[test]
+    fn a_refused_request_in_a_stream_does_not_count_against_the_route() {
+        let (error_type, status, message, counted) = recorded(
+            Dialect::Responses,
+            &responses_failed(serde_json::json!({
+                "code": "context_length_exceeded",
+                "message": "Your input exceeds the context window of this model.",
+            })),
+        )
+        .unwrap();
+        assert_eq!((error_type.as_str(), status), ("ProviderHttpError", 400));
+        assert!(!counted);
+        assert!(message.contains("exceeds the context window"), "{message}");
+
+        // A Chat chunk: the code says nothing known, the type does.
+        let chat = "data: {\"error\":{\"message\":\"bad\",\"type\":\"invalid_request_error\",\"code\":\"weird_param\"}}\n\n";
+        let (_, status, _, counted) = recorded(Dialect::Chat, chat).unwrap();
+        assert_eq!((status, counted), (400, false));
+        // A Gemini one by its number.
+        let gemini = "data: {\"error\":{\"code\":404,\"message\":\"no such model\",\"status\":\"NOT_FOUND\"}}\n\n";
+        let (_, status, _, counted) = recorded(Dialect::Gemini, gemini).unwrap();
+        assert_eq!((status, counted), (404, false));
+    }
+
+    #[test]
+    fn an_overload_in_a_stream_counts_against_the_route() {
+        let frames = "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"Overloaded\"}}\n\n";
+        let (error_type, status, message, counted) = recorded(Dialect::Anthropic, frames).unwrap();
+        assert_eq!((error_type.as_str(), status), ("ProviderHttpError", 529));
+        assert!(counted);
+        assert!(
+            message.contains("up reported an error partway through the answer: Overloaded"),
+            "{message}"
+        );
+    }
+
+    /// An error that names no status is the upstream's failure. Without a
+    /// message of its own, the row says so in a fixed sentence — never
+    /// the chunk's text, which can be the model's answer.
+    #[test]
+    fn an_error_that_names_no_status_is_the_upstreams_failure() {
+        let chat = "data: {\"choices\":[{\"delta\":{\"content\":\"the secret plan\"}}],\"error\":{\"code\":\"server_error\"}}\n\n";
+        let (error_type, status, message, counted) = recorded(Dialect::Chat, chat).unwrap();
+        assert_eq!((error_type.as_str(), status), ("ProviderError", 502));
+        assert!(counted);
+        assert!(message.ends_with(NO_MESSAGE), "{message}");
+        assert!(!message.contains("secret plan"), "{message}");
+
+        // Nor the text of an Anthropic error event without a message.
+        let anthropic = "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"api_error\"},\"note\":\"the secret plan\"}\n\n";
+        let (_, status, message, counted) = recorded(Dialect::Anthropic, anthropic).unwrap();
+        assert_eq!((status, counted), (500, true));
+        assert!(message.ends_with(NO_MESSAGE), "{message}");
+        assert!(!message.contains("secret plan"), "{message}");
+
+        // A `response.failed` without an error object is still a failure.
+        let (_, status, message, _) = recorded(
+            Dialect::Responses,
+            &responses_failed(serde_json::Value::Null),
+        )
+        .unwrap();
+        assert_eq!(status, 502);
+        assert!(message.ends_with(NO_MESSAGE), "{message}");
+    }
+
+    /// An OpenAI-compatible relay can send `"error": null` in every chunk.
+    #[test]
+    fn a_null_error_is_no_error() {
+        let chat = "data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}],\"error\":null}\n\ndata: [DONE]\n\n";
+        assert_eq!(recorded(Dialect::Chat, chat), None);
+        let gemini = "data: {\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"hi\"}]}}],\"error\":null}\n\n";
+        assert_eq!(recorded(Dialect::Gemini, gemini), None);
+
+        // And an error after such chunks is still seen.
+        let then = format!(
+            "{chat}data: {{\"error\":{{\"message\":\"boom\",\"type\":\"server_error\"}}}}\n\n"
+        );
+        assert_eq!(recorded(Dialect::Chat, &then).unwrap().1, 502);
+    }
+
+    /// Cut to 500 characters, as a refusal's body is.
+    #[test]
+    fn what_the_upstream_said_is_cut() {
+        let said = Said {
+            message: "x".repeat(2_000),
+            status: None,
+        };
+        let StreamOutcome::UpstreamError { message, .. } = failed_partway("anthropic-main", &said)
+        else {
+            panic!("not a failure");
+        };
+        assert!(
+            message.contains("anthropic-main reported an error partway through the answer: xxx")
+        );
+        assert!(message.len() < 600, "{}", message.len());
+    }
+
+    /// A refused credential is logged without the upstream's words, as
+    /// an answer that refuses it is.
+    #[test]
+    fn a_refused_credential_in_a_stream_keeps_its_words_out_of_the_row() {
+        let said = Said {
+            message: "arn:aws:iam::123456789012:user/gateway may not".into(),
+            status: Some(403),
+        };
+        let StreamOutcome::UpstreamError {
+            error_type,
+            message,
+            status_code,
+        } = failed_partway("up", &said)
+        else {
+            panic!("not a failure");
+        };
+        assert_eq!(
+            (error_type.as_str(), status_code),
+            ("UpstreamAuthError", 401)
+        );
+        assert!(crate::proxy::upstream_failed(&error_type, status_code));
+        assert!(!message.contains("arn:aws"), "{message}");
+    }
+
+    /// What a Bedrock stream that ended in `e` is recorded as, with
+    /// whether it counts against the route, and what the client is told.
+    fn bedrock_recorded(e: tw_bedrock::eventstream::StreamError) -> (String, i64, bool, String) {
+        let (told, outcome) = bedrock_ended("bedrock-main", e);
+        let StreamOutcome::UpstreamError {
+            error_type,
+            status_code,
+            ..
+        } = outcome
+        else {
+            panic!("not a failure");
+        };
+        let counted = crate::proxy::upstream_failed(&error_type, status_code);
+        (error_type, status_code, counted, told)
+    }
+
+    fn exception(kind: &str, message: &str) -> tw_bedrock::eventstream::StreamError {
+        tw_bedrock::eventstream::StreamError::Upstream {
+            kind: kind.into(),
+            message: message.into(),
+        }
+    }
+
+    /// An exception Bedrock reports in its stream is recorded as the same
+    /// exception as an answer: throttling stays a 429 and counts like one,
+    /// a request Bedrock refuses does not count against the route.
+    #[test]
+    fn a_bedrock_exception_in_the_stream_is_classified_by_its_name() {
+        let (error_type, status, counted, told) =
+            bedrock_recorded(exception("throttlingException", "Too many requests"));
+        assert_eq!(
+            (error_type.as_str(), status, counted),
+            ("UpstreamRateLimited", 429, true)
+        );
+        // The client is told what it was told before.
+        assert_eq!(
+            told,
+            "Bedrock ended the stream: throttlingException: Too many requests"
+        );
+
+        let (error_type, status, counted, _) =
+            bedrock_recorded(exception("validationException", "Malformed input request"));
+        assert_eq!(
+            (error_type.as_str(), status, counted),
+            ("ProviderHttpError", 400, false)
+        );
+
+        for (kind, expected) in [
+            ("internalServerException", 500),
+            ("modelStreamErrorException", 500),
+            ("serviceUnavailableException", 503),
+            ("modelTimeoutException", 504),
+            ("ThrottlingException", 429),
+        ] {
+            let (_, status, counted, _) = bedrock_recorded(exception(kind, "x"));
+            assert_eq!((status, counted), (expected, true), "{kind}");
+        }
+
+        // Access denied keeps the IAM principal out of the row, as a 403
+        // answer does.
+        let (_, outcome) = bedrock_ended(
+            "bedrock-main",
+            exception(
+                "accessDeniedException",
+                "User: arn:aws:iam::123456789012:user/gateway is not authorized",
+            ),
+        );
+        let StreamOutcome::UpstreamError {
+            error_type,
+            message,
+            status_code,
+        } = outcome
+        else {
+            panic!("not a failure");
+        };
+        assert_eq!(
+            (error_type.as_str(), status_code),
+            ("UpstreamAuthError", 401)
+        );
+        assert!(!message.contains("arn:aws"), "{message}");
+    }
+
+    /// A damaged frame broke the stream off: no usable answer arrived.
+    #[test]
+    fn a_damaged_bedrock_frame_broke_the_stream_off() {
+        let (error_type, status, counted, told) = bedrock_recorded(
+            tw_bedrock::eventstream::StreamError::Malformed("bad CRC".into()),
+        );
+        assert_eq!(
+            (error_type.as_str(), status, counted),
+            ("transport", 502, true)
+        );
+        assert!(told.starts_with("Bedrock ended the stream: "), "{told}");
+    }
+
+    /// A Gemini answer ends with its stream, whichever form the caller
+    /// reads it in.
+    #[test]
+    fn a_gemini_answer_has_no_last_frame() {
+        let last = b"data: {\"candidates\":[{\"finishReason\":\"STOP\"}]}\n\n";
+        for sse in [true, false] {
+            assert!(!LastFrame::new(Dialect::Gemini, sse).is_in(last));
+        }
+    }
+}
diff --git a/crates/gateway/src/proxy/responses_ws.rs b/crates/gateway/src/proxy/responses_ws.rs
index 94babf53..2e2ee8cf 100644
--- a/crates/gateway/src/proxy/responses_ws.rs
+++ b/crates/gateway/src/proxy/responses_ws.rs
@@ -226,7 +226,13 @@ fn continue_from(
 /// Send the turn's SSE to the client, a frame per event, and hand back the
 /// `response` of its `response.completed`, if it got that far.
 ///
-/// Dropping the response body cancels the turn: the pipeline's tail then
+/// The turn is over at its last event (`response.completed`,
+/// `response.incomplete` or `response.failed`), not when the upstream's
+/// stream ends: an upstream can keep its stream open a while after that
+/// event, and the client's next turn does not wait for it. The pipeline's
+/// tail records such a turn as finished.
+///
+/// Dropping the response body before that cancels the turn: the tail then
 /// records it as cancelled by the client, as for an HTTP stream.
 async fn relay(
     resp: axum::response::Response,
@@ -240,7 +246,7 @@ async fn relay(
     loop {
         tokio::select! {
             chunk = body.next() => {
-                let (frames, end) = match chunk {
+                let (frames, mut end) = match chunk {
                     Some(Ok(bytes)) => (decoder.feed(&bytes), false),
                     _ => (decoder.flush(), true),
                 };
@@ -253,6 +259,7 @@ async fn relay(
                     if event.get("type").and_then(Value::as_str) == Some("response.completed") {
                         completed = event.get_mut("response").map(Value::take);
                     }
+                    end |= tw_dialect::convert::ends_answer(Dialect::Responses, &f);
                     if tx.send(Message::Text(f.data.into())).await.is_err() {
                         return (Turn::Gone, None);
                     }
diff --git a/crates/gateway/src/proxy/transport.rs b/crates/gateway/src/proxy/transport.rs
index c9b5b8a6..3e862b06 100644
--- a/crates/gateway/src/proxy/transport.rs
+++ b/crates/gateway/src/proxy/transport.rs
@@ -217,17 +217,33 @@ pub(crate) fn transport_error(e: reqwest::Error) -> GatewayError {
     }
 }
 
-/// Turn a non-2xx upstream answer into the error the caller sees.
+/// The error an upstream status other than 2xx becomes — for an answer
+/// (see [`check_status`]) and for an error the upstream reports partway
+/// through a stream with the same meaning.
 ///
-/// 429 keeps the upstream's `Retry-After` so a client's retry policy does
-/// not hammer the same quota window. 401/403 become an auth error: the
-/// gateway's own credential for this upstream was refused, which is
-/// about the route, not the caller.
+/// 429 keeps the upstream's `Retry-After`, if it gave one, so a client's
+/// retry policy does not hammer the same quota window. 401/403 become an
+/// auth error: the gateway's own credential for this upstream was
+/// refused, which is about the route, not the caller.
 ///
 /// Every other status is kept as it is, in `ProviderHttpError`. Whether
 /// it is the upstream failing (5xx, 408) or the upstream refusing this
 /// request (any other 4xx) decides failover and the circuit breaker —
 /// see `routing::is_upstream_failure`.
+pub(crate) fn status_error(
+    status: u16,
+    retry_after_secs: Option<u32>,
+    message: String,
+) -> GatewayError {
+    match status {
+        429 => GatewayError::UpstreamRateLimited { retry_after_secs },
+        401 | 403 => GatewayError::UpstreamAuthError { status, message },
+        _ => GatewayError::ProviderHttpError { status, message },
+    }
+}
+
+/// Turn a non-2xx upstream answer into the error the caller sees (see
+/// [`status_error`]).
 ///
 /// The upstream's body goes to the caller **truncated**: error bodies
 /// have carried stack traces, AWS account ids and full debug strings,
@@ -245,7 +261,7 @@ async fn check_status(
             .get(reqwest::header::RETRY_AFTER)
             .and_then(|v| v.to_str().ok())
             .and_then(crate::error::parse_retry_after_seconds);
-        return Err(GatewayError::UpstreamRateLimited { retry_after_secs });
+        return Err(status_error(429, retry_after_secs, String::new()));
     }
     if !status.is_success() {
         let body = resp.text().await.unwrap_or_default();
@@ -261,12 +277,11 @@ async fn check_status(
         } else {
             body
         };
-        let (status, message) = (status.as_u16(), format!("{label}: {shown}"));
-        return Err(if status == 401 || status == 403 {
-            GatewayError::UpstreamAuthError { status, message }
-        } else {
-            GatewayError::ProviderHttpError { status, message }
-        });
+        return Err(status_error(
+            status.as_u16(),
+            None,
+            format!("{label}: {shown}"),
+        ));
     }
     Ok(resp)
 }
diff --git a/crates/test-support/src/mock_provider.rs b/crates/test-support/src/mock_provider.rs
index 3a5a2d28..4f63a9e4 100644
--- a/crates/test-support/src/mock_provider.rs
+++ b/crates/test-support/src/mock_provider.rs
@@ -117,6 +117,81 @@ impl MockProvider {
     }
 }
 
+/// The events of a streamed Responses answer, "ok". It ends in
+/// `response.completed` with usage of 23 input and 5 output tokens, or,
+/// when `failed`, in `response.failed` with the error "The model crashed."
+pub fn responses_answer(model: &str, failed: bool) -> String {
+    let event = |kind: &str, mut v: Value| {
+        v["type"] = json!(kind);
+        format!("event: {kind}\ndata: {v}\n\n")
+    };
+    let last = if failed {
+        event(
+            "response.failed",
+            json!({"response": {
+                "id": "resp_1", "model": model, "status": "failed",
+                "error": {"code": "server_error", "message": "The model crashed."}
+            }}),
+        )
+    } else {
+        event(
+            "response.completed",
+            json!({"response": {
+                "id": "resp_1", "model": model, "status": "completed",
+                "output": [{"type": "message", "id": "msg_1", "role": "assistant",
+                            "content": [{"type": "output_text", "text": "ok"}]}],
+                "usage": {"input_tokens": 23, "output_tokens": 5, "total_tokens": 28}
+            }}),
+        )
+    };
+    [
+        event(
+            "response.created",
+            json!({"response": {"id": "resp_1", "model": model, "status": "in_progress"}}),
+        ),
+        event(
+            "response.output_text.delta",
+            json!({"item_id": "msg_1", "output_index": 0, "content_index": 0, "delta": "ok"}),
+        ),
+        last,
+    ]
+    .concat()
+}
+
+/// An upstream that answers every `POST {path}` with `frames` as an event
+/// stream, then keeps the stream open for `linger` before it ends it — as
+/// the ChatGPT Codex backend often does after its last event. Returns its
+/// base URL.
+///
+/// wiremock sends a body in one piece and ends it, so this is a server
+/// of its own.
+pub async fn sse_upstream_lingering(
+    path: &str,
+    frames: String,
+    linger: std::time::Duration,
+) -> String {
+    let answer = axum::routing::post(move || {
+        let frames = frames.clone();
+        async move {
+            let body = async_stream::stream! {
+                yield Ok::<_, std::convert::Infallible>(bytes::Bytes::from(frames));
+                tokio::time::sleep(linger).await;
+            };
+            axum::response::Response::builder()
+                .header("content-type", "text/event-stream")
+                .body(axum::body::Body::from_stream(body))
+                .unwrap()
+        }
+    });
+    let app = axum::Router::new().route(path, answer);
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        let _ = axum::serve(listener, app).await;
+    });
+    format!("http://{addr}")
+}
+
 fn openai_sse_chunks(model: &str) -> Vec<u8> {
     let chunk = |delta: Value| {
         format!(
diff --git a/crates/test-support/tests/analytics_clickhouse.rs b/crates/test-support/tests/analytics_clickhouse.rs
index 96f59afa..48530fb8 100644
--- a/crates/test-support/tests/analytics_clickhouse.rs
+++ b/crates/test-support/tests/analytics_clickhouse.rs
@@ -11,6 +11,7 @@
 use rust_decimal::Decimal;
 use serde_json::Value;
 use std::str::FromStr;
+use think_watch_test_support::mock_provider::{responses_answer, sse_upstream_lingering};
 use think_watch_test_support::prelude::*;
 
 async fn seed_runtime(app: &TestApp) -> (String, uuid::Uuid) {
@@ -487,6 +488,313 @@ async fn a_stream_the_caller_leaves_is_billed_on_an_estimate() {
     assert_eq!(detail["usage_estimated"], true, "{detail}");
 }
 
+/// Codex hangs up as soon as it has read `response.completed`, without
+/// waiting for the stream to end, and an upstream can end its stream a
+/// while after that frame. The caller left having read the whole answer:
+/// the request finished, and is billed on the usage the upstream
+/// reported, not logged as cancelled.
+#[ignore = "integration test — run via `make test-it`"]
+#[tokio::test]
+async fn a_caller_that_leaves_after_the_last_frame_finished_the_request() {
+    let app = TestApp::spawn_with_clickhouse().await;
+    let upstream = sse_upstream_lingering(
+        "/v1/responses",
+        responses_answer("gpt-linger", false),
+        std::time::Duration::from_secs(30),
+    )
+    .await;
+    let (key, user_id) = seed_upstream(&app, &upstream, "openai", "gpt-linger").await;
+    // Forwarded as sent: the caller reads the upstream's own frames.
+    sqlx::query(
+        "UPDATE model_routes SET upstream_protocol = 'openai_responses' WHERE model_id = $1",
+    )
+    .bind("gpt-linger")
+    .execute(&app.db)
+    .await
+    .unwrap();
+    app.rebuild_gateway_router().await;
+
+    let mut resp = reqwest::Client::new()
+        .post(format!("{}/v1/responses", app.gateway_url))
+        .bearer_auth(&key)
+        .json(&json!({"model": "gpt-linger", "stream": true, "input": "hi"}))
+        .send()
+        .await
+        .unwrap();
+    assert!(resp.status().is_success());
+    // Read the way Codex does: up to the end of `response.completed`, then
+    // hang up.
+    let mut got = String::new();
+    while !got
+        .find("event: response.completed\n")
+        .is_some_and(|at| got[at..].contains("\n\n"))
+    {
+        let chunk = tokio::time::timeout(std::time::Duration::from_secs(10), resp.chunk())
+            .await
+            .expect("response.completed within 10s")
+            .unwrap()
+            .expect("the stream is still open");
+        got.push_str(&String::from_utf8_lossy(&chunk));
+    }
+    drop(resp);
+
+    let ch = app.state.clickhouse.as_ref().expect("clickhouse client");
+    let (status, input, output, cost, detail) = last_gateway_log(ch, user_id).await;
+    assert_eq!(status, 200, "{detail}");
+    assert_eq!((input, output), (23, 5), "{detail}");
+    assert!(cost > Decimal::ZERO, "{detail}");
+    assert!(detail.get("stream_outcome").is_none(), "{detail}");
+    assert!(detail.get("usage_estimated").is_none(), "{detail}");
+}
+
+/// The `gateway_logs` rows of `user_id`, `(status, detail)`, once there
+/// are `n` of them.
+async fn gateway_rows(ch: &clickhouse::Client, user_id: uuid::Uuid, n: usize) -> Vec<(i64, Value)> {
+    for _ in 0..200 {
+        let rows: Vec<(i64, String)> = ch
+            .query(
+                "SELECT ifNull(status_code, -1), ifNull(detail, '') FROM gateway_logs \
+                  WHERE user_id = ? ORDER BY created_at",
+            )
+            .bind(user_id.to_string())
+            .fetch_all()
+            .await
+            .expect("CH select");
+        if rows.len() >= n {
+            return rows
+                .into_iter()
+                .map(|(status, detail)| {
+                    (status, serde_json::from_str(&detail).unwrap_or(Value::Null))
+                })
+                .collect();
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+    }
+    panic!("{n} gateway_logs rows never landed for user {user_id}");
+}
+
+/// A row for a stream that the upstream failed partway, after the
+/// response went out as 200: the upstream's failure as `(status,
+/// error_type)`, with its words.
+fn assert_failed_partway(status: i64, detail: &Value, logged: (i64, &str), said: &str) {
+    assert_eq!(status, logged.0, "{detail}");
+    assert_eq!(detail["stream_outcome"], "upstream_error", "{detail}");
+    assert_eq!(detail["error_type"], logged.1, "{detail}");
+    assert!(
+        detail["error_message"]
+            .as_str()
+            .is_some_and(|m| m.contains(said)),
+        "{detail}"
+    );
+    // What the caller saw: a 200 stream that ended in the error.
+    assert_eq!(detail["client_status"], 200, "{detail}");
+}
+
+/// An upstream can open a stream with 200 and report an error in it:
+/// Responses `response.failed`. The stream ends normally, but the request
+/// failed: it is logged as the upstream's failure, with what the upstream
+/// said, and counts against the route's circuit breaker like a 5xx answer.
+#[ignore = "integration test — run via `make test-it`"]
+#[tokio::test]
+async fn an_error_the_upstream_reports_in_its_stream_fails_the_request() {
+    use wiremock::matchers::{method, path};
+    use wiremock::{Mock, MockServer, ResponseTemplate};
+
+    let app = TestApp::spawn_with_clickhouse().await;
+    for (k, v) in [
+        ("gateway.cb_enabled", json!(true)),
+        ("gateway.cb_error_pct", json!(50)),
+        ("gateway.cb_min_samples", json!(2)),
+        ("gateway.cb_window_secs", json!(60)),
+        ("gateway.cb_open_secs", json!(600)),
+    ] {
+        app.set_setting(k, v).await;
+    }
+    let upstream = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/v1/responses"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .set_body_raw(responses_answer("gpt-fails", true), "text/event-stream"),
+        )
+        .mount(&upstream)
+        .await;
+    let (key, user_id) = seed_upstream(&app, &upstream.uri(), "openai", "gpt-fails").await;
+    sqlx::query(
+        "UPDATE model_routes SET upstream_protocol = 'openai_responses' WHERE model_id = $1",
+    )
+    .bind("gpt-fails")
+    .execute(&app.db)
+    .await
+    .unwrap();
+    app.rebuild_gateway_router().await;
+
+    let gw = app.gateway_client();
+    gw.set_bearer(&key);
+    for _ in 0..2 {
+        let resp = gw
+            .post(
+                "/v1/responses",
+                json!({"model": "gpt-fails", "stream": true, "input": "hi"}),
+            )
+            .await
+            .unwrap();
+        resp.assert_ok();
+        assert!(String::from_utf8_lossy(&resp.body).contains("event: response.failed"));
+    }
+
+    let ch = app.state.clickhouse.as_ref().expect("clickhouse client");
+    for (status, detail) in gateway_rows(ch, user_id, 2).await {
+        assert_failed_partway(
+            status,
+            &detail,
+            (502, "ProviderError"),
+            "The model crashed.",
+        );
+    }
+    // Two failures of two: the route's breaker is open, and the next
+    // request does not reach the upstream.
+    let resp = gw
+        .post(
+            "/v1/responses",
+            json!({"model": "gpt-fails", "input": "hi"}),
+        )
+        .await
+        .unwrap();
+    assert!(!resp.status.is_success(), "{:?}", resp.body);
+    assert_eq!(
+        upstream.received_requests().await.unwrap_or_default().len(),
+        2,
+        "the route stayed closed"
+    );
+}
+
+/// An Anthropic upstream reports an overload partway through a stream as
+/// an `error` event, and ends the stream. Forwarded as sent or converted
+/// for a Chat caller, the request failed.
+#[ignore = "integration test — run via `make test-it`"]
+#[tokio::test]
+async fn an_anthropic_error_event_partway_through_a_stream_fails_the_request() {
+    use wiremock::matchers::{method, path};
+    use wiremock::{Mock, MockServer, ResponseTemplate};
+
+    let app = TestApp::spawn_with_clickhouse().await;
+    let event = |kind: &str, v: Value| format!("event: {kind}\ndata: {v}\n\n");
+    let frames = [
+        event(
+            "message_start",
+            json!({"type": "message_start", "message": {
+                "id": "msg_1", "type": "message", "role": "assistant", "model": "claude-fails",
+                "content": [], "stop_reason": null,
+                "usage": {"input_tokens": 12, "output_tokens": 1}
+            }}),
+        ),
+        event(
+            "content_block_start",
+            json!({"type": "content_block_start", "index": 0,
+                   "content_block": {"type": "text", "text": ""}}),
+        ),
+        event(
+            "content_block_delta",
+            json!({"type": "content_block_delta", "index": 0,
+                   "delta": {"type": "text_delta", "text": "Hel"}}),
+        ),
+        event(
+            "error",
+            json!({"type": "error",
+                   "error": {"type": "overloaded_error", "message": "Overloaded"}}),
+        ),
+    ]
+    .concat();
+    let upstream = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/v1/messages"))
+        .respond_with(ResponseTemplate::new(200).set_body_raw(frames, "text/event-stream"))
+        .mount(&upstream)
+        .await;
+    let (key, user_id) = seed_upstream(&app, &upstream.uri(), "anthropic", "claude-fails").await;
+
+    let gw = app.gateway_client();
+    gw.set_bearer(&key);
+    let ask = [
+        (
+            "/v1/messages",
+            json!({"model": "claude-fails", "max_tokens": 100, "stream": true,
+                   "messages": [{"role": "user", "content": "hi"}]}),
+        ),
+        (
+            "/v1/chat/completions",
+            json!({"model": "claude-fails", "stream": true,
+                   "messages": [{"role": "user", "content": "hi"}]}),
+        ),
+    ];
+    for (route, body) in ask {
+        gw.post(route, body).await.unwrap().assert_ok();
+    }
+
+    let ch = app.state.clickhouse.as_ref().expect("clickhouse client");
+    for (status, detail) in gateway_rows(ch, user_id, 2).await {
+        // An overload is 529 as an answer too, and counts against the route.
+        assert_failed_partway(status, &detail, (529, "ProviderHttpError"), "Overloaded");
+    }
+}
+
+/// A caller that hangs up on `response.failed` — the answer's last event —
+/// while the upstream keeps its stream open did not get a success: the
+/// request failed, as if the stream had run to its end.
+#[ignore = "integration test — run via `make test-it`"]
+#[tokio::test]
+async fn a_caller_that_leaves_after_response_failed_saw_the_request_fail() {
+    let app = TestApp::spawn_with_clickhouse().await;
+    let upstream = sse_upstream_lingering(
+        "/v1/responses",
+        responses_answer("gpt-fails-linger", true),
+        std::time::Duration::from_secs(30),
+    )
+    .await;
+    let (key, user_id) = seed_upstream(&app, &upstream, "openai", "gpt-fails-linger").await;
+    sqlx::query(
+        "UPDATE model_routes SET upstream_protocol = 'openai_responses' WHERE model_id = $1",
+    )
+    .bind("gpt-fails-linger")
+    .execute(&app.db)
+    .await
+    .unwrap();
+    app.rebuild_gateway_router().await;
+
+    let mut resp = reqwest::Client::new()
+        .post(format!("{}/v1/responses", app.gateway_url))
+        .bearer_auth(&key)
+        .json(&json!({"model": "gpt-fails-linger", "stream": true, "input": "hi"}))
+        .send()
+        .await
+        .unwrap();
+    assert!(resp.status().is_success());
+    let mut got = String::new();
+    while !got
+        .find("event: response.failed\n")
+        .is_some_and(|at| got[at..].contains("\n\n"))
+    {
+        let chunk = tokio::time::timeout(std::time::Duration::from_secs(10), resp.chunk())
+            .await
+            .expect("response.failed within 10s")
+            .unwrap()
+            .expect("the stream is still open");
+        got.push_str(&String::from_utf8_lossy(&chunk));
+    }
+    drop(resp);
+
+    let ch = app.state.clickhouse.as_ref().expect("clickhouse client");
+    let rows = gateway_rows(ch, user_id, 1).await;
+    assert_eq!(rows.len(), 1, "{rows:?}");
+    assert_failed_partway(
+        rows[0].0,
+        &rows[0].1,
+        (502, "ProviderError"),
+        "The model crashed.",
+    );
+}
+
 /// Input read from the prompt cache is billed at a tenth of the input
 /// price and input written to it at 1.25×, not all at the full price —
 /// and the budget counter is debited by the same weights.
diff --git a/crates/test-support/tests/gateway_responses_ws.rs b/crates/test-support/tests/gateway_responses_ws.rs
index 2d83560c..f6786b02 100644
--- a/crates/test-support/tests/gateway_responses_ws.rs
+++ b/crates/test-support/tests/gateway_responses_ws.rs
@@ -10,6 +10,7 @@
 
 use futures::{SinkExt, StreamExt};
 use serde_json::Value;
+use think_watch_test_support::mock_provider::{responses_answer, sse_upstream_lingering};
 use think_watch_test_support::prelude::*;
 use tokio_tungstenite::tungstenite::Message;
 use tokio_tungstenite::tungstenite::client::IntoClientRequest;
@@ -254,3 +255,62 @@ async fn a_turn_continues_from_the_connections_last_response() {
     assert!(said[1].1.contains("hi there"), "{body}");
     assert_eq!(said[2], ("user", "two".to_string()), "{body}");
 }
+
+/// An upstream can keep its stream open a while after `response.completed`,
+/// as the ChatGPT Codex backend's often does. The turn is over at that
+/// event: the next one on the connection does not wait for the upstream's
+/// stream to end, and each is billed and logged as finished.
+#[ignore = "integration test — run via `make test-it`"]
+#[tokio::test]
+async fn a_turn_ends_at_its_last_event_while_the_upstream_stream_stays_open() {
+    let app = TestApp::spawn_with_clickhouse().await;
+    let upstream = sse_upstream_lingering(
+        "/v1/responses",
+        responses_answer("ws-linger", false),
+        std::time::Duration::from_secs(30),
+    )
+    .await;
+    let (user_id, key) = seed(&app, &upstream, "ws-linger").await;
+    sqlx::query(
+        "UPDATE model_routes SET upstream_protocol = 'openai_responses' WHERE model_id = $1",
+    )
+    .bind("ws-linger")
+    .execute(&app.db)
+    .await
+    .unwrap();
+    app.rebuild_gateway_router().await;
+
+    let mut socket = connect(&app, &key).await;
+    for _ in 0..2 {
+        socket.send(create("ws-linger", "hi")).await.unwrap();
+        // `turn` gives each event 10s: a turn that waited for the previous
+        // one's stream to end would get nothing for 30.
+        let events = turn(&mut socket).await;
+        assert_eq!(
+            events.last().unwrap()["type"],
+            "response.completed",
+            "{events:?}"
+        );
+        assert_eq!(text_of(&events), "ok", "{events:?}");
+    }
+    socket.close(None).await.unwrap();
+
+    let ch = app.state.clickhouse.as_ref().expect("CH wired up");
+    let mut rows = Vec::new();
+    for _ in 0..100 {
+        rows = ch
+            .query(
+                "SELECT ifNull(input_tokens, -1), ifNull(output_tokens, -1), ifNull(status_code, -1) \
+                   FROM gateway_logs WHERE user_id = ?",
+            )
+            .bind(user_id.to_string())
+            .fetch_all::<(i64, i64, i64)>()
+            .await
+            .expect("CH query");
+        if rows.len() >= 2 {
+            break;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    }
+    assert_eq!(rows, vec![(23, 5, 200), (23, 5, 200)], "gateway_logs rows");
+}
diff --git a/deploy/helm/think-watch/Chart.yaml b/deploy/helm/think-watch/Chart.yaml
index c24e00b4..c53611cb 100644
--- a/deploy/helm/think-watch/Chart.yaml
+++ b/deploy/helm/think-watch/Chart.yaml
@@ -2,8 +2,8 @@ apiVersion: v2
 name: think-watch
 description: Enterprise AI API Gateway & MCP Management Platform
 type: application
-version: 3.3.0
-appVersion: "3.3.0"
+version: 3.3.1
+appVersion: "3.3.1"
 keywords:
   - ai
   - gateway
diff --git a/deploy/helm/think-watch/README.md b/deploy/helm/think-watch/README.md
index ca9e6ec1..7e3993c1 100644
--- a/deploy/helm/think-watch/README.md
+++ b/deploy/helm/think-watch/README.md
@@ -39,7 +39,7 @@ make helm-deploy HELM_VALUES=deploy/helm/think-watch/values-production.yaml.exam
 # Or one-off flags
 helm upgrade --install thinkwatch deploy/helm/think-watch \
   --namespace thinkwatch --create-namespace \
-  --set image.server.tag=3.3.0 \
+  --set image.server.tag=3.3.1 \
   --set ingress.enabled=true \
   --set ingress.gateway.host=api.example.com
 ```
diff --git a/deploy/helm/think-watch/values-production.yaml.example b/deploy/helm/think-watch/values-production.yaml.example
index 308c319c..2b1184ed 100644
--- a/deploy/helm/think-watch/values-production.yaml.example
+++ b/deploy/helm/think-watch/values-production.yaml.example
@@ -10,7 +10,7 @@ replicaCount: 3
 
 image:
   server:
-    tag: ""   # "" → .Chart.AppVersion; set e.g. "3.3.0" to pin (no "v")
+    tag: ""   # "" → .Chart.AppVersion; set e.g. "3.3.1" to pin (no "v")
   web:
     tag: ""
 
diff --git a/web/package.json b/web/package.json
index 22f90a06..70e57c97 100644
--- a/web/package.json
+++ b/web/package.json
@@ -1,7 +1,7 @@
 {
   "name": "web",
   "private": true,
-  "version": "3.3.0",
+  "version": "3.3.1",
   "type": "module",
   "packageManager": "pnpm@12.9.1",
   "scripts": {