From 6a650fb0476e9e2adb2a0f505932d2d81491570e Mon Sep 17 00:00:00 2001 From: fylorn <249551762+fylorn@users.noreply.github.com> Date: Wed, 14 Oct 2026 10:23:45 +0800 Subject: [PATCH] Request content: keep the live answer until the stored one has landed When a streamed request ended, the Content tab could switch from the live answer to the stored detail before the answer was on disk and show the response as not saved; reopening the request showed it in full. Core records the request row and the bodies on separate threads. The row (and `request_priced`) lands first; the request body was stored when the request started, while the heads and the answer are written after the request ends and has been redacted. `landed()` accepted the detail as soon as any one of them was there, so the request body alone was enough. - `caughtUp(d, live)` replaces `landed(d)`: the stored detail counts only once it has every head and body the live content showed (heads, client request, the serving hop's request, the client answer, and the upstream answer when it differs from what the client got) - The live subscription now runs until it closes on its own instead of stopping when the request leaves in-flight: `request_priced` can arrive before the last chunk and `end`, and stopping then lost the tail of the answer and never fetched the stored copy Co-Authored-By: Claude Opus 5.5 --- src/traffic/drawer/Content.tsx | 8 ++-- src/traffic/drawer/RequestDrawer.tsx | 39 ++++++++------- src/traffic/drawer/dialect.test.ts | 72 +++++++++++++++++++++++++++- src/traffic/drawer/live.ts | 16 +++++-- src/traffic/drawer/wireModel.ts | 40 ++++++++++++++-- 5 files changed, 144 insertions(+), 31 deletions(-) diff --git a/src/traffic/drawer/Content.tsx b/src/traffic/drawer/Content.tsx index 4f0026ba..c9048e90 100644 --- a/src/traffic/drawer/Content.tsx +++ b/src/traffic/drawer/Content.tsx @@ -14,7 +14,7 @@ import type { LiveContent } from "./live"; import { Parsed } from "./Parsed"; import { Raw } from "./Raw"; import { requestDrawerText } from "./RequestDrawer.i18n"; -import { fromDetail, fromLive, landed } from "./wireModel"; +import { caughtUp, fromDetail, fromLive } from "./wireModel"; export type ContentView = "parsed" | "raw"; @@ -39,8 +39,8 @@ export function useContentView(): [ContentView | null, (v: ContentView) => void] * 请求详情的「内容」：**解析**（按对话读出来）和**原始报文**（报文原样，客户端一侧、上游一侧）。 * * 在跑的请求看的是实时内容（`live`），一边收一边长；请求结束、存下的那一份取回来之后换成存下的 - * —— 两种整理成同一个样子（`wireModel`），换的那一下画面不跳。存下的那一份还没落盘（请求刚 - * 结束的那几毫秒）时接着画实时收到的。 + * —— 两种整理成同一个样子（`wireModel`），换的那一下画面不跳。存下的那一份还没赶上实时收到的 + * （回答比请求那一行晚落盘，`caughtUp`）时接着画实时收到的。 * * 安全命中的那几段（脱敏的凭据、它的占位符、可疑的工具调用、内容规则命中的字）在两种看法里都 * 标出来，颜色和安全日志里的一样（`needlesOf`）。 @@ -61,7 +61,7 @@ export function Content({ const running = d.in_flight; const dialect = dialectOf(d.row.path, d.row.translated?.from); const view: ContentView = chosen ?? (dialect ? "parsed" : "raw"); - const fromStream = live !== null && live.order.length > 0 && (running || !landed(d)); + const fromStream = live !== null && live.order.length > 0 && (running || !caughtUp(d, live)); const version = live?.version ?? 0; const model = useMemo( () => (fromStream && live ? fromLive(live, d) : fromDetail(d)), diff --git a/src/traffic/drawer/RequestDrawer.tsx b/src/traffic/drawer/RequestDrawer.tsx index bbee36f4..830c817b 100644 --- a/src/traffic/drawer/RequestDrawer.tsx +++ b/src/traffic/drawer/RequestDrawer.tsx @@ -21,13 +21,13 @@ import { PanelHeader, PanelHeaderSkeleton, PanelSkeleton } from "../PanelHeader" import { stateOf, type DrawerState } from "./parts"; import { Content, useContentView } from "./Content"; import { dialectOf } from "./dialect"; -import { useLiveContent } from "./live"; +import { type LiveContent, useLiveContent } from "./live"; import { Replay } from "./Replay"; import { requestDrawerText } from "./RequestDrawer.i18n"; import { Routing } from "./Routing"; import { Timeline } from "./Timeline"; import { Usage } from "./Usage"; -import { landed } from "./wireModel"; +import { caughtUp } from "./wireModel"; type Tab = "timeline" | "routing" | "payload" | "usage" | "replay"; @@ -136,24 +136,27 @@ function Detail({ id, onClose }: { id: number; onClose: () => void }) { }; }, []); /* - **实时内容结束了，取存下的那一份。**结局和正文是分开落盘的，正文可能晚几毫秒才到：取回来的 - 还在跑、或者还没有报文，隔一会儿再取，几次之后照取到的画 + **实时内容结束了，取存下的那一份。**结局和正文是分开落盘的，回答可能晚一阵才到：取回来的 + 还在跑、或者还没赶上实时收到的（`caughtUp`），隔一会儿再取，几次之后照取到的画 */ - const settle = useCallback(async () => { - let last: RequestDetail | null = null; - for (const ms of [0, 150, 400, 1_000, 2_500]) { - if (ms) await new Promise((r) => setTimeout(r, ms)); - if (!mounted.current) return; - try { - last = await call("RequestDetail", null, id); - if (landed(last)) break; - } catch { - // 下一轮再取 + const settle = useCallback( + async (live: LiveContent) => { + let last: RequestDetail | null = null; + for (const ms of [0, 150, 400, 1_000, 2_500]) { + if (ms) await new Promise((r) => setTimeout(r, ms)); + if (!mounted.current) return; + try { + last = await call("RequestDetail", null, id); + if (caughtUp(last, live)) break; + } catch { + // 下一轮再取 + } } - } - if (last && mounted.current) setD(last); - }, [id]); - const live = useLiveContent(id, running && sawContent, () => void settle()); + if (last && mounted.current) setD(last); + }, + [id], + ); + const live = useLiveContent(id, running && sawContent, (l) => void settle(l)); useEffect(() => { if (!running) return; diff --git a/src/traffic/drawer/dialect.test.ts b/src/traffic/drawer/dialect.test.ts index 5520d331..0326f874 100644 --- a/src/traffic/drawer/dialect.test.ts +++ b/src/traffic/drawer/dialect.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it } from "vitest"; import type { LiveBatch } from "@/types"; import { AnswerReader, dialectOf, readRequest, type ClientDialect } from "./dialect"; import { LiveContent, utf8Bytes } from "./live"; -import { bodyText, fromDetail, fromLive } from "./wireModel"; +import { bodyText, caughtUp, fromDetail, fromLive } from "./wireModel"; const sse = (events: [string | null, unknown][]) => events.map(([e, d]) => `${e ? `event: ${e}\n` : ""}data: ${typeof d === "string" ? d : JSON.stringify(d)}\n\n`).join(""); @@ -288,4 +288,74 @@ describe("实时内容", () => { expect(utf8Bytes("中文")).toBe(6); expect(utf8Bytes("😀")).toBe(4); }); + + describe("存下的那一份赶上实时收到的了没有", () => { + const view = (text: string) => ({ text, original_len: text.length, truncated: false }); + const stored = (over: Partial> = {}) => + ({ + row: { routing: { attempts: [{ provider: "anthropic" }] } }, + heads: [], + request_body: null, + request_after_plugins: null, + response_body: null, + upstream_request_body: null, + upstream_response_body: null, + plugins: [], + in_flight: false, + ...over, + }) as unknown as Parameters[0]; + const someHead = [{ side: "client", dir: "request", attempt: 0, line: "POST /v1/messages", headers: [] }]; + /** 一个流完了的请求：客户端收到的回答就是上游的原话 */ + const finished = (clientAnswer = "data: {}\n\n") => { + const live = new LiveContent(); + live.apply( + batch([ + head("client", "request", 0, "POST /v1/messages HTTP/1.1"), + body("client", "request", 0, '{"messages":[]}'), + head("upstream", "request", 1, "POST https://api.anthropic.com/v1/messages"), + head("upstream", "response", 1, "HTTP/2 200"), + head("client", "response", 0, "HTTP/1.1 200"), + body("upstream", "response", 1, "data: {}\n\n"), + body("client", "response", 0, clientAnswer), + ]), + ); + return live; + }; + + it("请求那一行和请求体先到、回答还没落盘：没赶上，接着画实时收到的", () => { + const live = finished(); + // 计价事件到的那一刻：请求体一开始就存了，报文头和回答还在打码、排队 + expect(caughtUp(stored({ request_body: view("{}") }), live)).toBe(false); + expect(caughtUp(stored({ request_body: view("{}"), heads: someHead }), live)).toBe(false); + expect( + caughtUp(stored({ request_body: view("{}"), heads: someHead, response_body: view("data: {}\n\n") }), live), + ).toBe(true); + }); + + it("还在跑的不算赶上", () => { + const all = { request_body: view("{}"), heads: someHead, response_body: view("data: {}\n\n") }; + expect(caughtUp(stored({ ...all, in_flight: true }), finished())).toBe(false); + }); + + it("交给客户端的和上游的原话不一样：上游那一份也到了才算，否则 response_body 可能还是上游的原话", () => { + const live = finished('data: {"choices":[]}\n\n'); + const base = { request_body: view("{}"), heads: someHead, response_body: view("data: {}\n\n") }; + expect(caughtUp(stored(base), live)).toBe(false); + expect(caughtUp(stored({ ...base, upstream_response_body: view("data: {}\n\n") }), live)).toBe(true); + }); + + it("回答那一跳另发了请求体：存下的也要有", () => { + const live = finished(); + live.apply(batch([body("upstream", "request", 1, '{"messages":[],"x":1}')])); + const base = { request_body: view("{}"), heads: someHead, response_body: view("data: {}\n\n") }; + expect(caughtUp(stored(base), live)).toBe(false); + expect(caughtUp(stored({ ...base, upstream_request_body: view("{}") }), live)).toBe(true); + }); + + it("没有实时内容的：有一样落盘了就算", () => { + expect(caughtUp(stored(), null)).toBe(false); + expect(caughtUp(stored({ request_body: view("{}") }), null)).toBe(true); + expect(caughtUp(stored({ request_body: view("{}") }), new LiveContent())).toBe(true); + }); + }); }); diff --git a/src/traffic/drawer/live.ts b/src/traffic/drawer/live.ts index 151fd0a7..8beacaa5 100644 --- a/src/traffic/drawer/live.ts +++ b/src/traffic/drawer/live.ts @@ -131,19 +131,25 @@ let seq = 0; const subId = (id: number) => `${id}.${Date.now().toString(36)}.${(seq += 1)}`; /** - * 请求 `id` 在跑时订阅它的实时内容（`on`）。返回收到的全部，还没订过是 null。 + * 请求 `id` 在跑时订阅它的实时内容（`start`）。返回收到的全部，还没订过是 null。 * * - **每一帧最多重画一次**：内容一批一批来（Rust 那边攒了 30 毫秒），重画跟着显示器的节拍走 * - **订阅结束时**（收到 `end`、断了、没订上）调一次 `onClosed`：该去取存下的详情了 - * - **请求结束、浮层关了、换了一条请求时退订**；`on` 变成 false 之后收到的那一份留着，画到 - * 存下的那一份取回来为止 + * - **订上了就订到它自己结束为止**，`start` 变回 false 也不退订：请求结束的消息（计价事件） + * 可能比最后那一截内容和 `end` 先到，那时就退订的话，回答少了结尾，也不会再去取存下的那一份 + * - 浮层关了、换了一条请求时退订；收到的那一份留着，画到存下的那一份赶上它为止 * - 先听、再订阅：补发的那一批可能比订阅的应答先到 */ -export function useLiveContent(id: number, on: boolean, onClosed: () => void): LiveContent | null { +export function useLiveContent( + id: number, + start: boolean, + onClosed: (live: LiveContent) => void, +): LiveContent | null { const [live, setLive] = useState(null); const [, setVersion] = useState(0); const closedCb = useRef(onClosed); closedCb.current = onClosed; + const on = start || live !== null; useEffect(() => { if (!on) return; @@ -160,7 +166,7 @@ export function useLiveContent(id: number, on: boolean, onClosed: () => void): L if (frame === 0) frame = requestAnimationFrame(paint); }; const closed = () => { - if (!gone) closedCb.current(); + if (!gone) closedCb.current(store); }; const un = subscribe("request-live", (e) => { if (e.payload.sub !== sub || store.closed) return; diff --git a/src/traffic/drawer/wireModel.ts b/src/traffic/drawer/wireModel.ts index 0d465006..5ac1606f 100644 --- a/src/traffic/drawer/wireModel.ts +++ b/src/traffic/drawer/wireModel.ts @@ -200,7 +200,41 @@ export function fromLive(live: LiveContent, d: RequestDetail): WireModel { return { client, upstream, afterPlugins: null }; } -/** 存下的详情里有没有这些报文（落盘了）：请求一结束取回来的那一份，正文可能晚几毫秒才到 */ -export function landed(d: RequestDetail): boolean { - return !d.in_flight && (d.heads.length > 0 || d.request_body !== null || d.response_body !== null); +/** 实时收到过这一段的正文（空的不算：空的不落盘） */ +const had = (s: Segment | undefined): s is Segment => s !== undefined && s.hasBody && s.bytes > 0; + +/** 两份实时的正文不是同一份 */ +function differs(a: Segment, b: Segment): boolean { + return a.bytes !== b.bytes || a.chunks.length !== b.chunks.length || a.chunks.some((c, i) => c !== b.chunks[i]); +} + +/** + * 存下的那一份赶上实时收到的了没有：请求结束了，实时收到过的报文头、正文，存下的里面都有了。 + * + * **请求那一行和正文分两条线程落盘**（core 的 `tw_store::task`）：计价事件到的时候那一行已经在 + * 库里，请求体在请求一开始就存了，报文头和回答却要等请求结束、打完码才写。只看「有一样落盘了」 + * 的话，取回来的那一份里回答还没到，「内容」就从实时的那一份换成了「未保存」。 + * + * 交给客户端的回答和上游的原话不一样（转换了格式、回答钩子、回答过滤）时两份都存，客户端那一份 + * 还没落盘时 `response_body` 是上游的原话：上游那一份也要到了才算。没有实时内容的（没订上、 + * 订阅前就结束了），有一样落盘了就算 + */ +export function caughtUp(d: RequestDetail, live: LiveContent | null): boolean { + if (d.in_flight) return false; + if (live === null || live.order.length === 0) { + return d.heads.length > 0 || d.request_body !== null || d.response_body !== null; + } + const anyHead = live.order.some((k) => live.segments.get(k)?.head); + if (anyHead && d.heads.length === 0) return false; + if (had(live.get("client", "request", 0)) && d.request_body === null) return false; + const attempts = live.attempts(); + const serving = attempts[attempts.length - 1]; + if (serving !== undefined && had(live.get("upstream", "request", serving)) && d.upstream_request_body === null) { + return false; + } + const cres = live.get("client", "response", 0); + if (!had(cres)) return true; + if (d.response_body === null) return false; + const up = serving === undefined ? undefined : live.get("upstream", "response", serving); + return !(had(up) && differs(cres, up) && d.upstream_response_body === null); }