diff --git a/src/traffic/drawer/Content.tsx b/src/traffic/drawer/Content.tsx index 4f0026b..c9048e9 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 bbee36f..830c817 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 5520d33..0326f87 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 151fd0a..8beacaa 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 0d46500..5ac1606 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); }