diff --git a/src/components/message-list/turn-footers.test.ts b/src/components/message-list/turn-footers.test.ts index 16965ba..8bfbbf4 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 11b579d..1348fde 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 9aef445..f83eb55 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 9050d68..2df04b1 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 2166407..1b9ca59 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 2ec0a54..f71a013 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 0000000..ff8edee --- /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 0000000..2fb4e13 --- /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, + }; +}