From f3d38af8662e029b7a0bbabdb36db1e8ff197f99 Mon Sep 17 00:00:00 2001 From: touch2be Date: Thu, 22 Oct 2026 13:50:09 +0200 Subject: [PATCH] fix(chat): preserve terminal state across stale reads and stream gaps Delayed session lists and snapshot reads could overwrite delivered run completion, while initial stream readiness skipped gap repair. Reconcile concurrent entries, guard list writes with request and session revisions, queue hydration repairs, and let terminal turn timestamps stop duration labels. Cover the real HostProvider races with integration regressions. --- .../message-list/turn-footers.test.ts | 18 ++ src/components/message-list/turn-footers.ts | 5 +- .../HostProvider.render.test.tsx | 288 +++++++++++++++++- src/features/host-provider/HostProvider.tsx | 140 ++++++--- .../host-provider/host-event-stream.ts | 11 +- .../host-provider.integration.test.ts | 5 +- .../host-transcript-reconcile.test.ts | 200 ++++++++++++ src/protocol/host-transcript-reconcile.ts | 82 +++++ 8 files changed, 693 insertions(+), 56 deletions(-) create mode 100644 src/protocol/host-transcript-reconcile.test.ts create mode 100644 src/protocol/host-transcript-reconcile.ts diff --git a/src/components/message-list/turn-footers.test.ts b/src/components/message-list/turn-footers.test.ts index 16965ba3..8bfbbf49 100644 --- a/src/components/message-list/turn-footers.test.ts +++ b/src/components/message-list/turn-footers.test.ts @@ -22,6 +22,24 @@ function message( } describe("buildTurnFooterByMessageId", () => { + test("a tool turn without a terminal timestamp keeps its timer running", () => { + const tool = message("tool", "assistant", 1_500); + tool.parts = [ + { id: "tool-1", type: "tool", tool: "shell", state: { status: "running" } }, + ] as MessageEntry["parts"]; + expect( + buildTurnFooterByMessageId([message("user", "user", 1_000), tool], true).get("tool")?.running, + ).toBe(true); + }); + + test("a durable terminal timestamp wins over a stale busy flag", () => { + const footer = buildTurnFooterByMessageId( + [message("user", "user", 1_000), message("answer", "assistant", 1_500, 301_000)], + true, + ).get("answer"); + expect(footer).toMatchObject({ running: false, completedAt: 301_000 }); + }); + test("marks only the active assistant turn as running", () => { const footers = buildTurnFooterByMessageId( [ diff --git a/src/components/message-list/turn-footers.ts b/src/components/message-list/turn-footers.ts index 11b579d2..1348fde1 100644 --- a/src/components/message-list/turn-footers.ts +++ b/src/components/message-list/turn-footers.ts @@ -97,7 +97,10 @@ export function buildTurnFooterByMessageId( for (const assistantTurn of assistantTurns) { const entry = assistantTurn.assistants.at(-1); if (!entry) continue; - const running = isBusy && assistantTurn.user?.info.id === latestUserId; + const running = + isBusy && + assistantTurn.user?.info.id === latestUserId && + typeof entry.info.time.completed !== "number"; const footer = footerFromTranscriptTurn(assistantTurn, running); if (footer) footerByMessageId.set(entry.info.id, footer); } diff --git a/src/features/host-provider/HostProvider.render.test.tsx b/src/features/host-provider/HostProvider.render.test.tsx index 9aef4459..f83eb559 100644 --- a/src/features/host-provider/HostProvider.render.test.tsx +++ b/src/features/host-provider/HostProvider.render.test.tsx @@ -12,7 +12,8 @@ import { } from "@/hooks/use-agent-state"; import { useActiveTranscriptContextMessages } from "@/features/session-transcript/active-session-transcript-provider"; import { IdentityActorProvider } from "@/features/identity/identity-actor-context"; -import type { HostSessionSnapshot, HostSessionSummary } from "@/protocol/host-types"; +import { buildTurnFooterByMessageId } from "@/components/message-list/turn-footers"; +import type { HostEvent, HostSessionSnapshot, HostSessionSummary } from "@/protocol/host-types"; import { STORAGE_KEYS } from "@/lib/constants"; import { makeProjectKey } from "@/hooks/agent-session-utils"; @@ -178,6 +179,291 @@ describe("HostProvider render integration", () => { }); }); + function timerFixture() { + const running: HostSessionSnapshot = { + ...snapshot, + status: "running", + entries: [ + { ...snapshot.entries[0]!, payload: { text: "work", runId: "run-1" } }, + { + id: "start-1", + sessionId: summary.id, + sequence: 2, + kind: "run_started", + payload: { runId: "run-1" }, + createdAt: "2026-01-01T00:00:01.000Z", + }, + { + id: "answer-1", + sessionId: summary.id, + sequence: 3, + kind: "assistant_message", + payload: { runId: "run-1", text: "Done" }, + createdAt: "2026-01-01T00:05:01.000Z", + }, + ], + }; + const terminal: HostEvent = { + sessionId: summary.id, + event: { + type: "entry_appended", + entry: { + id: "done-1", + sessionId: summary.id, + sequence: 4, + kind: "run_completed", + payload: { runId: "run-1" }, + createdAt: "2026-01-01T00:05:01.000Z", + }, + }, + }; + const finished: HostSessionSnapshot = { + ...running, + status: "idle", + entries: [ + ...running.entries, + terminal.event.type === "entry_appended" ? terminal.event.entry : neverEntry(), + ], + }; + return { running, terminal, finished }; + } + + function neverEntry(): never { + throw new Error("Expected durable terminal entry"); + } + + async function renderRunningChat(running: HostSessionSnapshot) { + host.listSessions.mockResolvedValue([{ ...summary, status: "running" }]); + host.readSession.mockResolvedValue(running); + render( + + + + + , + ); + await waitFor(() => expect(workspace.bootState).toBe("ready")); + await act(() => actions.selectSession(summary.id)); + await waitFor(() => { + expect(messages.some((message) => message.info.id === "run:run-1")).toBe(true); + expect(state.isBusy).toBe(true); + }); + } + + test("a delayed Session list cannot restart an already completed turn", async () => { + const { running, terminal } = timerFixture(); + await renderRunningChat(running); + let resolveOldList!: (value: HostSessionSummary[]) => void; + host.listSessions.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveOldList = resolve; + }), + ); + let refresh!: Promise; + await act(async () => { + refresh = actions.renameSession(summary.id, "renamed"); + }); + await waitFor(() => expect(resolveOldList).toBeTypeOf("function")); + host.listSessions.mockResolvedValue([summary]); + const dispatch = host.subscribe.mock.lastCall![0] as (event: HostEvent) => void; + await act(async () => dispatch(terminal)); + expect(state.isBusy).toBe(false); + expect( + messages.find((message) => message.info.id === "run:run-1")?.info.time.completed, + ).toBeDefined(); + await act(async () => { + resolveOldList([{ ...summary, status: "running" }]); + await refresh; + }); + expect(state.isBusy).toBe(false); + expect(buildTurnFooterByMessageId(messages, state.isBusy).get("run:run-1")?.running).toBe( + false, + ); + }); + + test("a delayed reconnect read cannot erase a delivered run completion", async () => { + const { running, terminal } = timerFixture(); + await renderRunningChat(running); + let resolveOldRead!: (value: HostSessionSnapshot) => void; + host.readSession.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveOldRead = resolve; + }), + ); + const connected = host.subscribe.mock.lastCall![2] as () => void; + await act(async () => connected()); + await waitFor(() => expect(resolveOldRead).toBeTypeOf("function")); + host.listSessions.mockResolvedValue([summary]); + const dispatch = host.subscribe.mock.lastCall![0] as (event: HostEvent) => void; + await act(async () => dispatch(terminal)); + expect(state.isBusy).toBe(false); + await act(async () => resolveOldRead(running)); + expect(state.isBusy).toBe(false); + expect( + messages.find((message) => message.info.id === "run:run-1")?.info.time.completed, + ).toBeDefined(); + }); + + test("first stream readiness repairs a completion missed before subscription", async () => { + const { running, finished } = timerFixture(); + host.subscribe.mockImplementation(() => vi.fn()); + await renderRunningChat(running); + host.listSessions.mockResolvedValue([summary]); + host.readSession.mockResolvedValue(finished); + const connected = host.subscribe.mock.lastCall![2] as () => void; + await act(async () => connected()); + await waitFor(() => expect(state.isBusy).toBe(false)); + expect( + messages.find((message) => message.info.id === "run:run-1")?.info.time.completed, + ).toBeDefined(); + }); + + test("stream readiness during an initial read queues a gap-repair read", async () => { + const { running, finished } = timerFixture(); + let resolveInitial!: (value: HostSessionSnapshot) => void; + host.listSessions.mockResolvedValue([{ ...summary, status: "running" }]); + host.readSession.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveInitial = resolve; + }), + ); + host.readSession.mockResolvedValue(finished); + render( + + + + + , + ); + await waitFor(() => expect(workspace.bootState).toBe("ready")); + act(() => { + void actions.selectSession(summary.id); + }); + await waitFor(() => expect(resolveInitial).toBeTypeOf("function")); + const connected = host.subscribe.mock.lastCall![2] as () => void; + await act(async () => connected()); + await act(async () => resolveInitial(running)); + await waitFor(() => expect(state.isBusy).toBe(false)); + expect(host.readSession).toHaveBeenCalledTimes(2); + expect( + messages.find((message) => message.info.id === "run:run-1")?.info.time.completed, + ).toBeDefined(); + }); + + test("a list started before completion cannot undo it even while the newer refresh is pending", async () => { + const { running, terminal } = timerFixture(); + await renderRunningChat(running); + let resolveOldList!: (value: HostSessionSummary[]) => void; + let resolveNewList!: (value: HostSessionSummary[]) => void; + host.listSessions.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveOldList = resolve; + }), + ); + let refresh!: Promise; + await act(async () => { + refresh = actions.renameSession(summary.id, "renamed"); + }); + host.listSessions.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveNewList = resolve; + }), + ); + const dispatch = host.subscribe.mock.lastCall![0] as (event: HostEvent) => void; + await act(async () => dispatch(terminal)); + await act(async () => { + resolveOldList([{ ...summary, status: "running" }]); + await refresh; + }); + expect(state.isBusy).toBe(false); + await act(async () => resolveNewList([summary])); + }); + + test("a delayed idle list cannot clear a newer run-start event without a second list request", async () => { + const { running, terminal } = timerFixture(); + await renderRunningChat(running); + const dispatch = host.subscribe.mock.lastCall![0] as (event: HostEvent) => void; + host.listSessions.mockResolvedValue([summary]); + await act(async () => dispatch(terminal)); + expect(state.isBusy).toBe(false); + let resolveList!: (value: HostSessionSummary[]) => void; + host.listSessions.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveList = resolve; + }), + ); + let refresh!: Promise; + await act(async () => { + refresh = actions.renameSession(summary.id, "renamed"); + }); + await act(async () => + dispatch({ + sessionId: summary.id, + event: { + type: "entry_appended", + entry: { + id: "start-2", + sessionId: summary.id, + sequence: 5, + kind: "run_started", + payload: { runId: "run-2" }, + createdAt: "2026-01-01T00:06:00.000Z", + }, + }, + }), + ); + await act(async () => { + resolveList([summary]); + await refresh; + }); + expect(state.isBusy).toBe(true); + }); + + test("a terminal event during the very first read schedules repair before a stream exists", async () => { + const { running, terminal, finished } = timerFixture(); + let resolveInitial!: (value: HostSessionSnapshot) => void; + host.listSessions.mockResolvedValue([{ ...summary, status: "running" }]); + host.readSession.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveInitial = resolve; + }), + ); + host.readSession.mockResolvedValue(finished); + render( + + + + + , + ); + await waitFor(() => expect(workspace.bootState).toBe("ready")); + act(() => { + void actions.selectSession(summary.id); + }); + await waitFor(() => expect(resolveInitial).toBeTypeOf("function")); + host.listSessions.mockResolvedValue([summary]); + const dispatch = host.subscribe.mock.lastCall![0] as (event: HostEvent) => void; + await act(async () => dispatch(terminal)); + await act(async () => resolveInitial(running)); + await waitFor(() => expect(state.isBusy).toBe(false)); + expect( + messages.find((message) => message.info.id === "run:run-1")?.info.time.completed, + ).toBeDefined(); + }); + test("bootstraps, hydrates, orchestrates Host mutations, persists UI state, and cleans up", async () => { const unsubscribe = vi.fn(); host.subscribe.mockImplementation((_event, _session, connected) => { diff --git a/src/features/host-provider/HostProvider.tsx b/src/features/host-provider/HostProvider.tsx index 9050d688..2df04b19 100644 --- a/src/features/host-provider/HostProvider.tsx +++ b/src/features/host-provider/HostProvider.tsx @@ -30,6 +30,7 @@ import { useActiveTranscriptSnapshot, useActiveTranscriptStore, } from "@/features/session-transcript/active-session-transcript-provider"; +import { reconcileHostTranscriptRead } from "@/protocol/host-transcript-reconcile"; import { createHostClient } from "@/protocol/host-client"; import type { HostSessionSnapshot, @@ -40,7 +41,6 @@ import { applyHostTranscriptEvent, createHostTranscriptStream, projectHostTranscriptStream, - rehydrateHostTranscriptStream, type HostTranscriptStream, } from "@/protocol/host-transcript"; import type { Workspace } from "@/types/workspace"; @@ -262,7 +262,30 @@ function HostProviderBody({ const setLastError = transportSlice.setter("lastError"); const activeSnapshotRef = useRef(null); const activeStreamRef = useRef(null); + const sessionRevisionsRef = useRef(new Map()); + const sessionListRequestRef = useRef(0); + const pendingHydrationIdsRef = useRef(new Set()); + const observeSession = useCallback((sessionId: string) => { + sessionRevisionsRef.current.set( + sessionId, + (sessionRevisionsRef.current.get(sessionId) ?? 0) + 1, + ); + }, []); const hydratingSessionIdsRef = useRef(new Set()); + const observeSessionEvent = useCallback( + (sessionId: string) => { + observeSession(sessionId); + // Before the first snapshot there is no stream to reduce into. Repair after + // the in-flight read so events delivered in that gap cannot be lost. + if ( + hydratingSessionIdsRef.current.has(sessionId) && + activeStreamRef.current?.snapshot.id !== sessionId + ) { + pendingHydrationIdsRef.current.add(sessionId); + } + }, + [observeSession], + ); const activeSessionIdRef = useRef(activeSessionId); const modelSelectionRevisionRef = useRef(0); const modelSelectionMutationRef = useRef>(Promise.resolve()); @@ -355,6 +378,8 @@ function HostProviderBody({ const refreshSessions = useCallback( async (directories = projectsRef.current) => { + const request = ++sessionListRequestRef.current; + const revisions = new Map(sessionRevisionsRef.current); if (directories.length === 0) { setSessions([]); setBusySessionIds(new Set()); @@ -362,6 +387,7 @@ function HostProviderBody({ } if (!host) return; const listed = await loadHostSessionSummaries(host, directories); + if (request !== sessionListRequestRef.current) return; setSessions(listed.map((item) => toSession(item, activeWorkspaceId))); const restoredSessionId = resolveRestoredSessionId( activeSessionIdRef.current, @@ -372,9 +398,17 @@ function HostProviderBody({ activeSessionIdRef.current = restoredSessionId; setActiveSessionId(restoredSessionId); } - setBusySessionIds( - new Set(listed.filter((item) => item.status === "running").map((item) => item.id)), - ); + setBusySessionIds((current) => { + const next = new Set( + listed.filter((item) => item.status === "running").map((item) => item.id), + ); + for (const [id, revision] of sessionRevisionsRef.current) { + if (revision === revisions.get(id)) continue; + if (current.has(id)) next.add(id); + else next.delete(id); + } + return next; + }); setQueuedPrompts((current) => { const next = { ...current }; for (const item of listed) { @@ -396,7 +430,10 @@ function HostProviderBody({ return; } if (!host) return; - if (hydratingSessionIdsRef.current.has(sessionId)) return; + if (hydratingSessionIdsRef.current.has(sessionId)) { + pendingHydrationIdsRef.current.add(sessionId); + return; + } hydratingSessionIdsRef.current.add(sessionId); const scope = { directory: activeTargetDirectoryRef.current ?? "", @@ -404,53 +441,60 @@ function HostProviderBody({ }; transcriptStore.select(scope); try { - const snapshot = await host.readSession(sessionId); - if (activeSessionIdRef.current !== sessionId) return; - const snapshotScope = { directory: snapshot.projectDirectory, sessionId }; - if (snapshotScope.directory !== scope.directory) transcriptStore.select(snapshotScope); - activeSnapshotRef.current = snapshot; - // Keep live text of a Run that is still streaming (send-now reloads mid-Run). - activeStreamRef.current = rehydrateHostTranscriptStream(snapshot, activeStreamRef.current); - modelSelectionRevisionRef.current += 1; - setSelectedModel(selectedModelFromHostSnapshot(snapshot)); - if (snapshot.reasoning) setReasoningEffortState(snapshot.reasoning); - const messages = projectHostTranscriptStream(activeStreamRef.current); - transcriptStore.dispatch({ - type: "page.loaded", - scope: snapshotScope, - phase: "initial", - messages, - hasMore: false, - nextCursor: null, - }); - // Lock skill selection to the first turn's allowlist when present so - // reloads keep the same catalog for prompt-cache stability. - if (messages.length > 0) { - const key = sessionSkillsKey(sessionId, null); - if (key) { - const locked = lockedSkillsFromSnapshot(snapshot); - if (locked) { - setEnabledSkillsByKey((current) => - current[key] ? current : { ...current, [key]: locked }, - ); + do { + pendingHydrationIdsRef.current.delete(sessionId); + const streamAtRead = activeStreamRef.current; + const read = await host.readSession(sessionId); + if (activeSessionIdRef.current !== sessionId) return; + const stream = reconcileHostTranscriptRead(read, activeStreamRef.current, streamAtRead); + const snapshot = stream.snapshot; + observeSession(sessionId); + const snapshotScope = { directory: snapshot.projectDirectory, sessionId }; + if (snapshotScope.directory !== scope.directory) transcriptStore.select(snapshotScope); + activeSnapshotRef.current = snapshot; + activeStreamRef.current = stream; + modelSelectionRevisionRef.current += 1; + setSelectedModel(selectedModelFromHostSnapshot(snapshot)); + if (snapshot.reasoning) setReasoningEffortState(snapshot.reasoning); + const messages = projectHostTranscriptStream(stream); + transcriptStore.dispatch({ + type: "page.loaded", + scope: snapshotScope, + phase: "initial", + messages, + hasMore: false, + nextCursor: null, + }); + // Lock skill selection to the first turn's allowlist when present so + // reloads keep the same catalog for prompt-cache stability. + if (messages.length > 0) { + const key = sessionSkillsKey(sessionId, null); + if (key) { + const locked = lockedSkillsFromSnapshot(snapshot); + if (locked) { + setEnabledSkillsByKey((current) => + current[key] ? current : { ...current, [key]: locked }, + ); + } } } - } - setQueuedPrompts((current) => ({ - ...current, - [sessionId]: projectHostFollowUps(snapshot.followUps), - })); - setBusySessionIds((current) => { - const next = new Set(current); - if (snapshot.status === "running") next.add(sessionId); - else next.delete(sessionId); - return next; - }); + setQueuedPrompts((current) => ({ + ...current, + [sessionId]: projectHostFollowUps(snapshot.followUps), + })); + setBusySessionIds((current) => { + const next = new Set(current); + if (snapshot.status === "running") next.add(sessionId); + else next.delete(sessionId); + return next; + }); + } while (pendingHydrationIdsRef.current.has(sessionId)); } finally { + pendingHydrationIdsRef.current.delete(sessionId); hydratingSessionIdsRef.current.delete(sessionId); } }, - [host, transcriptStore], + [host, transcriptStore, observeSession], ); const forgetAfterPartFollowUp = useCallback((sessionId: string, followUpId: string) => { @@ -536,6 +580,7 @@ function HostProviderBody({ activeSessionIdRef, activeStreamRef, setActiveSnapshot, + onSessionEventObserved: observeSessionEvent, setBusySessionIds, transcriptStore, refreshSessions, @@ -741,6 +786,7 @@ function HostProviderBody({ const compactActiveSession = async () => { if (!activeSessionId) throw new Error(t("compaction.selectSession")); + observeSession(activeSessionId); setBusySessionIds((current) => new Set(current).add(activeSessionId)); try { await requireHost().compact(activeSessionId); @@ -875,6 +921,7 @@ function HostProviderBody({ }), }); } + observeSession(sessionId!); setBusySessionIds((current) => new Set(current).add(sessionId!)); // Prefer the session key; if this was a pending new chat, migrate the // pending allowlist onto the created Session so later turns stay stable. @@ -1264,6 +1311,7 @@ function HostProviderBody({ forgetAfterPartFollowUp, host, hydrateTranscript, + observeSession, rememberActiveSession, queuedPrompts, queueController, diff --git a/src/features/host-provider/host-event-stream.ts b/src/features/host-provider/host-event-stream.ts index 2166407c..1b9ca591 100644 --- a/src/features/host-provider/host-event-stream.ts +++ b/src/features/host-provider/host-event-stream.ts @@ -48,6 +48,7 @@ export interface HostEventDispatcherDependencies { setBusySessionIds: Dispatch>>; transcriptStore: ActiveSessionTranscriptStore; refreshSessions: () => Promise; + onSessionEventObserved?: (sessionId: string) => void; onFollowUpDispatched?: (sessionId: string, followUpId: string) => void; /** Fires once, when the whole assistant turn reaches a terminal entry. */ onTurnFinished?: (sessionId: string) => void; @@ -70,11 +71,13 @@ export function createHostEventDispatcher({ setBusySessionIds, transcriptStore, refreshSessions, + onSessionEventObserved, onFollowUpDispatched, onTurnFinished, onModelPartEnded, }: HostEventDispatcherDependencies): (hostEvent: HostEvent) => void { return (hostEvent) => { + onSessionEventObserved?.(hostEvent.sessionId); const terminal = isTerminalHostEvent(hostEvent); if ( hostEvent.event.type === "entry_appended" && @@ -127,18 +130,13 @@ export function createHostReconnectHandler({ activeSessionIdRef, hydrateTranscript, }: Pick) { - let hasConnected = false; return () => { - if (!hasConnected) { - hasConnected = true; - return; - } const sessionId = activeSessionIdRef.current; if (sessionId) void hydrateTranscript(sessionId).catch(notifyUnknownError); }; } -/** Owns subscription lifetime and rehydrates after every reconnect except initial readiness. */ +/** Owns subscription lifetime and repairs the read/subscription gap on every readiness, including the first. */ export function useHostEventStream(options: UseHostEventStreamOptions): void { const onFollowUpDispatchedRef = useRef(options.onFollowUpDispatched); onFollowUpDispatchedRef.current = options.onFollowUpDispatched; @@ -172,5 +170,6 @@ export function useHostEventStream(options: UseHostEventStreamOptions): void { options.activeStreamRef, options.setActiveSnapshot, options.setBusySessionIds, + options.onSessionEventObserved, ]); } diff --git a/src/features/host-provider/host-provider.integration.test.ts b/src/features/host-provider/host-provider.integration.test.ts index 2ec0a543..f71a0135 100644 --- a/src/features/host-provider/host-provider.integration.test.ts +++ b/src/features/host-provider/host-provider.integration.test.ts @@ -39,7 +39,7 @@ describe("HostProvider public event pipeline", () => { "session-1", ); }); - test("rehydrates the active Session after reconnect but not initial stream readiness", async () => { + test("rehydrates the active Session on initial stream readiness and after reconnect", async () => { const hydrateTranscript = vi.fn(async () => {}); const connected = createHostReconnectHandler({ activeSessionIdRef: { current: "session-1" }, @@ -47,10 +47,11 @@ describe("HostProvider public event pipeline", () => { }); connected(); - expect(hydrateTranscript).not.toHaveBeenCalled(); + expect(hydrateTranscript).toHaveBeenCalledTimes(1); connected(); await Promise.resolve(); expect(hydrateTranscript).toHaveBeenCalledWith("session-1"); + expect(hydrateTranscript).toHaveBeenCalledTimes(2); }); test("projects ordered deltas and durable tool output through the real transcript store", () => { diff --git a/src/protocol/host-transcript-reconcile.test.ts b/src/protocol/host-transcript-reconcile.test.ts new file mode 100644 index 00000000..ff8edee7 --- /dev/null +++ b/src/protocol/host-transcript-reconcile.test.ts @@ -0,0 +1,200 @@ +import { describe, expect, test } from "vite-plus/test"; +import type { HostSessionEntry, HostSessionSnapshot } from "./host-types"; +import { + applyHostTranscriptEvent, + createHostTranscriptStream, + projectHostTranscriptStream, +} from "./host-transcript"; +import { reconcileHostTranscriptRead } from "./host-transcript-reconcile"; + +function entry( + kind: HostSessionEntry["kind"], + sequence: number, + runId = "run-1", +): HostSessionEntry { + return { + id: `entry-${sequence}`, + sessionId: "session-1", + kind, + sequence, + payload: { runId, text: "Working" }, + createdAt: `2026-01-01T00:00:0${sequence}.000Z`, + }; +} +function snapshot(entries: HostSessionEntry[]): HostSessionSnapshot { + return { + id: "session-1", + projectDirectory: "/project", + title: "Test", + status: "running", + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + entries, + followUps: [], + model: null, + reasoning: null, + }; +} + +describe("reconcileHostTranscriptRead", () => { + test.each([ + ["run_completed", "idle"], + ["run_failed", "failed"], + ["run_aborted", "stopped"], + ["run_interrupted", "interrupted"], + ] as const)("keeps a live %s over an older running read", (kind, status) => { + const read = snapshot([entry("run_started", 1), entry("assistant_message", 2)]); + let live = createHostTranscriptStream(read); + live = applyHostTranscriptEvent(live, { + sessionId: read.id, + event: { type: "assistant_delta", runId: "run-1", delta: "partial" }, + }); + live = applyHostTranscriptEvent(live, { + sessionId: read.id, + event: { type: "entry_appended", entry: entry(kind, 3) }, + }); + const merged = reconcileHostTranscriptRead(read, live); + expect(merged.snapshot.status).toBe(status); + expect(merged.snapshot.entries).toHaveLength(3); + expect(merged.assistantTextByRun).toEqual({}); + expect(projectHostTranscriptStream(merged)[0]?.info.time.completed).toBeDefined(); + }); + + test("new run after completion stays running", () => { + const read = snapshot([entry("run_started", 1), entry("run_completed", 2)]); + const live = createHostTranscriptStream({ + ...read, + entries: [...read.entries, entry("run_started", 3, "run-2")], + }); + expect(reconcileHostTranscriptRead(read, live).snapshot.status).toBe("running"); + }); + + test("keeps a delta delivered during a read even when the read has newer durable entries", () => { + const atRead = createHostTranscriptStream(snapshot([entry("run_started", 1)])); + const live = applyHostTranscriptEvent(atRead, { + sessionId: "session-1", + event: { type: "assistant_delta", runId: "run-1", delta: "new text" }, + }); + const merged = reconcileHostTranscriptRead( + snapshot([entry("run_started", 1), entry("tool_call", 2)]), + live, + atRead, + ); + expect(merged.assistantTextByRun).toEqual({ "run-1": "new text" }); + expect(merged.snapshot.status).toBe("running"); + }); + + test("a durable read replaces unchanged partial buffers and uses fresh metadata", () => { + const atRead = { + ...createHostTranscriptStream(snapshot([entry("run_started", 1)])), + assistantTextByRun: { "run-1": "partial" }, + }; + const read = { + ...snapshot([entry("run_started", 1), entry("assistant_message", 2)]), + model: { connectionId: "new", modelId: "new-model" }, + }; + const merged = reconcileHostTranscriptRead(read, atRead, atRead); + expect(merged.assistantTextByRun).toEqual({}); + expect(merged.snapshot.model).toEqual(read.model); + }); + + test("keeps the next model segment's live buffer when its earlier message is already durable", () => { + const read = snapshot([ + entry("run_started", 1), + entry("assistant_message", 2), + entry("tool_call", 3), + ]); + const atRead = { + ...createHostTranscriptStream(read), + assistantTextByRun: { "run-1": "next segment" }, + }; + const merged = reconcileHostTranscriptRead(read, atRead, atRead); + expect(merged.assistantTextByRun).toEqual({ "run-1": "next segment" }); + }); + + test("new tool entries do not discard an unpersisted model segment", () => { + const atRead = { + ...createHostTranscriptStream(snapshot([entry("run_started", 1)])), + assistantTextByRun: { "run-1": "partial" }, + }; + const merged = reconcileHostTranscriptRead( + snapshot([entry("run_started", 1), entry("tool_call", 2)]), + atRead, + atRead, + ); + expect(merged.assistantTextByRun).toEqual({ "run-1": "partial" }); + }); + + test("a newer live entry does not revert the read's model and reasoning metadata", () => { + const read = { + ...snapshot([entry("run_started", 1)]), + model: { connectionId: "new", modelId: "new-model" }, + reasoning: "high" as const, + }; + const live = createHostTranscriptStream(snapshot([...read.entries, entry("run_completed", 2)])); + const merged = reconcileHostTranscriptRead(read, live); + expect(merged.snapshot.model).toEqual(read.model); + expect(merged.snapshot.reasoning).toBe("high"); + expect(merged.snapshot.status).toBe("idle"); + }); + + test("retains a model retry delivered during hydration and clears it on completion", () => { + const read = snapshot([entry("run_started", 1)]); + const atRead = createHostTranscriptStream(read); + const live = applyHostTranscriptEvent(atRead, { + sessionId: read.id, + event: { + type: "model_retry", + delayMs: 1000, + runId: "run-1", + attempt: 1, + maxAttempts: 3, + code: "overloaded", + }, + }); + expect(reconcileHostTranscriptRead(read, live, atRead).retryByRun?.["run-1"]).toEqual({ + attempt: 1, + maxAttempts: 3, + code: "overloaded", + }); + const completed = { + ...read, + status: "idle" as const, + entries: [...read.entries, entry("run_completed", 2)], + }; + expect(reconcileHostTranscriptRead(completed, live, atRead).retryByRun).toEqual({}); + }); + + test("keeps live model and reasoning changes newer than the read", () => { + const read = snapshot([entry("run_started", 1)]); + const model = { connectionId: "live", modelId: "live-model" }; + let live = createHostTranscriptStream(read); + live = applyHostTranscriptEvent(live, { + sessionId: read.id, + event: { + type: "entry_appended", + entry: { ...entry("model_changed", 2), payload: { model } }, + }, + }); + live = applyHostTranscriptEvent(live, { + sessionId: read.id, + event: { + type: "entry_appended", + entry: { ...entry("reasoning_changed", 3), payload: { reasoning: "high" } }, + }, + }); + const merged = reconcileHostTranscriptRead(read, live); + expect(merged.snapshot.model).toEqual(model); + expect(merged.snapshot.reasoning).toBe("high"); + }); + + test("does not merge another session", () => { + const read = snapshot([]); + const live = createHostTranscriptStream({ + ...read, + id: "other", + entries: [entry("run_completed", 1)], + }); + expect(reconcileHostTranscriptRead(read, live)).toEqual(createHostTranscriptStream(read)); + }); +}); diff --git a/src/protocol/host-transcript-reconcile.ts b/src/protocol/host-transcript-reconcile.ts new file mode 100644 index 00000000..2fb4e131 --- /dev/null +++ b/src/protocol/host-transcript-reconcile.ts @@ -0,0 +1,82 @@ +import type { HostSessionSnapshot } from "./host-types"; +import { createHostTranscriptStream, type HostTranscriptStream } from "./host-transcript"; + +/** A full read may finish after newer SSE entries. Never roll the transcript back. */ +export function reconcileHostTranscriptRead( + snapshot: HostSessionSnapshot, + live: HostTranscriptStream | null, + streamAtRead: HostTranscriptStream | null = live, +): HostTranscriptStream { + if (!live || live.snapshot.id !== snapshot.id) return createHostTranscriptStream(snapshot); + const sequence = (value: HostSessionSnapshot) => value.entries.at(-1)?.sequence ?? 0; + const liveIsNewer = sequence(live.snapshot) > sequence(snapshot); + const entries = new Map(snapshot.entries.map((entry) => [entry.id, entry])); + for (const entry of live.snapshot.entries) entries.set(entry.id, entry); + const merged = [...entries.values()].sort((a, b) => a.sequence - b.sequence); + const lifecycle = merged.findLast((entry) => + ["run_started", "run_completed", "run_failed", "run_aborted", "run_interrupted"].includes( + entry.kind, + ), + ); + const status = + liveIsNewer && lifecycle + ? ({ + run_started: "running", + run_completed: "idle", + run_failed: "failed", + run_aborted: "stopped", + run_interrupted: "interrupted", + }[lifecycle.kind] as HostSessionSnapshot["status"]) + : snapshot.status; + // Durable entries clear buffers only when they are new to the live stream. + // An earlier assistant_message from the same Run must not erase its next segment. + const liveEntryIds = new Set(live.snapshot.entries.map((entry) => entry.id)); + const assistantTextByRun = { ...live.assistantTextByRun }; + const reasoningTextByRun = { ...live.reasoningTextByRun }; + const retryByRun = { ...live.retryByRun }; + for (const entry of snapshot.entries) { + if (liveEntryIds.has(entry.id)) continue; + const runId = typeof entry.payload.runId === "string" ? entry.payload.runId : ""; + if ( + entry.kind === "assistant_message" && + live.assistantTextByRun[runId] === streamAtRead?.assistantTextByRun[runId] + ) + delete assistantTextByRun[runId]; + if ( + entry.kind === "assistant_reasoning" && + live.reasoningTextByRun[runId] === streamAtRead?.reasoningTextByRun[runId] + ) + delete reasoningTextByRun[runId]; + } + for (const entry of merged) { + if (["run_completed", "run_failed", "run_aborted", "run_interrupted"].includes(entry.kind)) { + const runId = typeof entry.payload.runId === "string" ? entry.payload.runId : ""; + delete assistantTextByRun[runId]; + delete reasoningTextByRun[runId]; + delete retryByRun[runId]; + } + } + return { + snapshot: { + ...snapshot, + // Preserve metadata changes delivered after this read's durable boundary. + ...(live.snapshot.entries.some( + (entry) => entry.kind === "model_changed" && entry.sequence > sequence(snapshot), + ) + ? { model: live.snapshot.model } + : {}), + ...(live.snapshot.entries.some( + (entry) => entry.kind === "reasoning_changed" && entry.sequence > sequence(snapshot), + ) + ? { reasoning: live.snapshot.reasoning } + : {}), + updatedAt: + live.snapshot.updatedAt > snapshot.updatedAt ? live.snapshot.updatedAt : snapshot.updatedAt, + entries: merged, + status, + }, + assistantTextByRun, + reasoningTextByRun, + retryByRun, + }; +}