From 2dbaeef693ae69da1298769b1f27743fdfef9988 Mon Sep 17 00:00:00 2001 From: Josh Lehman Date: Tue, 25 Aug 2026 17:27:41 -0700 Subject: [PATCH] fix(ui): keep active commentary after session navigation (#129640) * fix(ui): restore active commentary after navigation * test(ui): wait for responsive activity layout --- .../OpenClawProtocol/GatewayModels.swift | 4 + .../src/schema/logs-chat.test.ts | 6 + .../gateway-protocol/src/schema/logs-chat.ts | 1 + .../server-methods/chat-history-handler.ts | 16 +- .../server.chat-history-cursor.test.ts | 36 ++++ .../activity-session-feed.capture.e2e.test.ts | 19 +- .../e2e/chat-active-cursor-replay.e2e.test.ts | 181 ++++++++++++++++ ui/src/pages/chat/chat-history-cursor.test.ts | 84 ++++++++ ui/src/pages/chat/chat-history.ts | 198 ++++++++++-------- ui/src/pages/chat/session-prefetch.test.ts | 69 ++++++ ui/src/pages/chat/session-prefetch.ts | 10 +- 11 files changed, 528 insertions(+), 96 deletions(-) create mode 100644 ui/src/e2e/chat-active-cursor-replay.e2e.test.ts diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift index d3ba872eb0ef..943c5864a986 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift @@ -19753,6 +19753,7 @@ public struct ChatHistoryDeltaResult: Codable, Sendable { public let deltacursor: String public let sessioninfo: AnyCodable public let agentslist: AnyCodable? + public let inflightrun: AnyCodable? public let metadata: AnyCodable? public init( @@ -19761,6 +19762,7 @@ public struct ChatHistoryDeltaResult: Codable, Sendable { deltacursor: String, sessioninfo: AnyCodable, agentslist: AnyCodable? = nil, + inflightrun: AnyCodable? = nil, metadata: AnyCodable? = nil) { self.kind = kind @@ -19768,6 +19770,7 @@ public struct ChatHistoryDeltaResult: Codable, Sendable { self.deltacursor = deltacursor self.sessioninfo = sessioninfo self.agentslist = agentslist + self.inflightrun = inflightrun self.metadata = metadata } @@ -19777,6 +19780,7 @@ public struct ChatHistoryDeltaResult: Codable, Sendable { case deltacursor = "deltaCursor" case sessioninfo = "sessionInfo" case agentslist = "agentsList" + case inflightrun = "inFlightRun" case metadata } } diff --git a/packages/gateway-protocol/src/schema/logs-chat.test.ts b/packages/gateway-protocol/src/schema/logs-chat.test.ts index a78059d24b1a..115fa16ab5c0 100644 --- a/packages/gateway-protocol/src/schema/logs-chat.test.ts +++ b/packages/gateway-protocol/src/schema/logs-chat.test.ts @@ -58,6 +58,12 @@ describe("ChatHistoryCursorResultSchema", () => { sessionInfo, }; expect(Value.Check(ChatHistoryCursorResultSchema, delta)).toBe(true); + expect( + Value.Check(ChatHistoryCursorResultSchema, { + ...delta, + inFlightRun: { runId: "run-live", text: "still working" }, + }), + ).toBe(true); expect(Value.Check(ChatHistoryCursorResultSchema, { kind: "reset" })).toBe(true); expect(Value.Check(ChatHistoryCursorResultSchema, { ...delta, extra: true })).toBe(false); expect(Value.Check(ChatHistoryCursorResultSchema, { kind: "reset", messages: [] })).toBe(false); diff --git a/packages/gateway-protocol/src/schema/logs-chat.ts b/packages/gateway-protocol/src/schema/logs-chat.ts index f48dfb0e1376..1422c446f0e1 100644 --- a/packages/gateway-protocol/src/schema/logs-chat.ts +++ b/packages/gateway-protocol/src/schema/logs-chat.ts @@ -45,6 +45,7 @@ export const ChatHistoryDeltaResultSchema = closedObject({ deltaCursor: Type.String(), sessionInfo: Type.Unknown(), agentsList: Type.Optional(Type.Unknown()), + inFlightRun: Type.Optional(Type.Unknown()), metadata: Type.Optional(Type.Unknown()), }); diff --git a/src/gateway/server-methods/chat-history-handler.ts b/src/gateway/server-methods/chat-history-handler.ts index 83ccc29509ae..a01f5736598d 100644 --- a/src/gateway/server-methods/chat-history-handler.ts +++ b/src/gateway/server-methods/chat-history-handler.ts @@ -453,11 +453,6 @@ async function handleChatHistoryRequest({ agentId: activeRunAgentId, defaultAgentId: compatibilityOwnerAgentId, }); - const boundedInFlightRun = boundInFlightRunSnapshotForChatHistory({ - snapshot: inFlightRun, - messages: capped, - maxBytes: maxHistoryBytes, - }); if (cursor !== undefined) { if (!sessionId || !storePath || resolveClaudeCliBindingSessionId(entry)) { respond(true, { kind: "reset" }); @@ -501,15 +496,26 @@ async function handleChatHistoryRequest({ return; } sessionInfo.activeLeafEntryId = delta.activeLeafEntryId; + const boundedInFlightRun = boundInFlightRunSnapshotForChatHistory({ + snapshot: inFlightRun, + messages: delta.messages, + maxBytes: maxHistoryBytes, + }); respond(true, { kind: "delta", messages: delta.messages, deltaCursor: delta.deltaCursor, sessionInfo, + ...(boundedInFlightRun ? { inFlightRun: boundedInFlightRun } : {}), ...(startupMetadata ? { metadata: startupMetadata } : {}), }); return; } + const boundedInFlightRun = boundInFlightRunSnapshotForChatHistory({ + snapshot: inFlightRun, + messages: capped, + maxBytes: maxHistoryBytes, + }); const payload = { sessionKey, sessionId, diff --git a/src/gateway/server.chat-history-cursor.test.ts b/src/gateway/server.chat-history-cursor.test.ts index c12481e5256b..96064a545779 100644 --- a/src/gateway/server.chat-history-cursor.test.ts +++ b/src/gateway/server.chat-history-cursor.test.ts @@ -158,6 +158,42 @@ describe("chat.history cursor catch-up", () => { }); }); + test.each(["chat.history", "chat.startup"] as const)( + "%s returns the active run snapshot with an empty cached delta", + async (method) => { + const { context } = await createCursorSession(); + context.chatAbortControllers.set("run-active", { + controller: new AbortController(), + sessionId, + sessionKey: "main", + startedAtMs: 1_000, + expiresAtMs: Date.now() + 60_000, + projectSessionActive: true, + }); + context.chatRunState.getOrCreate("run-active").buffer = "still working"; + const page = await callChat<{ deltaCursor?: string }>(context, method); + + const delta = await callChat<{ + inFlightRun?: unknown; + kind?: string; + messages?: unknown[]; + }>(context, method, { cursor: page.payload?.deltaCursor }); + + expect(delta).toMatchObject({ + ok: true, + payload: { + kind: "delta", + messages: [], + inFlightRun: { + runId: "run-active", + text: "still working", + startedAt: 1_000, + }, + }, + }); + }, + ); + test("does not advance a cursor past messages appended after the projection check", async () => { const { context, storePath } = await createCursorSession(); const scope = currentScope(storePath); diff --git a/ui/src/e2e/activity-session-feed.capture.e2e.test.ts b/ui/src/e2e/activity-session-feed.capture.e2e.test.ts index 55776ceea084..7ddd07f3df3b 100644 --- a/ui/src/e2e/activity-session-feed.capture.e2e.test.ts +++ b/ui/src/e2e/activity-session-feed.capture.e2e.test.ts @@ -352,14 +352,19 @@ suite.define(() => { .evaluateAll((buttons) => buttons.map((button) => Math.round(button.getBoundingClientRect().top)), ); - const [timeFilterBox, peopleControlBox] = await Promise.all([ - timeFilter.boundingBox(), - peopleControl.boundingBox(), - ]); expect(new Set(timeButtonTops)).toHaveLength(1); - expect(timeFilterBox).not.toBeNull(); - expect(peopleControlBox).not.toBeNull(); - expect(Math.abs(timeFilterBox!.y - peopleControlBox!.y)).toBeLessThan(2); + await expect + .poll(async () => { + const [timeFilterBox, peopleControlBox] = await Promise.all([ + timeFilter.boundingBox(), + peopleControl.boundingBox(), + ]); + if (!timeFilterBox || !peopleControlBox) { + return Number.POSITIVE_INFINITY; + } + return Math.abs(timeFilterBox.y - peopleControlBox.y); + }) + .toBeLessThan(2); const automationGroupChildTops = await automationGroup .locator(":scope > *") .evaluateAll((children) => diff --git a/ui/src/e2e/chat-active-cursor-replay.e2e.test.ts b/ui/src/e2e/chat-active-cursor-replay.e2e.test.ts new file mode 100644 index 000000000000..54f87d7f81b5 --- /dev/null +++ b/ui/src/e2e/chat-active-cursor-replay.e2e.test.ts @@ -0,0 +1,181 @@ +import { mkdir } from "node:fs/promises"; +import path from "node:path"; +import { expect, it } from "vitest"; +import { + chatSessionListResponse, + createChatFlowE2eSuite, + expectDefined, + installMockGateway, + requireRecord, +} from "./chat-flow.test-support.ts"; + +const suite = createChatFlowE2eSuite(); + +suite.define(() => { + it("restores active commentary when an evicted session revalidates from its cursor", async () => { + const artifactDir = process.env.OPENCLAW_UI_E2E_ARTIFACT_DIR?.trim(); + if (artifactDir) { + await mkdir(artifactDir, { recursive: true }); + } + const context = await suite.newBrowserContext({ + locale: "en-US", + ...(artifactDir + ? { recordVideo: { dir: artifactDir, size: { height: 900, width: 1280 } } } + : {}), + serviceWorkers: "block", + viewport: { height: 900, width: 1280 }, + }); + const page = await context.newPage(); + const sessionKeys = ["session-a", "session-b", "session-c", "session-d", "session-e"].map( + (name) => `agent:main:${name}`, + ); + const sessionA = expectDefined(sessionKeys[0], "session A key"); + const sessionB = expectDefined(sessionKeys[1], "session B key"); + const sessionC = expectDefined(sessionKeys[2], "session C key"); + const sessionD = expectDefined(sessionKeys[3], "session D key"); + const sessionE = expectDefined(sessionKeys[4], "session E key"); + const commentary = "Checking the workspace after returning."; + const runStartedAt = Date.now() - 10_000; + const pageResponse = (sessionKey: string, label: string) => ({ + deltaCursor: `cursor-${label.toLowerCase()}`, + messages: [{ role: "user", content: `${label} cached prompt`, timestamp: 1 }], + sessionId: `${label.toLowerCase()}-session`, + sessionInfo: { key: sessionKey, kind: "direct", updatedAt: 1 }, + }); + const historyCases = [ + { + match: { cursor: "cursor-b", sessionKey: sessionB }, + response: { + kind: "delta", + messages: [], + deltaCursor: "cursor-b-current", + sessionInfo: { + key: sessionB, + kind: "direct", + sessionId: "b-session", + updatedAt: 2, + hasActiveRun: true, + activeRunIds: ["run-b"], + status: "running", + }, + inFlightRun: { + runId: "run-b", + text: "", + startedAt: runStartedAt, + events: [ + { + runId: "run-b", + seq: 1, + stream: "item", + ts: runStartedAt, + sessionKey: sessionB, + data: { + kind: "preamble", + itemId: "preamble-restored", + progressText: commentary, + }, + }, + ], + }, + }, + }, + { match: { sessionKey: sessionA }, response: pageResponse(sessionA, "A") }, + { match: { sessionKey: sessionB }, response: pageResponse(sessionB, "B") }, + { match: { sessionKey: sessionC }, response: pageResponse(sessionC, "C") }, + { match: { sessionKey: sessionD }, response: pageResponse(sessionD, "D") }, + { match: { sessionKey: sessionE }, response: pageResponse(sessionE, "E") }, + ]; + const sessionRows = sessionKeys.map((key, index) => + key === sessionB + ? { + key, + kind: "direct" as const, + label: "Session B", + updatedAt: sessionKeys.length - index, + activeRunIds: ["run-b"], + hasActiveRun: true, + status: "running" as const, + } + : { + key, + kind: "direct" as const, + label: `Session ${String.fromCharCode(65 + index)}`, + updatedAt: sessionKeys.length - index, + }, + ); + const gateway = await installMockGateway(page, { + methodResponses: { + "chat.history": { cases: historyCases }, + "chat.startup": { cases: historyCases }, + "sessions.list": chatSessionListResponse(sessionRows), + }, + sessionKey: sessionA, + }); + + try { + await page.goto(`${suite.server.baseUrl}chat`); + const sessionLink = (sessionKey: string) => + page.locator( + `.sidebar-recent-session[data-session-key="${sessionKey}"] a.sidebar-recent-session__link`, + ); + await page.getByText("A cached prompt", { exact: true }).waitFor({ timeout: 10_000 }); + await sessionLink(sessionB).click(); + await page.getByText("B cached prompt", { exact: true }).waitFor({ timeout: 10_000 }); + await sessionLink(sessionA).click(); + await sessionLink(sessionC).click(); + await page.getByText("C cached prompt", { exact: true }).waitFor({ timeout: 10_000 }); + await sessionLink(sessionD).click(); + await page.getByText("D cached prompt", { exact: true }).waitFor({ timeout: 10_000 }); + await sessionLink(sessionE).click(); + await page.getByText("E cached prompt", { exact: true }).waitFor({ timeout: 10_000 }); + + const cursorRequests = async () => + [ + ...(await gateway.getRequests("chat.startup")), + ...(await gateway.getRequests("chat.history")), + ] + .map((request) => requireRecord(request.params)) + .filter((params) => params.sessionKey === sessionB && params.cursor === "cursor-b"); + const cursorRequestsBeforeReturn = (await cursorRequests()).length; + await sessionLink(sessionB).click(); + await expect + .poll(async () => (await cursorRequests()).length) + .toBeGreaterThan(cursorRequestsBeforeReturn); + const activeRunState = () => + page.locator('openclaw-chat-pane[aria-hidden="false"]').evaluate((element) => { + const state = ( + element as HTMLElement & { + state?: { + chatRunId?: string | null; + chatStreamSegments?: Array<{ text?: string }>; + }; + } + ).state; + return { + runId: state?.chatRunId ?? null, + segmentTexts: state?.chatStreamSegments?.map((segment) => segment.text) ?? [], + }; + }); + await expect.poll(activeRunState).toEqual({ + runId: "run-b", + segmentTexts: expect.arrayContaining([commentary]), + }); + await page + .locator('openclaw-chat-pane[aria-hidden="false"] .chat-thread p') + .getByText(commentary, { exact: true }) + .waitFor({ timeout: 10_000 }); + expect((await cursorRequests()).at(-1)).toMatchObject({ + cursor: "cursor-b", + sessionKey: sessionB, + }); + if (artifactDir) { + await page.screenshot({ + fullPage: true, + path: path.join(artifactDir, "cursor-active-commentary-return.png"), + }); + } + } finally { + await suite.closeBrowserContext(context); + } + }); +}); diff --git a/ui/src/pages/chat/chat-history-cursor.test.ts b/ui/src/pages/chat/chat-history-cursor.test.ts index 2179840257cb..05c0477a91be 100644 --- a/ui/src/pages/chat/chat-history-cursor.test.ts +++ b/ui/src/pages/chat/chat-history-cursor.test.ts @@ -45,6 +45,24 @@ function seedCachedHistory( return cache; } +async function loadHistoryWithBrowserTimers(state: ReturnType): Promise { + const globalWithWindow = globalThis as typeof globalThis & { + window?: Window & typeof globalThis; + }; + const previousWindow = globalWithWindow.window; + globalWithWindow.window = globalThis as unknown as Window & typeof globalThis; + try { + await loadChatHistory(state); + await vi.waitFor(() => expect(state.chatToolMessages).toHaveLength(1)); + } finally { + if (previousWindow) { + globalWithWindow.window = previousWindow; + } else { + Reflect.deleteProperty(globalWithWindow, "window"); + } + } +} + describe("chat history cursor revalidation", () => { it("keeps cached paint while replay updates an existing tool message in place", async () => { const cached = message( @@ -119,6 +137,72 @@ describe("chat history cursor revalidation", () => { ).toBe("cursor-2"); }); + it("restores active commentary and tools with an empty cached delta", async () => { + const cached = message("user", "cached", "cached-user", 1); + const handler = vi.fn(async (_params?: unknown) => ({ + kind: "delta", + messages: [], + deltaCursor: "cursor-2", + sessionInfo: { + key: "main", + kind: "direct", + sessionId: "session-cursor", + updatedAt: 2, + hasActiveRun: true, + activeRunIds: ["run-live"], + status: "running", + }, + inFlightRun: { + runId: "run-live", + text: "", + startedAt: 900, + events: [ + { + runId: "run-live", + seq: 1, + stream: "item", + ts: 900, + sessionKey: "main", + data: { + kind: "preamble", + itemId: "preamble-restored", + progressText: "Checking the workspace", + }, + }, + { + runId: "run-live", + seq: 2, + stream: "tool", + ts: 1_000, + sessionKey: "main", + data: { + toolCallId: "call-restored", + name: "read", + phase: "start", + args: { path: "README.md" }, + }, + }, + ], + }, + })); + const state = createState(handler); + seedCachedHistory(state, [cached], "cursor-1"); + + await loadHistoryWithBrowserTimers(state); + + expect(state.chatRunId).toBe("run-live"); + expect(state.chatStreamSegments).toContainEqual( + expect.objectContaining({ + itemId: "preamble-restored", + runId: "run-live", + text: "Checking the workspace", + }), + ); + expect(state.chatToolMessages).toContainEqual( + expect.objectContaining({ runId: "run-live", toolCallId: "call-restored" }), + ); + }); + it("clears a rejected cursor before falling back to a full tail fetch", async () => { const cached = message("user", "cached", "cached-user", 1); const fresh = message("assistant", "fresh", "fresh-assistant", 2); diff --git a/ui/src/pages/chat/chat-history.ts b/ui/src/pages/chat/chat-history.ts index 097089d7e439..c22de0c56aa2 100644 --- a/ui/src/pages/chat/chat-history.ts +++ b/ui/src/pages/chat/chat-history.ts @@ -313,6 +313,7 @@ type ChatHistoryDeltaResult = { messages: unknown[]; deltaCursor: string; sessionInfo: GatewaySessionRow; + inFlightRun?: ChatHistoryResult["inFlightRun"]; metadata?: ChatMetadataResult; }; @@ -385,6 +386,17 @@ function runProjectionsUnchanged( ); } +function readChatRunProjections(state: ChatState, sessionKey: string, agentId?: string) { + return getChatSessionProjection( + state, + state.chatMessages, + readChatSessionProjectionScope(state, { + sessionKey, + ...(agentId ? { agentId } : {}), + }), + ).runs; +} + function mergeInFlightAssistantTails( snapshotTail: string | null, cumulativeLiveTail: string | null, @@ -400,6 +412,83 @@ function mergeInFlightAssistantTails( return cumulativeLiveTail; } +function applyInFlightRunSnapshot(params: { + state: ChatState; + run: ChatHistoryResult["inFlightRun"]; + sessionInfo: GatewaySessionRow | undefined; + previousRunProjections: ReturnType["runs"]; + runProjectionsBeforeApply: ReturnType["runs"]; + currentRunProjections: ReturnType["runs"]; + resetStream: boolean; + activeStreamBeforeReset: string | null; +}): void { + const { + state, + run, + sessionInfo, + previousRunProjections, + runProjectionsBeforeApply, + currentRunProjections, + resetStream, + activeStreamBeforeReset, + } = params; + const inFlightRunId = run?.runId?.trim(); + if (!inFlightRunId || !run) { + return; + } + const projectedInFlightRun = currentRunProjections[inFlightRunId]; + const sameRunContinued = + state.chatRunId === inFlightRunId && + projectedInFlightRun?.status === "streaming" && + onlyInFlightRunProjectionChanged(previousRunProjections, currentRunProjections, inFlightRunId); + const activeRunIds = sessionInfo?.activeRunIds; + const inFlightRunIsActive = + isSessionRunActive(sessionInfo ?? {}) && + (!Array.isArray(activeRunIds) || activeRunIds.includes(inFlightRunId)) && + (!projectedInFlightRun || projectedInFlightRun.status === "streaming"); + const canAdoptInFlightRun = + inFlightRunIsActive && + ((resetStream && + !state.chatRunId && + runProjectionsUnchanged(previousRunProjections, runProjectionsBeforeApply)) || + sameRunContinued); + if (canAdoptInFlightRun) { + // Canonical run projections change on every live delta or terminal. + // Their identity fences ABA races where a run starts and finishes while + // history is pending; deltas from this same live run must still merge. + state.chatRunId = inFlightRunId; + } + if (!inFlightRunIsActive || state.chatRunId !== inFlightRunId) { + return; + } + const snapshotStartedAt = + typeof run.startedAt === "number" && Number.isFinite(run.startedAt) ? run.startedAt : null; + const snapshotTail = resolveInFlightAssistantTail(state.chatMessages, run.text, inFlightRunId); + const liveTail = sameRunContinued + ? resolveInFlightAssistantTail( + state.chatMessages, + extractText(projectedInFlightRun?.message), + inFlightRunId, + ) + : activeStreamBeforeReset; + const tail = mergeInFlightAssistantTails(snapshotTail, liveTail); + state.chatStream = tail; + state.chatStreamStartedAt = snapshotStartedAt ?? state.chatStreamStartedAt ?? Date.now(); + const persistedBoundary = latestPersistedSteerBoundary(state.chatMessages, inFlightRunId); + if (tail && persistedBoundary) { + markChatStreamAfterBoundary(state, { + runId: inFlightRunId, + boundaryRunId: persistedBoundary.runId, + timestamp: state.chatStreamStartedAt, + }); + } + state.chatRunStartup = { state: "activity", runId: inFlightRunId }; + // Disconnect cleanup intentionally removes transient activity rows while + // retaining the owned run. Replay fills that gap; per-identity sequence + // fences keep a delayed snapshot from replacing newer live progress. + replayInFlightRunEvents(state, run); +} + export function resolveChatHistoryPagination( result: ChatHistoryResult | undefined, ): ChatHistoryPagination { @@ -1531,14 +1620,7 @@ async function loadChatHistoryUncached( ); const startedAtMs = controlUiNowMs(); const previousMessages = state.chatMessages; - const previousRunProjections = getChatSessionProjection( - state, - previousMessages, - readChatSessionProjectionScope(state, { - sessionKey, - ...(requestAgentId ? { agentId: requestAgentId } : {}), - }), - ).runs; + const previousRunProjections = readChatRunProjections(state, sessionKey, requestAgentId); const previousPagination = state.chatHistoryPagination; const previousSessionId = state.currentSessionId ?? null; const previousDisplayedLeafEntryId = state.chatDisplayedLeafEntryId; @@ -1601,6 +1683,8 @@ async function loadChatHistoryUncached( } } if (isChatHistoryCursorResult(response) && response.kind === "delta") { + const runProjectionsBeforeApply = readChatRunProjections(state, sessionKey, requestAgentId); + const activeStreamBeforeApply = state.chatRunId ? state.chatStream : null; const runActive = isSessionRunActive(response.sessionInfo); for (const payload of response.messages) { applySessionMessagePayload(state, payload, runActive, { kind: "history-delta" }); @@ -1612,6 +1696,17 @@ async function loadChatHistoryUncached( state.chatThinkingLevel = response.sessionInfo.thinkingLevel ?? null; state.chatQueueModeOverride = response.sessionInfo.queueMode; state.chatEffectiveQueueMode = response.sessionInfo.effectiveQueueMode; + const currentRunProjections = readChatRunProjections(state, sessionKey, requestAgentId); + applyInFlightRunSnapshot({ + state, + run: response.inFlightRun, + sessionInfo: response.sessionInfo, + previousRunProjections, + runProjectionsBeforeApply, + currentRunProjections, + resetStream: !state.chatRunId || state.chatRunId === previousRunId, + activeStreamBeforeReset: activeStreamBeforeApply, + }); commitCurrentChatHistorySnapshot(state, response.deltaCursor ?? null); recordChatHistoryTiming(state, "applied", startedAtMs, { requestSessionKey: sessionKey, @@ -1625,6 +1720,7 @@ async function loadChatHistoryUncached( messages: state.chatMessages, deltaCursor: response.deltaCursor, sessionInfo: response.sessionInfo, + ...(response.inFlightRun ? { inFlightRun: response.inFlightRun } : {}), ...(response.metadata ? { metadata: response.metadata } : {}), sourceCanonicalListRevision: response.sourceCanonicalListRevision, }; @@ -1635,14 +1731,7 @@ async function loadChatHistoryUncached( const res = response; // Fence concurrent run lifecycle before applying the response. A remount // may replace the map itself, so compare its canonical run entries. - const runProjectionsBeforeApply = getChatSessionProjection( - state, - state.chatMessages, - readChatSessionProjectionScope(state, { - sessionKey, - ...(requestAgentId ? { agentId: requestAgentId } : {}), - }), - ).runs; + const runProjectionsBeforeApply = readChatRunProjections(state, sessionKey, requestAgentId); const messages = Array.isArray(res.messages) ? res.messages : []; const nextPagination = resolveChatHistoryPagination(res); const nextSessionId = resolveChatHistorySessionId(res); @@ -1794,73 +1883,16 @@ async function loadChatHistoryUncached( } } - const inFlightRunId = res.inFlightRun?.runId?.trim(); - const activeRunIds = res.sessionInfo?.activeRunIds; - const projectedInFlightRun = inFlightRunId ? historyProjection.runs[inFlightRunId] : undefined; - const sameRunContinued = Boolean( - inFlightRunId && - state.chatRunId === inFlightRunId && - projectedInFlightRun?.status === "streaming" && - onlyInFlightRunProjectionChanged( - previousRunProjections, - historyProjection.runs, - inFlightRunId, - ), - ); - const inFlightRunIsActive = Boolean( - inFlightRunId && - isSessionRunActive(res.sessionInfo ?? {}) && - (!Array.isArray(activeRunIds) || activeRunIds.includes(inFlightRunId)) && - (!projectedInFlightRun || projectedInFlightRun.status === "streaming"), - ); - const canAdoptInFlightRun = Boolean( - inFlightRunId && - inFlightRunIsActive && - ((resetStream && - !state.chatRunId && - runProjectionsUnchanged(previousRunProjections, runProjectionsBeforeApply)) || - sameRunContinued), - ); - if (inFlightRunId && canAdoptInFlightRun) { - // Canonical run projections change on every live delta or terminal. - // Their identity fences ABA races where a run starts and finishes while - // history is pending; deltas from this same live run must still merge. - state.chatRunId = inFlightRunId; - } - if (inFlightRunIsActive && res.inFlightRun && state.chatRunId === inFlightRunId) { - const snapshotStartedAt = - typeof res.inFlightRun.startedAt === "number" && Number.isFinite(res.inFlightRun.startedAt) - ? res.inFlightRun.startedAt - : null; - const snapshotTail = resolveInFlightAssistantTail( - state.chatMessages, - res.inFlightRun?.text, - inFlightRunId, - ); - const liveTail = sameRunContinued - ? resolveInFlightAssistantTail( - state.chatMessages, - extractText(projectedInFlightRun?.message), - inFlightRunId, - ) - : activeStreamBeforeReset; - const tail = mergeInFlightAssistantTails(snapshotTail, liveTail); - state.chatStream = tail; - state.chatStreamStartedAt = snapshotStartedAt ?? state.chatStreamStartedAt ?? Date.now(); - const persistedBoundary = latestPersistedSteerBoundary(state.chatMessages, inFlightRunId); - if (tail && persistedBoundary) { - markChatStreamAfterBoundary(state, { - runId: inFlightRunId, - boundaryRunId: persistedBoundary.runId, - timestamp: state.chatStreamStartedAt, - }); - } - state.chatRunStartup = { state: "activity", runId: inFlightRunId }; - // Disconnect cleanup intentionally removes transient activity rows while - // retaining the owned run. Replay fills that gap; per-identity sequence - // fences keep a delayed snapshot from replacing newer live progress. - replayInFlightRunEvents(state, res.inFlightRun); - } + applyInFlightRunSnapshot({ + state, + run: res.inFlightRun, + sessionInfo: res.sessionInfo, + previousRunProjections, + runProjectionsBeforeApply, + currentRunProjections: historyProjection.runs, + resetStream, + activeStreamBeforeReset, + }); recordChatHistoryTiming(state, "applied", startedAtMs, { requestSessionKey: sessionKey, diff --git a/ui/src/pages/chat/session-prefetch.test.ts b/ui/src/pages/chat/session-prefetch.test.ts index dcd278a96540..338fcf44029c 100644 --- a/ui/src/pages/chat/session-prefetch.test.ts +++ b/ui/src/pages/chat/session-prefetch.test.ts @@ -399,6 +399,75 @@ describe("recent session prefetch", () => { }); }); + it("retains the prior cursor when a delta carries transient active-run replay", async () => { + const sessionKey = "agent:main:active"; + cacheChatSessionSnapshot( + cache, + snapshotHost, + { sessionKey }, + { + deltaCursor: "cursor-1", + messages: [{ role: "user", content: "cached" }], + pagination: { hasMore: false, completeSnapshot: true }, + sessionId: "session-active", + }, + ); + const request = vi.fn(async () => ({ + kind: "delta", + messages: [], + deltaCursor: "cursor-2", + sessionInfo: { + key: sessionKey, + kind: "direct", + sessionId: "session-active", + updatedAt: 2, + hasActiveRun: true, + }, + inFlightRun: { + runId: "run-active", + events: [ + { + runId: "run-active", + seq: 1, + stream: "item", + ts: 1, + sessionKey, + data: { kind: "preamble", itemId: "progress", progressText: "Still working" }, + }, + ], + }, + })); + + updatePrefetch({ + client: { request } as unknown as GatewayBrowserClient, + listRevision: 1, + openSessionKeys: [], + rows: [row(sessionKey, NOW + 1)], + }); + await vi.advanceTimersByTimeAsync(2_000); + await settlePromises(); + + expect(readChatSessionSnapshot(cache, snapshotHost, { sessionKey })?.deltaCursor).toBe( + "cursor-1", + ); + }); + + it("leaves active sessions for the presented pane to revalidate", async () => { + const sessionKey = "agent:main:active"; + const request = vi.fn(); + + updatePrefetch({ + client: { request } as unknown as GatewayBrowserClient, + listRevision: 1, + openSessionKeys: [], + rows: [{ ...row(sessionKey, NOW + 1), hasActiveRun: true, status: "running" }], + }); + await vi.advanceTimersByTimeAsync(2_000); + await settlePromises(); + + expect(request).not.toHaveBeenCalled(); + }); + it("skips the cycle when another tab holds the Web Lock", async () => { const request = vi.fn(); const locksRequest = vi.fn( diff --git a/ui/src/pages/chat/session-prefetch.ts b/ui/src/pages/chat/session-prefetch.ts index 5758bfce3233..ad0a142238b2 100644 --- a/ui/src/pages/chat/session-prefetch.ts +++ b/ui/src/pages/chat/session-prefetch.ts @@ -3,6 +3,7 @@ import type { ReactiveController, ReactiveControllerHost } from "lit"; import type { GatewayBrowserClient } from "../../api/gateway.ts"; import type { GatewaySessionRow } from "../../api/types.ts"; import type { ApplicationContext } from "../../app/context.ts"; +import { isSessionRunActive } from "../../lib/session-run-state.ts"; import { requestChatSessionSnapshot } from "./chat-history.ts"; import { appendChatMessageToCache, @@ -264,7 +265,9 @@ class SessionPrefetcher { } cached = { ...updated, - deltaCursor: result.deltaCursor, + // Prefetch does not own transient run replay. Keep the prior cursor + // so the opening pane can consume the same authoritative snapshot. + ...(result.inFlightRun ? {} : { deltaCursor: result.deltaCursor }), ...(Object.hasOwn(result.sessionInfo, "activeLeafEntryId") ? { displayedLeafEntryId: result.sessionInfo.activeLeafEntryId?.trim() || null } : {}), @@ -304,6 +307,11 @@ class SessionPrefetcher { const seen = new Set(); let deferMs: number | null = null; for (const row of rows) { + // The presented pane owns transient run adoption and replay. Background + // prefetch only warms durable history, so it must not consume active state. + if (isSessionRunActive(row)) { + continue; + } const snapshotKey = resolveChatSnapshotKey(snapshot.snapshotHost, { sessionKey: row.key, agentId: row.agentId,