v0.68.0, and a null error in a stream chunk is not an error

Cargo.toml

@@ -54,7 +54,7 @@ categories = ["network-programming", "web-programming::http-server"]
 
 # 二进制走 CalVer，crate 走 SemVer —— 两者是两回事，卖给不同的人。
 # 这里是 crate 的版本。
-version = "0.67.1"
+version = "0.68.0"
 
 [workspace.dependencies]
 # ── 内部 crate ───────────────────────────────────────────

crates/tw-dialect/src/chat/stream.rs

@@ -42,9 +42,14 @@ impl Parser {
         let Ok(v) = serde_json::from_str::<Value>(data) else {
             return;
         };
-        if let Some(e) = v.get("error") {
+        // 有的中转每一块都带着 `"error": null`：那不是错误
+        if let Some(e) = v.get("error").filter(|e| !e.is_null()) {
+            // 没写说明的不拿整块凑：那一块里可能是回答的正文
             out.push(Event::Error {
-                message: str_of(e, "message").unwrap_or(data).to_string(),
+                message: str_of(e, "message")
+                    .or_else(|| e.as_str())
+                    .unwrap_or("the upstream returned an error")
+                    .to_string(),
             });
             return;
         }
@@ -303,6 +308,34 @@ mod tests {
         out
     }
 
+    /// 有的中转每一块都带着 `"error": null`：照常读内容。真的错误没写说明时，不拿那一块凑
+    #[test]
+    fn a_null_error_is_read_as_content_and_a_bare_error_has_no_chunk_for_a_message() {
+        let ev = parse(&[
+            "data: {\"choices\":[{\"delta\":{\"content\":\"secret answer\"}}],\"error\":null}\n\n",
+            "data: [DONE]\n\n",
+        ]);
+        assert!(
+            !ev.iter().any(|e| matches!(e, Event::Error { .. })),
+            "{ev:#?}"
+        );
+        assert!(
+            ev.iter().any(
+                |e| matches!(e, Event::Delta { delta: Delta::Text(t), .. } if t == "secret answer")
+            ),
+            "{ev:#?}"
+        );
+        let ev = parse(&[
+            "data: {\"error\":{\"code\":\"server_error\"},\"note\":\"secret answer\"}\n\n",
+        ]);
+        assert_eq!(
+            ev,
+            vec![Event::Error {
+                message: "the upstream returned an error".into()
+            }]
+        );
+    }
+
     #[test]
     fn reasoning_text_and_two_tool_calls_become_four_blocks() {
         let ev = parse(&[

crates/tw-dialect/src/convert.rs

@@ -1067,6 +1067,17 @@ mod stream_error_tests {
             r#"{"id":"c1","choices":[{"index":0,"delta":{"content":"hi"}}]}"#,
         );
         assert_eq!(stream_error(Dialect::Chat, &delta), None);
+        // 有的中转每一块都带着 `"error": null`
+        let relayed = frame(
+            None,
+            r#"{"id":"c1","choices":[{"index":0,"delta":{"content":"hi"}}],"error":null}"#,
+        );
+        assert_eq!(stream_error(Dialect::Chat, &relayed), None);
+        let gemini = frame(
+            None,
+            r#"{"candidates":[{"content":{"parts":[{"text":"hi"}]}}],"error":null}"#,
+        );
+        assert_eq!(stream_error(Dialect::Gemini, &gemini), None);
     }
 
     #[test]

crates/tw-dialect/src/gemini/request.rs

@@ -128,7 +128,8 @@ pub fn decode_request(
                 parts.push(Part::ToolResult(ToolResult {
                     id,
                     content: vec![Part::Text(response_text(body))],
-                    is_error: body.get("error").is_some() && body.get("output").is_none(),
+                    is_error: body.get("error").is_some_and(|e| !e.is_null())
+                        && body.get("output").is_none(),
                 }));
             } else if let Some(key) = ["fileData", "executableCode", "codeExecutionResult"]
                 .into_iter()

crates/tw-dialect/src/gemini/stream.rs

@@ -41,7 +41,8 @@ impl Parser {
     }
 
     pub fn chunk(&mut self, v: &Value, out: &mut Vec<Event>) {
-        if let Some(e) = v.get("error") {
+        // `"error": null` 不是错误
+        if let Some(e) = v.get("error").filter(|e| !e.is_null()) {
             out.push(Event::Error {
                 message: str_of(e, "message")
                     .unwrap_or("the upstream returned an error")

crates/tw-gateway/src/plugin/reply/chat.rs

@@ -152,7 +152,7 @@ impl Codec {
             out.push(Out::Keep(f));
             return Ok(());
         };
-        if data.get("error").is_some() {
+        if data.get("error").is_some_and(|e| !e.is_null()) {
             self.close_lane(chain, out).await?;
             self.settle(chain, out).await?;
             out.push(Out::Keep(f));

crates/tw-gateway/src/plugin/reply/gemini.rs

@@ -80,7 +80,7 @@ impl Codec {
                 self.envelope.insert(k.into(), v.clone());
             }
         }
-        if data.get("error").is_some() {
+        if data.get("error").is_some_and(|e| !e.is_null()) {
             let end = self.end(chain).await?;
             if !end.is_empty() {
                 out.push(self.chunk(end));

crates/tw-gateway/src/plugin/view/gemini.rs

@@ -272,7 +272,7 @@ fn part(
             key,
             &id,
             &response_text(body),
-            body.get("error").is_some() && body.get("output").is_none(),
+            body.get("error").is_some_and(|e| !e.is_null()) && body.get("output").is_none(),
         );
     }
     let label = p

crates/tw-gateway/src/pulse.rs

@@ -53,7 +53,7 @@ pub(crate) fn advances(dialect: Dialect, ty: &str, v: &serde_json::Value) -> boo
 
 /// Chat Completions 的一块：有错误、有内容、有结束原因，或者是收尾的那块用量
 fn chat_advances(v: &serde_json::Value) -> bool {
-    if v.get("error").is_some() {
+    if v.get("error").is_some_and(|e| !e.is_null()) {
         return true;
     }
     let Some(choices) = v.get("choices").and_then(|c| c.as_array()) else {
@@ -84,7 +84,7 @@ fn chat_advances(v: &serde_json::Value) -> bool {
 
 /// Gemini 的一块：有错误、被拦下，或者某个候选有内容、有结束原因
 fn gemini_advances(v: &serde_json::Value) -> bool {
-    if v.get("error").is_some() || v.get("promptFeedback").is_some() {
+    if v.get("error").is_some_and(|e| !e.is_null()) || v.get("promptFeedback").is_some() {
         return true;
     }
     let Some(candidates) = v.get("candidates").and_then(|c| c.as_array()) else {

crates/tw-gateway/src/server/pipeline/opening.rs

@@ -264,7 +264,8 @@ fn error_in(dialect: Dialect, ty: &str, v: &serde_json::Value, data: &str) -> Op
             }
             _ => None,
         },
-        Dialect::Chat => v.get("error").map(|e| {
+        // 有的中转每一块都带着 `"error": null`：那不是错误
+        Dialect::Chat => v.get("error").filter(|e| !e.is_null()).map(|e| {
             let kind = Some(text_of(e, "code"))
                 .filter(|c| !c.is_empty())
                 .unwrap_or_else(|| text_of(e, "type"));
@@ -276,7 +277,7 @@ fn error_in(dialect: Dialect, ty: &str, v: &serde_json::Value, data: &str) -> Op
                 .unwrap_or_else(|| openai_status(&kind));
             error(status, data, kind, text_of(e, "message"))
         }),
-        Dialect::Gemini => v.get("error").map(|e| {
+        Dialect::Gemini => v.get("error").filter(|e| !e.is_null()).map(|e| {
             let status = e
                 .get("code")
                 .and_then(|c| c.as_u64())
@@ -480,6 +481,15 @@ mod tests {
         ));
     }
 
+    /// 有的中转每一块都带着 `"error": null`：那是内容，不是上游报错
+    #[test]
+    fn a_null_error_is_no_error() {
+        let chat = "data: {\"choices\":[{\"delta\":{\"content\":\"Hi\"}}],\"error\":null}\n\n";
+        assert_eq!(first_verdict(Dialect::Chat, chat), Some(Judge::Content));
+        let gemini = "data: {\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"Hi\"}]}}],\"error\":null}\n\n";
+        assert_eq!(first_verdict(Dialect::Gemini, gemini), Some(Judge::Content));
+    }
+
     #[test]
     fn a_gemini_error_carries_its_http_code() {
         let err = "data: {\"error\":{\"code\":503,\"status\":\"UNAVAILABLE\",\"message\":\"overloaded\"}}\n\n";

release-notes/0.68.0.md

@@ -0,0 +1,33 @@
+This release counts the network traffic between the gateway and its upstreams, lets a running request's content be watched live, records where each security hit was and what was done about it, and fixes Codex on a ChatGPT account (its finished turns were recorded as cancelled and its longer sessions had no usage) and upstreams that write `"error": null` in every streamed chunk.
+
+**Upgrade notes**
+
+- The control-plane protocol version (`CONTROL_API_VERSION`) goes from 43 to 44. ThinkWatch Lite connects only to a core with the same protocol version: ThinkWatch Lite 2026.10.11 includes 0.67.1 (protocol 43) and does not connect to 0.68.0. A server used with it stays on 0.67.1 until the app is updated to a release that includes 0.68.0; `sudo twcore upgrade --version 0.67.1 --restart` switches a server back.
+- **The request store's schema goes to 27: the request history, stored payloads and the security log are cleared on the first start.** Recorded requests have no traffic figures, and recorded security events have no detail.
+- No configuration key changes and no new message codes.
+- Changed in the protocol:
+  - `HistoryRow` loses `bytes` and gains `sent_bytes`, `received_bytes` and `egress`. `RequestFinished` carries `sent_bytes` and `received_bytes` instead of `bytes`; `RequestFailed` and `RequestCancelled` carry them as optional (absent when no attempt reached an upstream). `RequestRouted` and in-flight requests carry `egress`; each attempt in the routing view carries `proxy`. `ReplayOriginal.bytes` becomes `received_bytes`.
+  - `Summary`, `CostBucket` and `CostBucketGroup` gain `sent_bytes` and `received_bytes`; `CostBucket` also gains its token counts and the first-token `ttft_p50_ms`, `ttft_p95_ms` and `ttft_samples` of its bucket. `CostDim` gains `egress` (the direct group is named `""`).
+  - New endpoints `GET /latency/client` (first-token latency by key) and `GET /request/{id}/live` (live content, an event stream; `404 control.request_not_running` once the request has ended).
+  - `RequestDetail` gains `heads`, `upstream_request_body` and `upstream_response_body`. **`response_body` is now what the client received**; for a converted request the upstream's own answer is in `upstream_response_body`.
+  - `SecurityEventView` gains `session`, `sent_model`, `direction`, `locations`, `more_locations`, `rule_snapshot` and `outcome_detail`. The live `SecretsFound` items, `ContentMatched` and `ToolCallFlagged` carry the same detail as `detail`.
+
+**Traffic.** Each request records how much it sent to upstreams and received from them: request and response bodies as they crossed the network, after the gateway's own compression and before any decoding, summed over every attempt of a failover. It also records the proxy it went out through (`egress`), or none for a direct connection. The totals are in the summary, per time bucket, and grouped by model, upstream, key or proxy.
+
+**Live content.** `GET /request/{id}/live` streams a running request's request and response — request and status lines, headers and bodies, the client's side and each upstream attempt's side — first everything so far, then each piece as it passes. Secrets are replaced and credentials masked exactly as in what is written to disk. Finished requests keep the headers, and the upstream's side of the bodies when it differs from the client's (a converted format, or a request with parts removed for the upstream).
+
+**First-token latency over time.** Every time bucket carries the median and 95th percentile of its first-token times, and first-token latency can be grouped by key.
+
+**Security log detail.** Every security event now says where each hit was and what exactly was done:
+- Every place the rule matched, in order, up to 50 per event (with a count of the rest): which part of the request or answer (system prompt, a message with its index and role, a tool call or tool result with the tool's name, the answer text), its JSON path in the client's own format, and up to 80 characters of context on either side.
+- The rule as it was when it matched: a built-in rule's id and name with the core version that evaluated it, or a custom rule's pattern and how it matches, so the entry stays readable after the rule is changed or deleted.
+- What was done: the placeholders a replaced credential was sent as; for a cut tool call, the tool, its arguments and the sentence the client received; for a refused request, the refusal; for removed characters, how many segments were removed.
+- The session the request belonged to and the model that was sent.
+
+Contexts and matches are redacted exactly like stored payloads, so no secret is written in clear. The detail is kept in the request log and expires with it. Changes to the guard settings are not written to the security log.
+
+**Codex on a ChatGPT account.**
+- Codex stops reading as soon as it has `response.completed` and closes the connection, without waiting for the stream to end; the ChatGPT backend often ends its stream a little later. Core reported the ending only when the upstream stream ended, so a client closing first was recorded as a cancellation, and most Codex turns appeared cancelled. Once the answer's last frame has reached the client, a request whose client then leaves is recorded as finished, with its usage and cost.
+- The ChatGPT backend's `usage` carries a per-item attribution that grows with the conversation. Usage objects were only read up to 8 KB, so longer sessions recorded no tokens and no cost. They are now read up to 1 MiB.
+
+**`"error": null` in a stream.** Some OpenAI-compatible relays and Gemini-format upstreams write `"error": null` in every streamed chunk. Core took such a chunk for an error reported by the upstream: the request failed at its first chunk and moved to the next upstream, a converted stream ended with an error and lost its text, reply plugins left those chunks untouched, and a Gemini tool result carrying `"error": null` was marked as failed. A `null` error is now ignored everywhere. An error that carries no message is recorded with a fixed sentence instead of the chunk it came in, which could hold answer text.