fix: 修复嵌套 iframe 污染 embed 会话与协助卡后流式显示异常

顶层窗口通过 topOwned 标记自愈老版嵌套页写脏的 embed.session;协助卡提交后避免冻结 checkpoint 覆盖实时 SSE,并仅在服务端回显新 human 消息时清除乐观气泡。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
董宏凯andCursor committed 2026-08-26 16:28:10 +08:00
1 parent eba3008482
commit 33639e5b89
5 files changed
+528 -96

No files matched your search

@@ -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);
});
});
+161 -68
View File
@@ -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 });
}
+115 -27
View File
@@ -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<void>;
joinStream: (runId: string) => Promise<unknown>;
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<AgentThreadState>
>({});
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<string> } | 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<Set<string>>(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<AgentThreadState>,
);
@@ -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;
@@ -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);
});
@@ -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<T extends Record<string, unknown>>(
live: T | undefined,
overrides: Partial<T>,
isLiveStreaming: boolean,
): T | undefined {
const overrideKeys = Object.keys(overrides);
if (overrideKeys.length === 0) {
return live;
}
if (isLiveStreaming) {
const { messages: _ignored, ...rest } = overrides as Partial<T> & {
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<Message[] | undefined>
): Set<string> {
const ids = new Set<string>();
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<string>,
): 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;
}