diff --git a/packages/gateway-client/src/session-projection.test.ts b/packages/gateway-client/src/session-projection.test.ts index 3b4478d19b73..f5f0116e3968 100644 --- a/packages/gateway-client/src/session-projection.test.ts +++ b/packages/gateway-client/src/session-projection.test.ts @@ -188,6 +188,21 @@ describe("session transcript projection", () => { ]); }); + it("adopts a durable assistant identity from the same live run", () => { + const synthetic = createMessage("assistant", "streamed final"); + const persisted = createMessage("assistant", "persisted final", { + id: "assistant-final", + seq: 2, + }); + let state = projectLiveSessionMessage(createSessionProjection(primaryScope), synthetic, { + runId: "final-run", + }); + + state = projectLiveSessionMessage(state, persisted, { runId: "final-run" }); + + expect(state.messages).toEqual([persisted]); + }); + it("does not adopt an ambiguous synthetic final across distinct same-run assistants", () => { const synthetic = createMessage("assistant", "delta-only final", { idempotencyKey: "final-run", diff --git a/packages/gateway-client/src/session-projection.ts b/packages/gateway-client/src/session-projection.ts index 5d8c905d033f..8b228aa82f51 100644 --- a/packages/gateway-client/src/session-projection.ts +++ b/packages/gateway-client/src/session-projection.ts @@ -300,6 +300,21 @@ function entryMatches( if (sameTranscriptIdentity(left.identity, right.identity)) { return true; } + const durableEntry = left.identity?.id ? left : right.identity?.id ? right : null; + const provisionalEntry = durableEntry === left ? right : durableEntry === right ? left : null; + if ( + durableEntry?.live && + provisionalEntry?.live && + durableEntry.identity?.role === "assistant" && + provisionalEntry.identity?.role === "assistant" && + !durableEntry.identity.isImported && + !provisionalEntry.identity.isImported && + !provisionalEntry.identity.id && + durableEntry.identity.runId && + durableEntry.identity.runId === provisionalEntry.identity.runId + ) { + return true; + } const persisted = left.identity; const observed = right.identity; if ( diff --git a/src/tui/tui-event-handlers.test.ts b/src/tui/tui-event-handlers.test.ts index 104862523799..9fc5b2cd705f 100644 --- a/src/tui/tui-event-handlers.test.ts +++ b/src/tui/tui-event-handlers.test.ts @@ -3176,72 +3176,6 @@ describe("tui-event-handlers: handleAgentEvent", () => { await vi.waitFor(() => expect(loadHistory).toHaveBeenCalledTimes(2)); }); - it("waits for terminal persistence before refreshing a displayed local final", () => { - const { - state, - chatLog, - loadHistory, - handleChatEvent, - handleSessionsChangedEvent, - handleSessionMessageEvent, - } = createHandlersHarness({ state: { activeChatRunId: "run-active" } }); - - handleSessionMessageEvent(makeSessionMessageEvent(state)); - handleChatEvent({ - runId: "run-active", - state: "final", - message: { content: [{ type: "text", text: "keep this visible" }] }, - }); - - expect(chatLog.finalizeAssistant).toHaveBeenCalledWith("keep this visible", "run-active"); - expect(loadHistory).not.toHaveBeenCalled(); - - handleSessionMessageEvent(makeSessionMessageEvent(state, { updatedAt: 200 })); - expect(loadHistory).not.toHaveBeenCalled(); - - handleSessionsChangedEvent({ - runId: "run-active", - phase: "end", - }); - - expect(loadHistory).toHaveBeenCalledTimes(1); - }); - - it.each(["end", "error"] as const)( - "releases a displayed client run when its internal agent reports %s", - (phase) => { - const { - state, - chatLog, - loadHistory, - handleChatEvent, - handleSessionsChangedEvent, - handleSessionMessageEvent, - } = createHandlersHarness({ state: { activeChatRunId: "run-client-visible" } }); - - handleSessionMessageEvent(makeSessionMessageEvent(state)); - handleChatEvent({ - runId: "run-client-visible", - state: "final", - message: { content: [{ type: "text", text: "keep the aliased reply visible" }] }, - }); - - expect(chatLog.finalizeAssistant).toHaveBeenCalledWith( - "keep the aliased reply visible", - "run-client-visible", - ); - expect(loadHistory).not.toHaveBeenCalled(); - - handleSessionsChangedEvent({ - runId: "run-internal-agent", - clientRunId: "run-client-visible", - phase, - }); - - expect(loadHistory).toHaveBeenCalledTimes(1); - }, - ); - it("refreshes after a local final when terminal persistence arrives first", () => { const { state, @@ -3269,115 +3203,6 @@ describe("tui-event-handlers: handleAgentEvent", () => { expect(loadHistory).toHaveBeenCalledTimes(1); }); - it("defers the first external update after a visible final until persistence", () => { - const { - state, - chatLog, - loadHistory, - handleChatEvent, - handleSessionsChangedEvent, - handleSessionMessageEvent, - } = createHandlersHarness({ state: { activeChatRunId: "run-active" } }); - - handleChatEvent({ - runId: "run-active", - state: "final", - message: { content: [{ type: "text", text: "keep this visible" }] }, - }); - - expect(chatLog.finalizeAssistant).toHaveBeenCalledWith("keep this visible", "run-active"); - expect(loadHistory).not.toHaveBeenCalled(); - - handleSessionMessageEvent(makeSessionMessageEvent(state)); - - expect(loadHistory).not.toHaveBeenCalled(); - - handleSessionsChangedEvent({ - runId: "run-active", - phase: "end", - }); - - expect(loadHistory).toHaveBeenCalledTimes(1); - }); - - it.each([ - { firstPersistedRunId: "run-first", secondPersistedRunId: "run-second" }, - { firstPersistedRunId: "run-second", secondPersistedRunId: "run-first" }, - ])( - "waits for both visible finals when $firstPersistedRunId persists first", - ({ firstPersistedRunId, secondPersistedRunId }) => { - const { - state, - chatLog, - loadHistory, - handleChatEvent, - handleSessionsChangedEvent, - handleSessionMessageEvent, - } = createHandlersHarness({ state: { activeChatRunId: null } }); - - for (const runId of ["run-first", "run-second"]) { - handleChatEvent({ - runId, - state: "final", - message: { content: [{ type: "text", text: `${runId} visible response` }] }, - }); - } - - expect(chatLog.finalizeAssistant).toHaveBeenCalledTimes(2); - handleSessionMessageEvent(makeSessionMessageEvent(state, { updatedAt: 200 })); - expect(loadHistory).not.toHaveBeenCalled(); - - handleSessionsChangedEvent({ - runId: firstPersistedRunId, - phase: "end", - }); - expect(loadHistory).not.toHaveBeenCalled(); - - handleSessionsChangedEvent({ - runId: secondPersistedRunId, - phase: "end", - }); - expect(loadHistory).toHaveBeenCalledTimes(1); - }, - ); - - it("coalesces external updates until every concurrently displayed final is persisted", () => { - const { - loadHistory, - handleChatEvent, - handleSessionsChangedEvent, - handleSessionMessageEvent, - } = createHandlersHarness({ state: { activeChatRunId: null } }); - - for (const runId of ["run-first", "run-second", "run-third"]) { - handleChatEvent({ - runId, - state: "final", - message: { content: [{ type: "text", text: `${runId} visible response` }] }, - }); - } - - for (let index = 0; index < 250; index += 1) { - handleSessionMessageEvent({ - updatedAt: index, - }); - } - - for (const runId of ["run-third", "run-first"]) { - handleSessionsChangedEvent({ - runId, - phase: "end", - }); - expect(loadHistory).not.toHaveBeenCalled(); - } - - handleSessionsChangedEvent({ - runId: "run-second", - phase: "end", - }); - expect(loadHistory).toHaveBeenCalledTimes(1); - }); - it("does not reload until an optimistic submit is resolved", () => { const { state, loadHistory, handleSessionMessageEvent, flushPendingHistoryRefreshIfIdle } = createHandlersHarness({ @@ -3392,47 +3217,6 @@ describe("tui-event-handlers: handleAgentEvent", () => { expect(loadHistory).toHaveBeenCalledTimes(1); }); - - it("waits for persistence when a refresh arrives before an optimistic run is accepted", () => { - const { - state, - chatLog, - loadHistory, - handleAgentEvent, - handleChatEvent, - handleSessionsChangedEvent, - handleSessionMessageEvent, - } = createHandlersHarness({ - state: { activeChatRunId: null, pendingSubmit: sendingSubmit("run-pending") }, - }); - - handleSessionMessageEvent(makeSessionMessageEvent(state)); - expect(loadHistory).not.toHaveBeenCalled(); - - state.pendingSubmit = acceptedSubmit("run-pending"); - handleAgentEvent({ - runId: "run-pending", - sessionKey: state.currentSessionKey, - }); - handleChatEvent({ - runId: "run-pending", - state: "final", - message: { content: [{ type: "text", text: "keep accepted output visible" }] }, - }); - - expect(chatLog.finalizeAssistant).toHaveBeenCalledWith( - "keep accepted output visible", - "run-pending", - ); - expect(loadHistory).not.toHaveBeenCalled(); - - handleSessionsChangedEvent({ - runId: "run-pending", - phase: "end", - }); - - expect(loadHistory).toHaveBeenCalledTimes(1); - }); }); describe("sessions.changed history reload", () => { diff --git a/src/tui/tui-event-handlers.ts b/src/tui/tui-event-handlers.ts index 80d1312d23fa..b2141bfc6dcf 100644 --- a/src/tui/tui-event-handlers.ts +++ b/src/tui/tui-event-handlers.ts @@ -17,7 +17,7 @@ import { getTuiSessionProjection, hasDisplayableTuiSessionFinal, isIdentityOnlyTuiSessionInvalidation, - isReplayableTuiSessionMessage, + projectTuiSessionMessage, projectTuiSessionFinal, readTuiSessionProjectionScope, reduceTuiSessionProjection, @@ -324,10 +324,8 @@ export function createEventHandlers(context: EventHandlerContext) { if (!suppressEmptyExternalPlaceholder) { projectTuiSessionFinal(state, evt, finalText, hasStreamedText); } - // Skip the history reload when the final event produced displayable - // output. loadHistory() does clearAll() + rebuild from server data, - // but the server may not have persisted this message yet — causing - // the just-rendered final message to vanish (#87922). + // Skip history reload for displayable output: loadHistory() rebuilds from + // server data that may not contain the final yet, making it vanish (#87922). maybeRefreshHistoryForRun(evt.runId, { hasDisplayableFinal: !suppressEmptyExternalPlaceholder, wasPendingChatRun: isPendingChatRun, @@ -509,13 +507,12 @@ export function createEventHandlers(context: EventHandlerContext) { return; } - if (isReplayableTuiSessionMessage(evt)) { - reduceTuiSessionProjection(state, { - type: "messagePersisted", - message: evt.message, - envelope: evt, - scope: readTuiSessionProjectionScope(state), - }); + const unboundDisplayedRunIds = [...finalizedRunsWithDisplay.keys()].filter( + (runId) => !persistedTerminalRunIds.has(runId), + ); + const authoritativeRunId = projectTuiSessionMessage(state, evt, unboundDisplayedRunIds); + if (authoritativeRunId) { + runCoordinator.notePersistedRun(authoritativeRunId); } const liveUserMessage = readTuiSessionUserMessage(evt); if (liveUserMessage) { @@ -541,11 +538,9 @@ export function createEventHandlers(context: EventHandlerContext) { } } - if (runCoordinator.deferSessionMessageRefresh()) { + if (runCoordinator.routeSessionMessageRefresh(Boolean(liveUserMessage || authoritativeRunId))) { void refreshSessionInfo?.(); - return; } - flushPendingHistoryRefreshIfIdle(); }; const handleAgentEvent = (payload: unknown) => { diff --git a/src/tui/tui-pty-harness-fixture-test-support.ts b/src/tui/tui-pty-harness-fixture-test-support.ts index df3d5e260273..9d4d898ad61c 100644 --- a/src/tui/tui-pty-harness-fixture-test-support.ts +++ b/src/tui/tui-pty-harness-fixture-test-support.ts @@ -1,15 +1,60 @@ // Keeps fake-terminal test-only logs and opaque-session fixtures independently bounded. -import { writeFile } from "node:fs/promises"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; import path from "node:path"; import { pathToFileURL } from "node:url"; import { TUI_PTY_ASSISTANT_FIXTURE_SCRIPT } from "./tui-pty-assistant-fixture-test-support.js"; import { TUI_PTY_GAP_HISTORY_FIXTURE_SCRIPT } from "./tui-pty-gap-fixture-test-support.js"; +import { + waitForFixtureLogEntry, + type FixtureLogEntry, +} from "./tui-pty-harness-assertion-test-support.js"; import { TUI_PTY_RENDERING_FIXTURE_SCRIPT } from "./tui-pty-rendering-test-support.js"; import { TUI_PTY_RESET_FIXTURE } from "./tui-pty-reset-fixture-test-support.js"; import { TUI_PTY_SESSION_SUBSCRIPTION_FIXTURE_SCRIPT } from "./tui-pty-subscription-fixture-test-support.js"; +import { startPty, type PtyRun } from "./tui-pty-test-support.js"; export * from "./tui-pty-harness-assertion-test-support.js"; +const activeRuns: PtyRun[] = []; +const OUTPUT_TIMEOUT_MS = 2_000; +const EXIT_TIMEOUT_MS = 4_000; + +export async function disposeActiveTuiFixtures(): Promise { + for (const run of activeRuns.splice(0)) { + await run.dispose(); + } +} + +export async function startTuiFixture(opts: { env?: NodeJS.ProcessEnv } = {}) { + const tempDir = await mkdtemp(path.join(tmpdir(), "openclaw-tui-pty-")); + const scriptPath = await writeTuiPtyFixtureScript(tempDir); + const logPath = path.join(tempDir, "fixture-log.jsonl"); + const run = startPty(process.execPath, ["--import", "tsx", scriptPath], { + activeRuns, + cwd: process.cwd(), + env: { + OPENCLAW_THEME: "dark", + OPENCLAW_TUI_PTY_LOG_PATH: logPath, + NO_COLOR: undefined, + ...opts.env, + }, + exitTimeoutMs: EXIT_TIMEOUT_MS, + outputTimeoutMs: OUTPUT_TIMEOUT_MS, + }); + + return { + run, + logPath, + waitForLogEntry: async (predicate: (entry: FixtureLogEntry) => boolean, timeoutMs?: number) => + await waitForFixtureLogEntry(logPath, predicate, timeoutMs ?? OUTPUT_TIMEOUT_MS, run.output), + cleanup: async () => { + await run.dispose(); + await rm(tempDir, { recursive: true, force: true }); + }, + }; +} + export async function writeTuiPtyFixtureScript(dir: string) { // Temp files sit outside the repo package scope; .mts preserves the ESM contract under tsx. const scriptPath = path.join(dir, "run-tui-pty-fixture.mts"); @@ -47,6 +92,8 @@ export async function writeTuiPtyFixtureScript(dir: string) { const dynamicCommandDescription = process.env.OPENCLAW_TUI_PTY_DYNAMIC_COMMAND_DESCRIPTION; const thinkingLabel = process.env.OPENCLAW_TUI_PTY_THINKING_LABEL; const safeThinkingLabel = process.env.OPENCLAW_TUI_PTY_SAFE_THINKING_LABEL; + const liveReplyHistory: unknown[] = []; + let liveReplySequence = 0; const thinkingLevels = [ ...(thinkingLabel ? [{ id: "fixture-thinking", label: thinkingLabel }] : []), ...(safeThinkingLabel ? [{ id: "fixture-thinking-safe", label: safeThinkingLabel }] : []), @@ -155,6 +202,71 @@ export async function writeTuiPtyFixtureScript(dir: string) { thinking: opts.thinking, }); const runId = opts.runId ?? "run-pty-fixture"; + if (opts.message.startsWith("live reply dedupe proof: ")) { + const reply = opts.message.endsWith("first") ? "TUI_LIVE_FIRST" : "TUI_LIVE_SECOND"; + const userSequence = ++liveReplySequence; + const assistantSequence = ++liveReplySequence; + const userMessage = { + role: "user", + content: [{ type: "text", text: opts.message }], + __openclaw: { + id: "live-user-" + userSequence, + idempotencyKey: runId + ":user", + seq: userSequence, + }, + }; + const assistantMessage = { + role: "assistant", + content: [{ type: "text", text: reply }], + __openclaw: { id: "live-assistant-" + assistantSequence, seq: assistantSequence }, + }; + liveReplyHistory.push(userMessage, assistantMessage); + queueMicrotask(() => { + this.onEvent?.({ + event: "session.message", + payload: { + sessionKey: opts.sessionKey, + message: userMessage, + messageId: userMessage.__openclaw.id, + messageSeq: userSequence, + }, + }); + this.onEvent?.({ + event: "chat", + payload: { + runId, + sessionKey: opts.sessionKey, + seq: assistantSequence, + state: "delta", + message: { role: "assistant", content: [{ type: "text", text: reply }] }, + }, + }); + this.onEvent?.({ + event: "chat", + payload: { + runId, + sessionKey: opts.sessionKey, + seq: assistantSequence + 1, + state: "final", + message: { role: "assistant", content: [{ type: "text", text: reply }] }, + }, + }); + this.onEvent?.({ + event: "session.message", + payload: { + sessionKey: opts.sessionKey, + message: assistantMessage, + messageId: assistantMessage.__openclaw.id, + messageSeq: assistantSequence, + }, + }); + this.onEvent?.({ + event: "sessions.changed", + payload: { sessionKey: opts.sessionKey, runId, phase: "end" }, + }); + }); + return { runId }; + } if (opts.message === "tui error redaction proof") { const escape = String.fromCharCode(27); throw new Error("gateway down", { @@ -342,6 +454,9 @@ export async function writeTuiPtyFixtureScript(dir: string) { record("loadHistory", { sessionKey }); const gapHistory = loadGapHistory(sessionKey); if (gapHistory) { return gapHistory; } + if (liveReplyHistory.length > 0) { + return { messages: [...liveReplyHistory] }; + } const rapidSwitchMarker = sessionKey.endsWith("switch-a") ? "A" : sessionKey.endsWith("switch-b") diff --git a/src/tui/tui-pty-harness.e2e.test.ts b/src/tui/tui-pty-harness.e2e.test.ts index 1609439d53e5..fb340add7cf3 100644 --- a/src/tui/tui-pty-harness.e2e.test.ts +++ b/src/tui/tui-pty-harness.e2e.test.ts @@ -1,20 +1,18 @@ // Exercises the fake-backend TUI PTY harness and visible terminal output. -import { mkdtemp, rm } from "node:fs/promises"; -import { tmpdir } from "node:os"; -import path from "node:path"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { sleep } from "../utils/sleep.js"; import { exerciseTuiCommandSurface } from "./tui-pty-command-surfaces-test-support.js"; import { approveWorkspaceSkill, COMPACT_TERMINAL_SIZES, + disposeActiveTuiFixtures, exerciseFragmentedUnicodePrompt, exerciseNarrowTerminalRendering, exerciseTerminalOutputSafety, objectFieldEquals, readFixtureLog, - waitForFixtureLogEntry, - writeTuiPtyFixtureScript, + startTuiFixture, + waitForSynchronizedFrameRows, type FixtureLogEntry, } from "./tui-pty-harness-fixture-test-support.js"; import { @@ -23,44 +21,10 @@ import { streamingPrefixFrame, toolFrame, } from "./tui-pty-rendering-test-support.js"; -import { startPty, type PtyRun } from "./tui-pty-test-support.js"; - -const activeRuns: PtyRun[] = []; const STARTUP_TIMEOUT_MS = 20_000; -const OUTPUT_TIMEOUT_MS = 2_000; -const EXIT_TIMEOUT_MS = 4_000; const TEST_TIMEOUT_MS = 5_000; const STARTUP_TEST_TIMEOUT_MS = 25_000; -async function startTuiFixture(opts: { env?: NodeJS.ProcessEnv } = {}) { - const tempDir = await mkdtemp(path.join(tmpdir(), "openclaw-tui-pty-")); - const scriptPath = await writeTuiPtyFixtureScript(tempDir); - const logPath = path.join(tempDir, "fixture-log.jsonl"); - const run = startPty(process.execPath, ["--import", "tsx", scriptPath], { - activeRuns, - cwd: process.cwd(), - env: { - OPENCLAW_THEME: "dark", - OPENCLAW_TUI_PTY_LOG_PATH: logPath, - NO_COLOR: undefined, - ...opts.env, - }, - exitTimeoutMs: EXIT_TIMEOUT_MS, - outputTimeoutMs: OUTPUT_TIMEOUT_MS, - }); - - return { - run, - logPath, - waitForLogEntry: async (predicate: (entry: FixtureLogEntry) => boolean, timeoutMs?: number) => - await waitForFixtureLogEntry(logPath, predicate, timeoutMs ?? OUTPUT_TIMEOUT_MS, run.output), - cleanup: async () => { - await run.dispose(); - await rm(tempDir, { recursive: true, force: true }); - }, - }; -} - it("rejects rendering oracle false positives", () => { const tokens = Array.from({ length: 64 }, (_, i) => `T${String(i).padStart(3, "0")}`); const promptFrame = [`burst streaming proof ${tokens.join(" ")}`, "local ready | idle"]; @@ -122,9 +86,7 @@ describe.sequential("TUI PTY harness", () => { }, STARTUP_TEST_TIMEOUT_MS); afterAll(async () => { - for (const run of activeRuns.splice(0)) { - await run.dispose(); - } + await disposeActiveTuiFixtures(); for (const started of [ fixture, compactFooterFixture, @@ -362,6 +324,33 @@ describe.sequential("TUI PTY harness", () => { }, STARTUP_TEST_TIMEOUT_MS, ); + + it( + "renders each live assistant reply once without replaying stale history", + async () => { + const liveFixture = await startTuiFixture({ + env: { OPENCLAW_TUI_PTY_COLS: "220", OPENCLAW_TUI_PTY_ROWS: "50" }, + }); + try { + await liveFixture.run.waitForOutput("local ready", STARTUP_TIMEOUT_MS); + await liveFixture.run.write("live reply dedupe proof: first\r", { delay: false }); + await liveFixture.run.waitForOutput("TUI_LIVE_FIRST"); + await liveFixture.run.write("live reply dedupe proof: second\r", { delay: false }); + const rows = await waitForSynchronizedFrameRows( + liveFixture.run, + (frame) => frame.some((row) => row.includes("TUI_LIVE_SECOND")), + STARTUP_TIMEOUT_MS, + ); + const assistantRows = rows.filter( + (row) => row.includes("TUI_LIVE_FIRST") || row.includes("TUI_LIVE_SECOND"), + ); + expect(assistantRows).toEqual(["TUI_LIVE_FIRST", "TUI_LIVE_SECOND"]); + } finally { + await liveFixture.cleanup(); + } + }, + STARTUP_TEST_TIMEOUT_MS, + ); // prettier-ignore const editorInputCases = [ ["recalls submitted input history through literal terminal navigation", [["w", "history recall proof\r"], ["s", "history recall proof"], ["o", "PTY_RESPONSE: history recall proof"], ["w", "\u001b[A\u0005 edited\r"], ["s", "history recall proof edited"]]], diff --git a/src/tui/tui-run-lifecycle.ts b/src/tui/tui-run-lifecycle.ts index 640787f16b9e..42e5a523ce9e 100644 --- a/src/tui/tui-run-lifecycle.ts +++ b/src/tui/tui-run-lifecycle.ts @@ -70,18 +70,13 @@ export function createTuiRunLifecycle(context: TuiRunLifecycleContext) { let streamingWatchdogRunId: string | null = null; const flushPendingHistoryRefreshIfIdle = () => { - if ( - state.activeChatRunId || - hasPendingSubmit(state) || - runCoordinator.isSessionMessagePersistencePending - ) { + if (state.activeChatRunId || hasPendingSubmit(state)) { return; } - if (!runCoordinator.pendingHistoryRefresh && !runCoordinator.hasPendingSessionMessageRefresh) { + if (!runCoordinator.pendingHistoryRefresh) { return; } runCoordinator.pendingHistoryRefresh = false; - runCoordinator.consumeSessionMessageRefresh(); runCoordinator.queueHistoryReload(); }; @@ -201,7 +196,6 @@ export function createTuiRunLifecycle(context: TuiRunLifecycleContext) { }; const markSubmittedRunRegistered = (runId: string) => { - runCoordinator.bindRegisteredPendingRun(runId); clearPendingSubmitDraft(state, runId); }; diff --git a/src/tui/tui-session-projection.ts b/src/tui/tui-session-projection.ts index a343637a849f..fc910d8c7c4d 100644 --- a/src/tui/tui-session-projection.ts +++ b/src/tui/tui-session-projection.ts @@ -51,7 +51,7 @@ export function isIdentityOnlyTuiSessionInvalidation(event: SessionChangedEvent) } /** Provider-local imports require a complete source or persisted—not envelope—sequence. */ -export function isReplayableTuiSessionMessage(event: SessionMessageEvent): boolean { +function isReplayableTuiSessionMessage(event: SessionMessageEvent): boolean { const identity = readSessionMessageIdentity(event.message, event); return Boolean( identity && @@ -110,6 +110,35 @@ export function reduceTuiSessionProjection( return projection; } +/** Promote a durable assistant row into the matching live run projection. */ +export function projectTuiSessionMessage( + state: TuiStateAccess, + event: SessionMessageEvent, + unboundDisplayedRunIds: readonly string[], +): string | undefined { + if (!isReplayableTuiSessionMessage(event)) { + return undefined; + } + const identity = readSessionMessageIdentity(event.message, event); + const assistantMessageId = + identity?.role === "assistant" && !identity.isImported ? (identity.id ?? undefined) : undefined; + // Some transcript envelopes omit run identity. A single unbound displayed + // final is the only non-ambiguous live owner that can adopt its durable ID. + const authoritativeRunId = assistantMessageId + ? (identity?.runId ?? + event.clientRunId ?? + state.activeChatRunId ?? + (unboundDisplayedRunIds.length === 1 ? unboundDisplayedRunIds[0] : undefined)) + : undefined; + reduceTuiSessionProjection(state, { + type: "messagePersisted", + message: event.message, + envelope: authoritativeRunId ? { ...event, runId: authoritativeRunId } : event, + scope: readTuiSessionProjectionScope(state), + }); + return authoritativeRunId; +} + /** Retain the assistant reply actually rendered until authoritative history adopts it. */ export function projectTuiSessionFinal( state: TuiStateAccess, diff --git a/src/tui/tui-session-run-coordinator.test.ts b/src/tui/tui-session-run-coordinator.test.ts index 90c1bc56ac9b..e607afaa2ff0 100644 --- a/src/tui/tui-session-run-coordinator.test.ts +++ b/src/tui/tui-session-run-coordinator.test.ts @@ -181,28 +181,6 @@ describe("TuiSessionRunCoordinator", () => { expect(coordinator.resolveMostRecentPromotableRun()).toBe("run-pending"); }); - it("keeps every displayed final behind its own persistence barrier", () => { - const { coordinator } = createCoordinator(); - coordinator.noteFinalizedRun("run-first", { displayedFinal: true }); - coordinator.noteFinalizedRun("run-second", { displayedFinal: true }); - - expect(coordinator.deferSessionMessageRefresh()).toBe(true); - coordinator.notePersistedRun("run-second"); - expect(coordinator.isSessionMessagePersistencePending).toBe(true); - coordinator.notePersistedRun("run-first"); - expect(coordinator.isSessionMessagePersistencePending).toBe(false); - expect(coordinator.hasPendingSessionMessageRefresh).toBe(true); - }); - - it("recognizes persistence that arrives before a visible final", () => { - const { coordinator } = createCoordinator(); - coordinator.notePersistedRun("run-first"); - coordinator.noteFinalizedRun("run-first", { displayedFinal: true }); - - expect(coordinator.deferSessionMessageRefresh()).toBe(false); - expect(coordinator.isSessionMessagePersistencePending).toBe(false); - }); - it("serializes queued reloads and replays a gated terminal event", async () => { let resolveHistory: ((result: TuiHistoryLoadResult) => void) | undefined; const { coordinator, loadHistory, finalizeHistoryOwnedRun, replayHistoryRunEvent } = diff --git a/src/tui/tui-session-run-coordinator.ts b/src/tui/tui-session-run-coordinator.ts index 83763ae367a2..78fea47f5bd7 100644 --- a/src/tui/tui-session-run-coordinator.ts +++ b/src/tui/tui-session-run-coordinator.ts @@ -65,14 +65,12 @@ export class TuiSessionRunCoordinator { pendingHistoryRefresh = false; private readonly historyReloadRuns = new Map(); - private readonly sessionMessagePersistenceRunIds = new Set(); private readonly confirmedStreamRunIds = new Set(); private readonly retiredOrphanRunIds = new Map(); private rejectUnconfirmedRuns = false; private historyReloadInFlight = false; private historyReloadQueued = false; private historyReloadGeneration = 0; - private sessionMessageRefreshPending = false; constructor(private readonly context: TuiSessionRunCoordinatorContext) { this.streamAssembler = new TuiStreamAssembler((runId) => { @@ -205,9 +203,6 @@ export class TuiSessionRunCoordinator { this.noteCompletedRun(runId); if (options?.displayedFinal) { this.finalizedRunsWithDisplay.set(runId, Date.now()); - if (this.sessionMessageRefreshPending && !this.persistedTerminalRunIds.has(runId)) { - this.sessionMessagePersistenceRunIds.add(runId); - } } this.dropSessionRun(runId); this.pruneRunMap(this.finalizedRuns); @@ -218,11 +213,6 @@ export class TuiSessionRunCoordinator { this.liveTerminalErrorMessages.delete(retainedRunId); } } - for (const retainedRunId of this.sessionMessagePersistenceRunIds) { - if (!this.finalizedRunsWithDisplay.has(retainedRunId) && retainedRunId !== runId) { - this.sessionMessagePersistenceRunIds.delete(retainedRunId); - } - } } notePostFinalizingRun(runId: string): void { @@ -233,50 +223,18 @@ export class TuiSessionRunCoordinator { notePersistedRun(runId: string): void { this.persistedTerminalRunIds.set(runId, Date.now()); this.pruneRunMap(this.persistedTerminalRunIds); - this.sessionMessagePersistenceRunIds.delete(runId); } - bindRegisteredPendingRun(runId: string): void { - if ( - this.sessionMessageRefreshPending && - this.context.state.pendingSubmit?.runId === runId && - !this.persistedTerminalRunIds.has(runId) - ) { - // A transcript event may precede submit acceptance and its visible final. - this.sessionMessagePersistenceRunIds.add(runId); + routeSessionMessageRefresh(projected: boolean): boolean { + if (projected) { + return true; } - } - - deferSessionMessageRefresh(): boolean { - this.sessionMessageRefreshPending = true; - const activeRunId = this.context.state.activeChatRunId; - if (activeRunId && !this.persistedTerminalRunIds.has(activeRunId)) { - this.sessionMessagePersistenceRunIds.add(activeRunId); + if (this.context.state.activeChatRunId || hasPendingSubmit(this.context.state)) { + this.pendingHistoryRefresh = true; + return true; } - // All already-visible finals must be durable before a destructive rebuild. - for (const runId of this.finalizedRunsWithDisplay.keys()) { - if (!this.persistedTerminalRunIds.has(runId)) { - this.sessionMessagePersistenceRunIds.add(runId); - } - } - return Boolean( - activeRunId || - hasPendingSubmit(this.context.state) || - this.sessionMessagePersistenceRunIds.size, - ); - } - - get hasPendingSessionMessageRefresh(): boolean { - return this.sessionMessageRefreshPending; - } - - get isSessionMessagePersistencePending(): boolean { - return this.sessionMessagePersistenceRunIds.size > 0; - } - - consumeSessionMessageRefresh(): void { - this.sessionMessageRefreshPending = false; - this.sessionMessagePersistenceRunIds.clear(); + this.queueHistoryReload(); + return false; } isHistoryReloadingRun(runId: string): boolean { @@ -440,7 +398,6 @@ export class TuiSessionRunCoordinator { this.rejectUnconfirmedRuns = false; this.historyReloadQueued = false; this.pendingHistoryRefresh = false; - this.consumeSessionMessageRefresh(); this.streamAssembler.clear(); } }