diff --git a/crates/agent-gateway/test/websocket/v2_kbrain_test.go b/crates/agent-gateway/test/websocket/v2_kbrain_test.go index 981cf1dbc..bc697e7e2 100644 --- a/crates/agent-gateway/test/websocket/v2_kbrain_test.go +++ b/crates/agent-gateway/test/websocket/v2_kbrain_test.go @@ -25,6 +25,8 @@ import ( // client request suppression, disconnect, after_seq replay, and one terminal. func TestV2KBrainRelayBrowserReconnect(t *testing.T) { firstEventWritten := make(chan struct{}) + releaseFirstEvents := make(chan struct{}) + var firstReleaseOnce sync.Once releaseEvents := make(chan struct{}) eventsDone := make(chan struct{}) sseHandlerDone := make(chan struct{}) @@ -60,6 +62,12 @@ func TestV2KBrainRelayBrowserReconnect(t *testing.T) { if !ok { t.Fatal("K-brain test server does not support streaming flush") } + // 提交确认收帧器会跳过无关帧，确认去重完成后才发送首批事件。 + select { + case <-releaseFirstEvents: + case <-r.Context().Done(): + return + } _, _ = fmt.Fprint(w, "data: {\"version\":\"kbrain.agent.v1\",\"seq\":1,\"conversation_id\":\"session-ws\",\"run_id\":\"kb-run-ws\",\"type\":\"run.accepted\"}\n\n") _, _ = fmt.Fprint(w, "data: {\"version\":\"kbrain.agent.v1\",\"seq\":2,\"conversation_id\":\"session-ws\",\"run_id\":\"kb-run-ws\",\"type\":\"assistant.text.delta\",\"payload\":{\"text\":\"first\"}}\n\n") flusher.Flush() @@ -77,7 +85,12 @@ func TestV2KBrainRelayBrowserReconnect(t *testing.T) { http.NotFound(w, r) } })) - defer kbrainServer.Close() + defer func() { + // 先释放夹具再关闭服务器，断言失败也不会留住 SSE 处理器。 + firstReleaseOnce.Do(func() { close(releaseFirstEvents) }) + releaseOnce.Do(func() { close(releaseEvents) }) + kbrainServer.Close() + }() cfg := newV2TestConfig() cfg.KBrainURL = kbrainServer.URL @@ -140,7 +153,12 @@ func TestV2KBrainRelayBrowserReconnect(t *testing.T) { t.Fatalf("duplicate chat accepted = %#v, want deduped run %q", duplicateAccepted, accepted.GetRunId()) } - <-firstEventWritten + firstReleaseOnce.Do(func() { close(releaseFirstEvents) }) + select { + case <-firstEventWritten: + case <-time.After(2 * time.Second): + t.Fatal("K-brain SSE did not emit first token after submit acknowledgements") + } var firstTokenSeq int64 for { frame := receiveWebFrame(t, browserConn) @@ -161,6 +179,14 @@ func TestV2KBrainRelayBrowserReconnect(t *testing.T) { t.Fatalf("first token seq = %d, want positive sequence", firstTokenSeq) } + // 浏览器断开后，独立观察 Gateway 接收事件的边界。 + // SSE Flush 或处理器返回只表示生产端已发送，不能证明中继已写入重放缓冲区。 + ingested := sm.SubscribeConversationStream(cfg.KBrainAgentID, conversationID, firstTokenSeq, streamEpoch) + if ingested == nil { + t.Fatal("Gateway ingestion subscription is nil") + } + defer ingested.Cleanup() + // Close only the browser connection. Relay and K-brain continue on their // own context, so the later events remain in the Gateway replay ring. _ = browserConn.Close() @@ -176,6 +202,23 @@ func TestV2KBrainRelayBrowserReconnect(t *testing.T) { t.Fatal("K-brain SSE handler did not return after release") } + ingestionTimeout := time.NewTimer(2 * time.Second) + defer ingestionTimeout.Stop() +waitForTerminal: + for { + select { + case event, ok := <-ingested.EventCh: + if !ok { + t.Fatal("Gateway ingestion subscription closed before terminal event") + } + if event.Type == "run_finished" { + break waitForTerminal + } + case <-ingestionTimeout.C: + t.Fatal("Gateway did not ingest terminal event after browser disconnect") + } + } + reconnected, reconnectCleanup := dialV2Path(t, mux, "/ws/v2") defer reconnectCleanup() helloV2(t, reconnected, cfg.Token) diff --git a/crates/agent-gui/test/chat/rolling-stat-text.test.mjs b/crates/agent-gui/test/chat/rolling-stat-text.test.mjs new file mode 100644 index 000000000..8ab9c3ba0 --- /dev/null +++ b/crates/agent-gui/test/chat/rolling-stat-text.test.mjs @@ -0,0 +1,73 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import test from "node:test"; +import { createDomTestEnv } from "../helpers/dom-test-env.mjs"; + +const env = await createDomTestEnv(); +const { React, act, createRoot } = env; +const { RollingStatText } = env.loadModule("@liveagent/ui/components/chat/RollingStatText.tsx"); + +async function mount(text) { + const container = env.dom.window.document.createElement("div"); + env.dom.window.document.body.appendChild(container); + const root = createRoot(container); + const update = async (value) => { + await act(async () => root.render(React.createElement(RollingStatText, { text: value }))); + }; + await update(text); + return { + container, + update, + unmount: async () => { + await act(async () => root.unmount()); + container.remove(); + }, + }; +} + +test("数字更新保留标签和未变位的节点，仅数字条改变位置", async () => { + const probe = await mount("Context 12.3% · ↑1.2M"); + try { + const labels = [...probe.container.querySelectorAll("[data-stat-label]")]; + const digits = [...probe.container.querySelectorAll("[data-current-digit]")]; + const positions = digits.map((digit) => digit.style.transform); + assert.equal(probe.container.textContent, "Context 12.3% · ↑1.2M"); + await probe.update("Context 12.4% · ↑1.2M"); + assert.equal(probe.container.textContent, "Context 12.4% · ↑1.2M"); + const next = [...probe.container.querySelectorAll("[data-current-digit]")]; + for (let i = 0; i < next.length; i += 1) { + assert.equal(next[i], digits[i], "数字位更新不能重新挂载"); + assert.equal(next[i].style.transform === positions[i], i !== 2); + } + assert.deepEqual([...probe.container.querySelectorAll("[data-stat-label]")], labels); + assert.equal(next[2].style.transform, "translateY(-4em)"); + assert.equal(probe.container.querySelectorAll("[data-digit]").length, 50); + } finally { + await probe.unmount(); + } +}); + +test("跨位数和单位变化保持右侧数字位身份，数字下降也正确", async () => { + const probe = await mount("Time 9s"); + try { + const ones = probe.container.querySelector("[data-current-digit]"); + await probe.update("Time 10s"); + const digits = [...probe.container.querySelectorAll("[data-current-digit]")]; + assert.equal(digits[1], ones); + assert.equal(probe.container.textContent, "Time 10s"); + await probe.update("Time 2m"); + assert.equal(probe.container.querySelector("[data-current-digit]"), ones); + assert.equal(ones.style.transform, "translateY(-2em)"); + assert.equal(probe.container.textContent, "Time 2m"); + await probe.update("上下文 0.0%"); + assert.equal(probe.container.textContent, "上下文 0.0%"); + } finally { + await probe.unmount(); + } +}); + +test("减少动态偏好禁用数字位移过渡，样式只过渡 transform", () => { + const css = readFileSync(new URL("../../../agent-ui/src/styles/animations.css", import.meta.url), "utf8"); + assert.match(css, /\.stat-digit-ready\s*\{\s*transition: transform var\(--ui-duration-160ms\) var\(--ease-ui-enter\);/); + assert.match(css, /@media \(prefers-reduced-motion: reduce\)\s*\{\s*\.stat-digit-ready\s*\{\s*transition: none;/); +}); diff --git a/crates/agent-gui/test/trajectory/conversation-stats-hook.test.mjs b/crates/agent-gui/test/trajectory/conversation-stats-hook.test.mjs index 523b597be..ff4f0feff 100644 --- a/crates/agent-gui/test/trajectory/conversation-stats-hook.test.mjs +++ b/crates/agent-gui/test/trajectory/conversation-stats-hook.test.mjs @@ -108,6 +108,81 @@ async function drain(rounds = 8) { } } +test("同会话刷新等待期间保留读数，完成后允许历史缩减，切会话不串数据", async () => { + clearConversationStatsCache(); + let release; + let delayed = false; + let pages = [turnEvents(1), turnEvents(2)]; + const host = { + loadWindow: async () => { + if (delayed) await new Promise((resolve) => { release = resolve; }); + return { + eventsJson: JSON.stringify(pages.flat()), + truncated: false, + oldestSegmentIndex: 0, + returnedSegmentCount: 1, + totalSegmentCount: 1, + hasMoreBefore: false, + }; + }, + }; + const probe = mountHook({ conversationId: "refresh-stable", host, liveEvents: [], enabled: true }); + try { + await probe.mount(); + await drain(); + assert.equal(probe.seen.current.stats.turns, 2); + delayed = true; + pages = [turnEvents(1)]; + await probe.update({ authoritativeRevision: 1 }); + assert.equal(probe.seen.current.stats?.turns, 2, "请求返回前不能变成空白或部分读数"); + await act(async () => { release(); }); + await drain(); + assert.equal(probe.seen.current.stats.turns, 1, "新数据必须允许历史截断后的下降"); + pages = []; + await probe.update({ conversationId: "other-empty" }); + assert.equal(probe.seen.current.stats, null, "切会话不能保留旧会话读数"); + await act(async () => { release(); }); + await drain(); + await probe.update({ enabled: false }); + assert.equal(probe.seen.current.stats, null, "禁用必须清空展示"); + } finally { + await probe.unmount(); + } +}); + +test("刷新失败保留上次读数，成功返回空历史时才清空", async () => { + clearConversationStatsCache(); + let fail = false; + let empty = false; + const original = createFakeHost([turnEvents(1)]).host; + const host = { + loadWindow: async (...args) => { + if (fail) throw new Error("测试刷新失败"); + const payload = await original.loadWindow(...args); + return empty ? { ...payload, eventsJson: "[]" } : payload; + }, + }; + const probe = mountHook({ conversationId: "refresh-error", host, liveEvents: [], enabled: true }); + const warn = console.warn; + try { + await probe.mount(); + await drain(); + fail = true; + console.warn = () => {}; + await probe.update({ authoritativeRevision: 1 }); + await drain(); + assert.equal(probe.seen.current.stats.turns, 1, "请求失败不能清空旧读数"); + fail = false; + empty = true; + await probe.update({ authoritativeRevision: 2 }); + await drain(); + assert.equal(probe.seen.current.stats, null, "成功的空历史不能残留旧读数"); + } finally { + console.warn = warn; + await probe.unmount(); + } +}); + test("首窗读数立即可用，后台分页把更早的段补齐", async () => { clearConversationStatsCache(); const pages = [turnEvents(1), turnEvents(2), turnEvents(3)]; @@ -351,3 +426,65 @@ test("liveOwnership 语义沿用轨迹视图：authoritative 空集收敛僵尸 ); await web.unmount(); }); + +for (const failedCursor of [2, 1]) { + test(`刷新在游标 ${failedCursor} 失败时保留完整读数，下一版本成功后才替换`, async () => { + clearConversationStatsCache(); + const pages = [turnEvents(1), turnEvents(2), turnEvents(3)]; + let phase = "initial"; + const calls = []; + const host = { + loadWindow: async (_id, cursor) => { + calls.push({ phase, cursor }); + if (phase === "initial" || phase === "recovered" || phase === "empty") { + const events = phase === "initial" ? pages.flat() : phase === "recovered" ? pages[2] : []; + return { eventsJson: JSON.stringify(events), oldestSegmentIndex: 0, hasMoreBefore: false, truncated: false }; + } + if (cursor === failedCursor) throw new Error("测试历史分页失败"); + const index = cursor === undefined ? 2 : cursor - 1; + return { eventsJson: JSON.stringify(pages[index]), oldestSegmentIndex: index, hasMoreBefore: index > 0, truncated: false }; + }, + }; + const probe = mountHook({ conversationId: `refresh-page-error-${failedCursor}`, host, liveEvents: [], enabled: true }); + const warn = console.warn; + try { + await probe.mount(); await drain(); + const previous = probe.seen.current.stats; + assert.equal(previous.turns, 3); + phase = "failed"; console.warn = () => {}; + await probe.update({ authoritativeRevision: 1 }); await drain(); + assert.equal(probe.seen.current.stats, previous, "分页错误不能用部分结果覆盖完整快照"); + assert.equal(probe.seen.current.loading, false); + const callsAfterFailure = calls.length; + await probe.update({ liveEvents: [] }); await drain(); + assert.equal(probe.seen.current.stats, previous); + assert.equal(calls.length, callsAfterFailure, "失败后不自动重试"); + phase = "recovered"; + await probe.update({ authoritativeRevision: 2 }); await drain(); + assert.equal(probe.seen.current.stats.turns, 1, "成功刷新允许历史缩减"); + assert.equal(probe.seen.current.stats.approximate, false); + phase = "empty"; + await probe.update({ authoritativeRevision: 3 }); await drain(); + assert.equal(probe.seen.current.stats, null, "成功空历史清空旧读数"); + } finally { console.warn = warn; await probe.unmount(); } + }); +} + +test("首次加载后续分页失败时保留已有窗口并标为近似", async () => { + clearConversationStatsCache(); + const host = { + loadWindow: async (_id, cursor) => { + if (cursor !== undefined) throw new Error("测试首次分页失败"); + return { eventsJson: JSON.stringify(turnEvents(3)), oldestSegmentIndex: 2, hasMoreBefore: true, truncated: false }; + }, + }; + const probe = mountHook({ conversationId: "initial-page-error", host, liveEvents: [], enabled: true }); + const warn = console.warn; + try { + console.warn = () => {}; + await probe.mount(); await drain(); + assert.equal(probe.seen.current.stats.turns, 1); + assert.equal(probe.seen.current.stats.approximate, true); + assert.equal(probe.seen.current.loading, false); + } finally { console.warn = warn; await probe.unmount(); } +}); diff --git a/crates/agent-ui/src/components/chat/ConversationStatsBar.tsx b/crates/agent-ui/src/components/chat/ConversationStatsBar.tsx index 196d5e052..503d621e7 100644 --- a/crates/agent-ui/src/components/chat/ConversationStatsBar.tsx +++ b/crates/agent-ui/src/components/chat/ConversationStatsBar.tsx @@ -16,6 +16,7 @@ import { } from "../../lib/trajectory/stats"; import { ConfirmActionPopover } from "../ui/confirm-action-popover"; import { LabelTooltip } from "../ui/label-tooltip"; +import { RollingStatText } from "./RollingStatText"; /** 心跳与 hook 的重建节流同频（docs/design/composer-context-stats-bar.md §4.2）。 */ const HEARTBEAT_MS = 1_000; @@ -374,7 +375,14 @@ export function ConversationStatsBar(props: { {index > 0 ? ( ))} diff --git a/crates/agent-ui/src/components/chat/RollingStatText.tsx b/crates/agent-ui/src/components/chat/RollingStatText.tsx new file mode 100644 index 000000000..86d3628e9 --- /dev/null +++ b/crates/agent-ui/src/components/chat/RollingStatText.tsx @@ -0,0 +1,46 @@ +import { useEffect, useState } from "react"; + +const DIGITS = Array.from({ length: 10 }, (_, digit) => digit); + +/** 单位、标点和标签留在原位；每一位数字独立滚动，右侧位数身份保持稳定。 */ +function RollingDigit({ digit }: { digit: string }) { + const [ready, setReady] = useState(false); + useEffect(() => setReady(true), []); + return ( +