From 33639e5b893250eb56654ac01198698fbf5e50d7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Wed, 26 Aug 2026 16:28:10 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=B5=8C=E5=A5=97=20i?= =?UTF-8?q?frame=20=E6=B1=A1=E6=9F=93=20embed=20=E4=BC=9A=E8=AF=9D?= =?UTF-8?q?=E4=B8=8E=E5=8D=8F=E5=8A=A9=E5=8D=A1=E5=90=8E=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E6=98=BE=E7=A4=BA=E5=BC=82=E5=B8=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 顶层窗口通过 topOwned 标记自愈老版嵌套页写脏的 embed.session；协助卡提交后避免冻结 checkpoint 覆盖实时 SSE，并仅在服务端回显新 human 消息时清除乐观气泡。 Co-authored-by: Cursor --- .../src/core/embed/embed-session.test.ts | 66 ++++- frontend-web/src/core/embed/embed-session.ts | 229 ++++++++++++------ frontend-web/src/core/threads/hooks.ts | 142 ++++++++--- .../src/core/threads/stream-display.test.ts | 89 +++++++ .../src/core/threads/stream-display.ts | 98 ++++++++ 5 files changed, 528 insertions(+), 96 deletions(-) create mode 100644 frontend-web/src/core/threads/stream-display.test.ts create mode 100644 frontend-web/src/core/threads/stream-display.ts diff --git a/frontend-web/src/core/embed/embed-session.test.ts b/frontend-web/src/core/embed/embed-session.test.ts index 8df2bd63..d3e867be 100644 --- a/frontend-web/src/core/embed/embed-session.test.ts +++ b/frontend-web/src/core/embed/embed-session.test.ts @@ -97,6 +97,51 @@ test("nested embed=1 does not hide chrome on the same-origin top window", () => }); }); +test("top parent reclaims pollution from an old nested iframe (no topOwned)", () => { + withSharedTab(({ storage, asTop }) => { + asTop(); + syncEmbedSessionFromSearch(""); + // 模拟未升级的老版内层：仍写共用键且没有 topOwned。 + storage.setItem( + "embed.session", + JSON.stringify({ + embed: true, + theme: "light", + hideJump: false, + hideHeader: false, + hideChrome: false, + isolate: false, + }), + ); + + // 外层点菜单重算布局时会读 session —— 必须立刻自愈。 + assert.equal(isEmbedActive(), false); + assert.equal(isChromeHidden(), false); + assert.equal(storage.getItem("embed.session"), null); + }); +}); + +test("top sync after old nested pollution does not mark owner", () => { + withSharedTab(({ storage, asTop }) => { + asTop(); + storage.setItem( + "embed.session", + JSON.stringify({ + embed: true, + theme: "light", + hideJump: false, + hideHeader: false, + hideChrome: false, + isolate: false, + }), + ); + syncEmbedSessionFromSearch(""); + assert.equal(isEmbedActive(), false); + assert.equal(isChromeHidden(), false); + assert.equal(storage.getItem("embed.session"), null); + }); +}); + test("nested iframe reclaims a polluted shared embed.session key", () => { withSharedTab(({ storage, asTop, asIframe }) => { asTop(); @@ -111,7 +156,6 @@ test("nested iframe reclaims a polluted shared embed.session key", () => { isolate: false, }), ); - assert.equal(isEmbedActive(), true); asIframe(); syncEmbedSessionFromSearch("?embed=1"); @@ -129,6 +173,7 @@ test("nested iframe keeps the top window's own embed session", () => { syncEmbedSessionFromSearch("?embed=1&theme=dark"); syncEmbedSessionFromSearch(""); assert.equal(isEmbedActive(), true); + assert.equal(readEmbedSession()?.topOwned, true); asIframe(); syncEmbedSessionFromSearch("?embed=1"); @@ -162,3 +207,22 @@ test("nested reclaim strips embed but keeps top isolate", () => { assert.equal(isIsolatedSession(), true); }); }); + +test("top reclaim strips embed but keeps isolate without nested help", () => { + withSharedTab(({ storage, asTop }) => { + asTop(); + storage.setItem( + "embed.session", + JSON.stringify({ + embed: true, + theme: "light", + hideJump: false, + hideHeader: false, + hideChrome: false, + isolate: true, + }), + ); + assert.equal(isEmbedActive(), false); + assert.equal(isIsolatedSession(), true); + }); +}); diff --git a/frontend-web/src/core/embed/embed-session.ts b/frontend-web/src/core/embed/embed-session.ts index 54b8aeaa..cb44090b 100644 --- a/frontend-web/src/core/embed/embed-session.ts +++ b/frontend-web/src/core/embed/embed-session.ts @@ -31,15 +31,18 @@ export { resolveEmbedThemeMode } from "./theme-mode.mjs"; * 分区，同一标签里所有同源文档（含嵌套 iframe、换路径/换端口反向代理后仍同源的另一套部署） * **共用一份**。全量模式任务管理用 iframe 套外部页、外部页再套咱们 `embed=1` 深链时，内层 * 若写入同一个 `embed.session` 键，外层点左侧菜单重算 `PageLayout` 就会把导航栏/侧边栏藏掉， - * 路由看起来却没变。因此顶层窗口写 `embed.session`，嵌套 iframe 写 `embed.session.nested`。 + * 路由看起来却没变。因此： + * - 顶层窗口写 `embed.session`，嵌套 iframe（新版）写 `embed.session.nested`； + * - 顶层只信任带 `topOwned` 的会话（本窗自己从 URL 置位）；老版嵌套页仍写共用键时， + * 外层父页会在读取/同步时清掉无主污染，即使内层未升级也能自愈。 */ const EMBED_SESSION_KEY = "embed.session"; /** 嵌套 iframe（`window !== top`）专用，避免与顶层全量页共用 `embed.session`。 */ const EMBED_SESSION_NESTED_KEY = "embed.session.nested"; /** - * 顶层窗口认领「本页自己的去 chrome 会话」的标记。嵌套同源 iframe 看到此标记就不会 - * 清掉顶层合法的 embed/hideChrome 持久化。 + * 顶层窗口认领「本页自己的去 chrome 会话」的内存标记（配合 session 里的 `topOwned`）。 + * 嵌套同源 iframe 看到此标记就不会清掉顶层合法的 embed/hideChrome 持久化。 */ const TOP_CHROME_OWNER_FLAG = "__deerflowTopChromeEmbed"; @@ -73,6 +76,33 @@ function embedSessionKey(): string { return isTopLevelWindow() ? EMBED_SESSION_KEY : EMBED_SESSION_NESTED_KEY; } +export type EmbedTheme = "light" | "dark"; + +/** 本标签页的嵌入会话（按需置位的若干独立开关）。 */ +export interface EmbedSession { + /** 嵌入模式：去全局 chrome + 强制主题。 */ + embed: boolean; + /** 嵌入模式下强制的主题（仅 embed 为真时有意义）。 */ + theme: EmbedTheme; + /** 隐藏深链底部「快捷跳转」栏（TaskJumpBar：底部按钮 + 右侧悬浮菜单）。 */ + hideJump: boolean; + /** Only hide the global Header while retaining PageSidebar and normal app behavior. */ + hideHeader: boolean; + /** + * 仅去全局 chrome（Header 导航栏 + PageSidebar 侧边栏），**不**强制主题、**不**隔离鉴权。 + * 区别于 `embed`：嵌入态会额外强制主题 + 走 sessionStorage 隔离登录身份；`hideChrome` 只想要 + * 一个干净的全屏页面（如「大屏绘制」`?hideChrome=1`），其余行为与正常标签页完全一致。 + */ + hideChrome: boolean; + /** 独立会话：鉴权身份按本标签页隔离（走 sessionStorage），用于同浏览器登录不同账号互不覆盖。 */ + isolate: boolean; + /** + * 仅顶层窗口自己从 URL 置位去 chrome 时写入。缺省/false 表示可能是同源嵌套 iframe + * （含未升级的老版内层）写脏的共用键，顶层父页应忽略并清掉。 + */ + topOwned?: boolean; +} + function parseEmbedSession(raw: string | null): EmbedSession | null { if (!raw) return null; try { @@ -84,6 +114,7 @@ function parseEmbedSession(raw: string | null): EmbedSession | null { hideHeader: p.hideHeader === true, hideChrome: p.hideChrome === true, isolate: p.isolate === true, + topOwned: p.topOwned === true, }; } catch { return null; @@ -99,14 +130,90 @@ function sessionHidesChrome(session: EmbedSession | null | undefined): boolean { ); } -/** 顶层窗口同步过自己的去 chrome 参数后打标，供同源嵌套 iframe 识别「这不是我写脏的」。 */ +function emptySession(): EmbedSession { + return { + embed: false, + theme: "light", + hideJump: false, + hideHeader: false, + hideChrome: false, + isolate: false, + topOwned: false, + }; +} + +function stripChromeFlags(session: EmbedSession): EmbedSession { + return { + ...session, + embed: false, + hideJump: false, + hideHeader: false, + hideChrome: false, + topOwned: false, + }; +} + +function persistTopSession(next: EmbedSession): void { + const store = getSessionStorage(); + if (!store) return; + if ( + !next.embed && + !next.hideJump && + !next.hideHeader && + !next.hideChrome && + !next.isolate + ) { + store.removeItem(EMBED_SESSION_KEY); + return; + } + const payload: EmbedSession = { + embed: next.embed, + theme: next.theme, + hideJump: next.hideJump, + hideHeader: next.hideHeader, + hideChrome: next.hideChrome, + isolate: next.isolate, + }; + if (next.topOwned) payload.topOwned = true; + store.setItem(EMBED_SESSION_KEY, JSON.stringify(payload)); +} + +function isTopChromeOwned(session?: EmbedSession | null): boolean { + const w = getWindow(); + if (w?.[TOP_CHROME_OWNER_FLAG] === true) return true; + return session?.topOwned === true; +} + +/** 顶层窗口从自己的 URL 置位去 chrome 后打标（内存 + session.topOwned）。 */ function markTopChromeOwner(session: EmbedSession): void { const w = getWindow(); if (!w || !isTopLevelWindow()) return; - if (!sessionHidesChrome(session)) return; + if (!sessionHidesChrome(session) || !session.topOwned) return; w[TOP_CHROME_OWNER_FLAG] = true; } +/** + * 顶层父页自愈：老版同源嵌套 iframe 仍会写共用 `embed.session`（无 `topOwned`）。 + * 若本窗从未从自己的 URL 认领过嵌入态，则清掉去 chrome 标志，保留 isolate。 + */ +function reclaimUnownedTopChrome(): void { + if (!isTopLevelWindow()) return; + try { + const store = getSessionStorage(); + if (!store) return; + const cur = parseEmbedSession(store.getItem(EMBED_SESSION_KEY)); + if (!cur || !sessionHidesChrome(cur)) return; + if (isTopChromeOwned(cur)) { + const w = getWindow(); + if (w) w[TOP_CHROME_OWNER_FLAG] = true; + return; + } + persistTopSession(stripChromeFlags(cur)); + } catch { + /* noop */ + } +} + /** * 旧代码把嵌套 iframe 的 embed 写进了共用的 `embed.session`。嵌套侧启动时把这份 * 遗产迁到 nested 键；若顶层并未认领去 chrome，再把顶层键里的 embed/hideChrome 清掉。 @@ -127,56 +234,18 @@ function reclaimPollutedTopChrome(): void { if (isTopLevelWindow()) return; try { const top = getWindow()?.top as TopChromeOwnerWindow | null | undefined; - if (!top || top[TOP_CHROME_OWNER_FLAG] === true) return; const store = getSessionStorage(); if (!store) return; const cur = parseEmbedSession(store.getItem(EMBED_SESSION_KEY)); if (!cur) return; - const next: EmbedSession = { - ...cur, - embed: false, - hideJump: false, - hideHeader: false, - hideChrome: false, - }; - if ( - !next.embed && - !next.hideJump && - !next.hideHeader && - !next.hideChrome && - !next.isolate - ) { - store.removeItem(EMBED_SESSION_KEY); - return; - } - store.setItem(EMBED_SESSION_KEY, JSON.stringify(next)); + // 顶层已认领（内存标记或 session.topOwned）则不动。 + if (top?.[TOP_CHROME_OWNER_FLAG] === true || cur.topOwned === true) return; + persistTopSession(stripChromeFlags(cur)); } catch { /* 跨源 top：sessionStorage 本就不与外层其它 origin 共享，无需回收 */ } } -export type EmbedTheme = "light" | "dark"; - -/** 本标签页的嵌入会话（按需置位的若干独立开关）。 */ -export interface EmbedSession { - /** 嵌入模式：去全局 chrome + 强制主题。 */ - embed: boolean; - /** 嵌入模式下强制的主题（仅 embed 为真时有意义）。 */ - theme: EmbedTheme; - /** 隐藏深链底部「快捷跳转」栏（TaskJumpBar：底部按钮 + 右侧悬浮菜单）。 */ - hideJump: boolean; - /** Only hide the global Header while retaining PageSidebar and normal app behavior. */ - hideHeader: boolean; - /** - * 仅去全局 chrome（Header 导航栏 + PageSidebar 侧边栏），**不**强制主题、**不**隔离鉴权。 - * 区别于 `embed`：嵌入态会额外强制主题 + 走 sessionStorage 隔离登录身份；`hideChrome` 只想要 - * 一个干净的全屏页面（如「大屏绘制」`?hideChrome=1`），其余行为与正常标签页完全一致。 - */ - hideChrome: boolean; - /** 独立会话：鉴权身份按本标签页隔离（走 sessionStorage），用于同浏览器登录不同账号互不覆盖。 */ - isolate: boolean; -} - /** URL 布尔参数的三态：显式开 / 显式关 / 未带。 */ type FlagParam = "on" | "off" | null; @@ -223,9 +292,20 @@ export function readEmbedThemeParam(search: string): string { return (toSearchParams(search).get("theme") ?? "").trim(); } +function searchEnablesChromeHide(search: string): boolean { + return ( + readEmbedParam(search) === "on" || + readHideChromeParam(search) === "on" || + readHideHeaderParam(search) === "on" || + readHideJumpParam(search) === "on" + ); +} + /** 读本浏览上下文的嵌入会话；无 → null。顶层与嵌套 iframe 使用不同的 sessionStorage 键。 */ export function readEmbedSession(): EmbedSession | null { try { + // 顶层每次读取前清掉老版内层写脏的无主 embed，保证点菜单重算布局时不会去 chrome。 + reclaimUnownedTopChrome(); return parseEmbedSession(getSessionStorage()?.getItem(embedSessionKey()) ?? null); } catch { return null; @@ -238,11 +318,26 @@ function writeEmbedSession(next: EmbedSession): void { const store = getSessionStorage(); if (!store) return; const key = embedSessionKey(); - if (!next.embed && !next.hideJump && !next.hideHeader && !next.hideChrome && !next.isolate) { + if ( + !next.embed && + !next.hideJump && + !next.hideHeader && + !next.hideChrome && + !next.isolate + ) { store.removeItem(key); return; } - store.setItem(key, JSON.stringify(next)); + const payload: EmbedSession = { + embed: next.embed, + theme: next.theme, + hideJump: next.hideJump, + hideHeader: next.hideHeader, + hideChrome: next.hideChrome, + isolate: next.isolate, + }; + if (next.topOwned) payload.topOwned = true; + store.setItem(key, JSON.stringify(payload)); } catch { /* noop */ } @@ -295,19 +390,15 @@ export function isIsolatedSession(): boolean { * 按 URL 参数同步 sessionStorage 会话（AppProviders 在每次 search 变化时调用一次）。 * **合并语义**：on 置位、off 清位、未带则沿用既有值——这样登录后跳到不带参数的干净路由时， * 之前置位的开关不会被抹掉。 + * + * 顶层额外规则：只有本窗 URL 显式开过嵌套开关时才把会话标为 `topOwned`；未认领的 + * chrome 标志一律丢弃（挡住老版同源嵌套 iframe 对共用键的污染）。 */ export function syncEmbedSessionFromSearch(search: string): void { migrateLegacySharedSessionIfNested(); - const cur = - readEmbedSession() ?? - { - embed: false, - theme: "light" as EmbedTheme, - hideJump: false, - hideHeader: false, - hideChrome: false, - isolate: false, - }; + if (isTopLevelWindow()) reclaimUnownedTopChrome(); + + const cur = readEmbedSession() ?? emptySession(); const next: EmbedSession = { ...cur }; const embedParam = readEmbedParam(search); @@ -334,6 +425,17 @@ export function syncEmbedSessionFromSearch(search: string): void { if (isolateParam === "on") next.isolate = true; else if (isolateParam === "off") next.isolate = false; + if (isTopLevelWindow()) { + if (searchEnablesChromeHide(search)) { + next.topOwned = true; + } else if (!isTopChromeOwned(next)) { + // 未认领：丢掉可能由老版内层写脏的 chrome 标志，保留 isolate。 + Object.assign(next, stripChromeFlags(next), { isolate: next.isolate }); + } else { + next.topOwned = true; + } + } + writeEmbedSession(next); if (isTopLevelWindow()) markTopChromeOwner(next); else reclaimPollutedTopChrome(); @@ -378,15 +480,6 @@ export function resolveForcedEmbedTheme(search: string): EmbedTheme | undefined * `migrateAuthToSession()` 负责，调用顺序见 appearance 设置页。 */ export function setIsolatedSession(on: boolean): void { - const cur = - readEmbedSession() ?? - { - embed: false, - theme: "light" as EmbedTheme, - hideJump: false, - hideHeader: false, - hideChrome: false, - isolate: false, - }; + const cur = readEmbedSession() ?? emptySession(); writeEmbedSession({ ...cur, isolate: on }); } diff --git a/frontend-web/src/core/threads/hooks.ts b/frontend-web/src/core/threads/hooks.ts index 8f19906e..1593f32d 100644 --- a/frontend-web/src/core/threads/hooks.ts +++ b/frontend-web/src/core/threads/hooks.ts @@ -24,6 +24,12 @@ import type { UploadedFileInfo } from "../uploads"; import { promptInputFilePartToFile, uploadFiles } from "../uploads"; import type { AgentThread, AgentThreadState } from "./types"; +import { + collectHumanMessageIds, + hasServerEchoedOptimisticHuman, + mergeStreamDisplayValues, + shouldDiscardTerminalOverlay, +} from "./stream-display"; import { isAbortLikeError, isTransientStreamTransportError, @@ -282,6 +288,19 @@ export async function stopActiveThreadBeforeClarificationReply( storedRunId = null; } } + // 协助卡出现时图已经 jump_to=end，run 多半已是 terminal。再 stop()/cancel + // 会再触发 onFinish → getState overlay，把刚发出的用户回复冻住、整段攒着出。 + if (storedRunId && threadId) { + try { + const run = await getAPIClient().runs.get(threadId, storedRunId); + if (run && !isInflightRunStatus(run.status)) { + clearActiveRunId(threadId); + return; + } + } catch { + /* 404 / 网络：按下面的 best-effort stop 走 */ + } + } if (!thread.isLoading && !storedRunId) return; try { await stopThreadRun(threadId); @@ -580,6 +599,16 @@ export function useThreadStream({ threadId && typeof window !== "undefined" && readActiveRunId(threadId), ), ); + const sendInFlightRef = useRef(false); + const sendEpochRef = useRef(0); + // Filled after useStream; onFinish/onError may read it asynchronously. + const threadStreamRef = useRef<{ + isLoading: boolean; + stop: () => void | Promise; + joinStream: (runId: string) => Promise; + messages: Message[]; + values?: AgentThreadState; + } | null>(null); const markStreamActivity = useCallback(() => { lastStreamActivityAtRef.current = Date.now(); @@ -867,11 +896,21 @@ export function useThreadStream({ // Do not trust a clean SSE close as "the run is done". streamResumable // chats keep the backend run alive on disconnect; the SDK still fires // onFinish and would otherwise restore the send button. + // + // If a newer turn is already streaming (common: 用户协助卡提交后，上一轮 + // interrupt 的迟到 onFinish 才到), do not freeze/rejoin — that would + // steal the live SSE and dump tokens only at the end. + if (threadStreamRef.current?.isLoading || sendInFlightRef.current) { + return; + } setStreamHoldOpen(true); void confirmOrRejoinRef.current("finish").then((result) => { if (result === "rejoined") { return; } + if (threadStreamRef.current?.isLoading || sendInFlightRef.current) { + return; + } setStreamHoldOpen(false); listeners.current.onFinish?.(state.values); void queryClient.invalidateQueries({ queryKey: ["threads", "search"] }); @@ -885,7 +924,6 @@ export function useThreadStream({ const [threadValueOverrides, setThreadValueOverrides] = useState< Partial >({}); - const sendInFlightRef = useRef(false); // Carries the ask time of an in-flight send until the server-assigned human // message id shows up, so we can bind the timestamp to the real id (see the // reconciliation effect below). ``known`` is the set of human ids that @@ -893,13 +931,12 @@ export function useThreadStream({ const pendingAskTimeRef = useRef<{ ts: string; known: Set } | null>( null, ); + // Human ids known at optimistic-send time; used to clear optimistic only when + // a *new* server human appears (not when length bumps from tool/checkpoint). + const optimisticKnownHumanIdsRef = useRef>(new Set()); // Track message count before sending so we know when server has responded const prevMsgCountRef = useRef(thread.messages.length); - // Keep a ref to the latest `thread` so the threadId-change effect can stop - // an in-flight stream without re-running on every render that produces a - // new `thread` identity. - const threadStreamRef = useRef(thread); threadStreamRef.current = thread; const confirmOrRejoin = useCallback( @@ -915,18 +952,39 @@ export function useThreadStream({ if (streamReconciliationInFlightRef.current && reason === "poll") { return "unknown"; } + // Stale finish/error from a prior interrupt must not joinStream over a + // turn the user already started (协助卡提交后的实时流会被打断并攒批)。 + if ( + (reason === "finish" || reason === "error") && + (sendInFlightRef.current || threadStreamRef.current?.isLoading) + ) { + return "rejoined"; + } streamReconciliationInFlightRef.current = true; + const epochAtStart = sendEpochRef.current; try { const client = getAPIClient(isMock); let runId = activeRunIdRef.current ?? readActiveRunId(tid); + const terminalRunId = runId; const applyTerminalState = async (status: string) => { + const finalState = await client.threads.getState(tid); + if ( + shouldDiscardTerminalOverlay({ + epochAtStart, + currentEpoch: sendEpochRef.current, + sendInFlight: sendInFlightRef.current, + terminalRunId, + currentRunId: activeRunIdRef.current, + }) + ) { + return; + } clearActiveRunId(tid); activeRunIdRef.current = null; setStreamHoldOpen(false); streamReconnectAttemptsRef.current = 0; streamReconnectFailureToastShownRef.current = false; - const finalState = await client.threads.getState(tid); setThreadValueOverrides( finalState.values as Partial, ); @@ -985,7 +1043,7 @@ export function useThreadStream({ const streamAboutToEnd = reason === "finish" || reason === "error"; if ( - threadStreamRef.current.isLoading && + threadStreamRef.current?.isLoading && !streamAboutToEnd && reason !== "conflict" ) { @@ -1013,6 +1071,9 @@ export function useThreadStream({ streamReconnectAttemptsRef.current += 1; lastStreamActivityAtRef.current = Date.now(); const stream = threadStreamRef.current; + if (!stream) { + return "unknown"; + } runMetadataStorageRef.current?.removeItem(`lg:stream:${tid}`); void Promise.resolve(stream.stop()) .catch(() => undefined) @@ -1042,6 +1103,9 @@ export function useThreadStream({ lastStreamActivityAtRef.current = Date.now(); const stream = threadStreamRef.current; + if (!stream) { + return "unknown"; + } void stream.joinStream(runId).catch(() => { /* poll retries while the run stays non-terminal */ }); @@ -1063,7 +1127,7 @@ export function useThreadStream({ let disposed = false; const reconcileRun = async () => { if (disposed) return; - const dropped = !threadStreamRef.current.isLoading; + const dropped = !threadStreamRef.current?.isLoading; const idleFor = Date.now() - lastStreamActivityAtRef.current; if (!dropped && idleFor < STREAM_RECONCILIATION_IDLE_MS) { return; @@ -1072,7 +1136,7 @@ export function useThreadStream({ return; } const stored = activeRunIdRef.current ?? readActiveRunId(activeThreadId); - if (!stored && !streamHoldOpen && threadStreamRef.current.isLoading) { + if (!stored && !streamHoldOpen && threadStreamRef.current?.isLoading) { return; } if (!stored && !streamHoldOpen) { @@ -1120,7 +1184,9 @@ export function useThreadStream({ } startedRef.current = false; sendInFlightRef.current = false; + sendEpochRef.current += 1; prevMsgCountRef.current = 0; + optimisticKnownHumanIdsRef.current = new Set(); activeRunIdRef.current = null; setStreamHoldOpen(false); setOptimisticMessages([]); @@ -1207,15 +1273,22 @@ export function useThreadStream({ threadValueOverrides.title_provisional, ]); - // Clear optimistic when server messages arrive (count increases) + // Clear optimistic only when the server echoes a *new* human message. + // After 用户协助, a late checkpoint / tool result can bump length without + // the new answer yet — clearing then makes the bubble flash and vanish. useEffect(() => { + if (optimisticMessages.length === 0) { + return; + } if ( - optimisticMessages.length > 0 && - thread.messages.length > prevMsgCountRef.current + hasServerEchoedOptimisticHuman( + thread.messages, + optimisticKnownHumanIdsRef.current, + ) ) { setOptimisticMessages([]); } - }, [thread.messages.length, optimisticMessages.length]); + }, [thread.messages, optimisticMessages.length]); // Bind a just-sent question's ask time to the server-assigned human message // id once it appears. Persists to localStorage so the time shows on every @@ -1248,22 +1321,36 @@ export function useThreadStream({ return; } sendInFlightRef.current = true; + sendEpochRef.current += 1; + // Drop frozen checkpoint messages so the new SSE can paint live tokens. + // Keep title / title_provisional overlays from the prior reconcile. + setThreadValueOverrides((prev) => { + if (!("messages" in prev)) { + return prev; + } + const { messages: _dropped, ...rest } = prev; + return rest; + }); + setStreamHoldOpen(false); const text = message.text.trim(); // Wall-clock time the question was asked. Stored on the human message so // the chat can show "几点提问" and have it survive a reload (it round-trips // through the checkpoint inside additional_kwargs, same as `files`). const clientTs = new Date().toISOString(); + const knownHumanIds = collectHumanMessageIds( + thread.messages, + Array.isArray(thread.values?.messages) + ? (thread.values.messages as Message[]) + : undefined, + ); // Remember which human messages already exist so the reconciliation // effect can bind ``clientTs`` to the new one once the server returns it. pendingAskTimeRef.current = { ts: clientTs, - known: new Set( - thread.messages - .filter((m) => m.type === "human" && m.id) - .map((m) => m.id as string), - ), + known: knownHumanIds, }; + optimisticKnownHumanIdsRef.current = knownHumanIds; // Capture current count before showing optimistic messages prevMsgCountRef.current = thread.messages.length; @@ -1535,7 +1622,7 @@ export function useThreadStream({ persistActiveRunId(tid, runId); } try { - await Promise.resolve(threadStreamRef.current.stop()); + await Promise.resolve(threadStreamRef.current?.stop()); } catch { /* local abort is best-effort */ } @@ -1549,17 +1636,18 @@ export function useThreadStream({ } }, [isMock]); - // Merge thread with optimistic messages for display + // Merge thread with optimistic messages for display. While a live SSE is + // open, never let a stale getState snapshot's `messages` win — that is what + // made 协助卡提交后「发送中」攒到结束才一口气出字。 const mergedValues = - Object.keys(threadValueOverrides).length > 0 - ? { - ...thread.values, - ...threadValueOverrides, - } - : thread.values; + mergeStreamDisplayValues( + thread.values as AgentThreadState | undefined, + threadValueOverrides, + thread.isLoading, + ) ?? thread.values; const mergedMessages = - Array.isArray(mergedValues.messages) + Array.isArray(mergedValues?.messages) ? (mergedValues.messages as Message[]) : thread.messages; diff --git a/frontend-web/src/core/threads/stream-display.test.ts b/frontend-web/src/core/threads/stream-display.test.ts new file mode 100644 index 00000000..bc3216d3 --- /dev/null +++ b/frontend-web/src/core/threads/stream-display.test.ts @@ -0,0 +1,89 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import type { Message } from "@langchain/langgraph-sdk"; + +import { + collectHumanMessageIds, + hasServerEchoedOptimisticHuman, + mergeStreamDisplayValues, + shouldDiscardTerminalOverlay, +} from "./stream-display"; + +test("live streaming ignores frozen checkpoint messages so tokens can paint", () => { + const live = { + title: "live", + messages: [{ type: "human", id: "h1", content: "A" }], + }; + const overrides = { + title: "frozen", + messages: [{ type: "human", id: "h1", content: "A" }], + }; + + const duringStream = mergeStreamDisplayValues(live, overrides, true); + assert.deepEqual(duringStream?.messages, live.messages); + assert.equal(duringStream?.title, "frozen"); + + const afterStream = mergeStreamDisplayValues(live, overrides, false); + assert.equal(afterStream?.title, "frozen"); + assert.deepEqual(afterStream?.messages, overrides.messages); +}); + +test("a later send discards the previous run's getState overlay", () => { + assert.equal( + shouldDiscardTerminalOverlay({ + epochAtStart: 1, + currentEpoch: 2, + sendInFlight: false, + terminalRunId: "old-run", + currentRunId: "old-run", + }), + true, + ); + assert.equal( + shouldDiscardTerminalOverlay({ + epochAtStart: 2, + currentEpoch: 2, + sendInFlight: true, + terminalRunId: "old-run", + currentRunId: "old-run", + }), + true, + ); + assert.equal( + shouldDiscardTerminalOverlay({ + epochAtStart: 2, + currentEpoch: 2, + sendInFlight: false, + terminalRunId: "old-run", + currentRunId: "new-run", + }), + true, + ); + assert.equal( + shouldDiscardTerminalOverlay({ + epochAtStart: 1, + currentEpoch: 1, + sendInFlight: false, + terminalRunId: "run-1", + currentRunId: "run-1", + }), + false, + ); +}); + +test("optimistic human stays until a new server human id appears", () => { + const known = collectHumanMessageIds([ + { type: "human", id: "h1", content: "first" } as Message, + ]); + const afterToolOnly = [ + { type: "human", id: "h1", content: "first" }, + { type: "tool", id: "t1", content: "OK" }, + ] as Message[]; + assert.equal(hasServerEchoedOptimisticHuman(afterToolOnly, known), false); + + const afterEcho = [ + ...afterToolOnly, + { type: "human", id: "h2", content: "按方案 A" }, + ] as Message[]; + assert.equal(hasServerEchoedOptimisticHuman(afterEcho, known), true); +}); diff --git a/frontend-web/src/core/threads/stream-display.ts b/frontend-web/src/core/threads/stream-display.ts new file mode 100644 index 00000000..23690af2 --- /dev/null +++ b/frontend-web/src/core/threads/stream-display.ts @@ -0,0 +1,98 @@ +import type { Message } from "@langchain/langgraph-sdk"; + +/** + * After a run ends (especially an interrupt / 用户协助卡), `confirmOrRejoin` + * snapshots `threads.getState()` into an overlay so a premature SSE close can + * still show the checkpoint. That overlay must never win over a *new* live + * stream: `overrides.messages` would freeze the previous turn and dump the + * next answer only when the next `getState()` lands. + */ +export function mergeStreamDisplayValues>( + live: T | undefined, + overrides: Partial, + isLiveStreaming: boolean, +): T | undefined { + const overrideKeys = Object.keys(overrides); + if (overrideKeys.length === 0) { + return live; + } + if (isLiveStreaming) { + const { messages: _ignored, ...rest } = overrides as Partial & { + messages?: unknown; + }; + if (Object.keys(rest).length === 0) { + return live; + } + return { ...(live as T), ...rest }; + } + return { ...(live as T), ...overrides }; +} + +export function shouldDiscardTerminalOverlay(opts: { + epochAtStart: number; + currentEpoch: number; + sendInFlight: boolean; + terminalRunId: string | null; + currentRunId: string | null; +}): boolean { + if (opts.currentEpoch !== opts.epochAtStart) { + return true; + } + if (opts.sendInFlight) { + return true; + } + if ( + opts.currentRunId && + opts.terminalRunId && + opts.currentRunId !== opts.terminalRunId + ) { + return true; + } + return false; +} + +export function collectHumanMessageIds( + ...groups: Array +): Set { + const ids = new Set(); + for (const messages of groups) { + if (!Array.isArray(messages)) continue; + for (const message of messages) { + if ( + message?.type === "human" && + typeof message.id === "string" && + message.id && + !message.id.startsWith("opt-") + ) { + ids.add(message.id); + } + } + } + return ids; +} + +/** + * Drop the optimistic human bubble only when the server has echoed a *new* + * human message. A late checkpoint / tool result that merely increases + * `messages.length` must not make the user's input vanish. + */ +export function hasServerEchoedOptimisticHuman( + serverMessages: Message[] | undefined, + knownHumanIds: Set, +): boolean { + if (!Array.isArray(serverMessages)) { + return false; + } + for (const message of serverMessages) { + if ( + message?.type === "human" && + typeof message.id === "string" && + message.id && + !message.id.startsWith("opt-") && + !knownHumanIds.has(message.id) + ) { + return true; + } + } + return false; +}