diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index cef349dbfd..e975178da7 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -112,6 +112,21 @@ offline, then Updating messages until authoritative catch-up completes. Socket c certify that the displayed conversation is current. The timeline owner publishes freshness; the view renders it without a toast timer or a separate resume workflow. +The turn footer in a chat follows its messages, not the agent status. Turn activity comes from the +agent directory and messages from the chat's own catch-up, and either can land first. So the timeline +owner keeps an unsettled turn per chat: the turn its messages were captured during. It is set when a +catch-up becomes owed while the turn is open, or when cached messages are painted that were saved +while one was, and it is saved with those messages in the replica. It carries the turn's start time. +While it is set and the agent reads idle, the chat keeps showing that running turn, with its loader +and elapsed timer unchanged, and there is no separate catching-up footer. The commit that brings the +chat current clears it, in the same store write as the rows, so the final messages and the completed +footer land together. A failed catch-up clears it too, and the footer takes the true status. Losing +the connection is not a failed catch-up: none can run or fail until the host returns, so the running +footer stays beside "Reconnecting to host" and the catch-up on reconnect resolves it. Releasing it +offline would put a completed footer under messages known to be behind. Surfaces outside the timeline (the sidebar, tabs, the composer's stop and send controls) read the true status +and never see this state. Live delivery needs no hold because `applyStreamEvent` commits queued rows +before it applies turn liveness. + The draft-create handoff has the same lifetime: the viewed-timeline owner releases it when the sync stops owing that chat a catch-up, not when the first authoritative page lands. Releasing it at the first page leaves a chat this client just created looking hydrated but still catching up, which diff --git a/packages/app/e2e/browser/agent-turn-end.spec.ts b/packages/app/e2e/browser/agent-turn-end.spec.ts index 4c8a1828b1..d5f4ef26d4 100644 --- a/packages/app/e2e/browser/agent-turn-end.spec.ts +++ b/packages/app/e2e/browser/agent-turn-end.spec.ts @@ -21,7 +21,10 @@ interface Box { interface TurnEndFrame { at: number; running: boolean; + replyLength: number; footer: Box | null; + footerText: string; + elapsed: string | null; userRow: Box | null; replyVisible: boolean; } @@ -61,10 +64,13 @@ async function recordTurnEndFrames( const viewport = Array.from( document.querySelectorAll('[data-testid="agent-chat-scroll"]'), ).find((candidate) => isVisible(candidate)); - const footers = Array.from(viewport?.querySelectorAll('[data-testid="turn-footer"]') ?? []) - .map(boxOf) - .filter((box): box is Box => box !== null) - .sort((left, right) => left.top - right.top); + const footer = Array.from(viewport?.querySelectorAll('[data-testid="turn-footer"]') ?? []) + .map((element) => ({ element, box: boxOf(element) })) + .filter((row): row is { element: Element; box: Box } => row.box !== null) + .sort((left, right) => left.box.top - right.box.top) + .at(-1); + const elapsed = + viewport?.querySelector('[data-testid="turn-working-elapsed"]')?.textContent ?? null; const userRow = Array.from( viewport?.querySelectorAll('[data-testid="user-message"]') ?? [], ).find((candidate) => candidate.textContent?.includes(prompt)); @@ -74,10 +80,16 @@ async function recordTurnEndFrames( : Array.from(viewport?.querySelectorAll('[data-testid="assistant-message"]') ?? []).find( (candidate) => candidate.textContent?.includes(reply), ); + const assistantRows = Array.from( + viewport?.querySelectorAll('[data-testid="assistant-message"]') ?? [], + ); state.frames.push({ at: performance.now(), running: Boolean(viewport?.querySelector('[data-testid="turn-working-indicator"]')), - footer: footers.at(-1) ?? null, + replyLength: assistantRows.reduce((sum, row) => sum + (row.textContent?.length ?? 0), 0), + footer: footer?.box ?? null, + footerText: (footer?.element.textContent ?? "").replace(elapsed ?? "", ""), + elapsed, userRow: boxOf(userRow), replyVisible: boxOf(replyRow) !== null, }); @@ -147,7 +159,7 @@ function describeFrames(frames: TurnEndFrame[]): string { (frame, index) => `${index}: ${frame.running ? "running " : "complete"} footer=${ frame.footer ? `${frame.footer.top.toFixed(1)}+${frame.footer.height.toFixed(1)}` : "none" - } prompt=${frame.userRow?.top.toFixed(1) ?? "none"} reply=${frame.replyVisible}`, + } timer=${frame.elapsed ?? "none"} prompt=${frame.userRow?.top.toFixed(1) ?? "none"} reply=${frame.replyVisible}`, ) .join("\n"); } @@ -182,6 +194,192 @@ function expectSteadyTurnEnd(frames: TurnEndFrame[], input: { reply: string | nu ).toBeLessThanOrEqual(1); } +interface ReturnedTurn { + frames: TurnEndFrame[]; + reconnectedAt: number; + finalReplyLength: number; +} + +// The agent finishes while the socket is down, then the catch-up is held so the idle status +// reaches the chat first. Painted frames are sampled from the running footer until everything +// has settled. +async function returnAfterTurnFinished(input: { + page: Page; + agent: MockAgentWorkspace; + prompt: string; + leaveWhile: (gate: DaemonWebSocketGate) => Promise; +}): Promise { + const { page, agent, prompt, leaveWhile } = input; + const gate = await installDaemonWebSocketGate(page); + await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId }); + await expect(page.getByRole("textbox", { name: "Message agent..." })).toBeVisible(); + + await recordTurnEndFrames(page, { prompt, reply: "" }); + await leaveWhile(gate); + gate.holdTimelineResponses(agent.agentId); + const reconnectedAt = await page.evaluate(() => performance.now()); + gate.restore(); + await gate.waitForHeldTimelineResponse(); + await expectAgentIdle(page); + + await recordPaintsFor(page, 2_000); + gate.releaseHeldTimelineResponses(); + await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0); + await expect(page.getByTestId("turn-footer").last()).toBeVisible(); + await recordPaintsFor(page, 500); + const frames = await stopRecording(page); + return { frames, reconnectedAt, finalReplyLength: frames.at(-1)?.replyLength ?? 0 }; +} + +type DaemonWebSocketGate = Awaited>; + +async function leaveWithOnlyThePromptOnScreen( + page: Page, + agent: MockAgentWorkspace, + gate: DaemonWebSocketGate, + prompt: string, +): Promise { + gate.setAgentStreamItemSuppressed("assistant_message", true); + gate.holdNextAgentStreamEvent("turn_completed"); + gate.holdNextAgentUpdate(agent.agentId, "idle"); + gate.setServerMessageSuppressed("agent_update", true); + await submitMessage(page, prompt); + await Promise.all([ + gate.waitForHeldAgentStreamEvent("turn_completed"), + gate.waitForHeldAgentUpdate(agent.agentId, "idle"), + ]); + await expect(page.getByTestId("turn-working-indicator")).toBeVisible(); + await gate.drop(); + await agent.client.waitForFinish(agent.agentId, 15_000); + gate.setAgentStreamItemSuppressed("assistant_message", false); + gate.setServerMessageSuppressed("agent_update", false); +} + +async function leaveWithAPartialReplyOnScreen( + page: Page, + agent: MockAgentWorkspace, + gate: DaemonWebSocketGate, + prompt: string, +): Promise { + await agent.client.sendAgentMessage(agent.agentId, prompt); + await expect(page.getByTestId("turn-working-indicator")).toBeVisible(); + await expect(page.getByTestId("assistant-message").first()).toBeVisible(); + await gate.drop(); + await agent.client.waitForFinish(agent.agentId, 30_000); +} + +function elapsedSeconds(frame: TurnEndFrame): number { + const units: Record = { h: 3600, m: 60, s: 1 }; + return [...(frame.elapsed ?? "").matchAll(/(\d+)([hms])/g)].reduce( + (total, [, value, unit]) => total + Number(value) * units[unit], + 0, + ); +} + +function expectFooterUnchangedUntilMessagesLand( + { frames, reconnectedAt }: ReturnedTurn, + record: string, +): void { + const held = frames.slice( + frames.findIndex((frame) => frame.at >= reconnectedAt), + frames.findIndex((frame) => frame.at >= reconnectedAt && !frame.running), + ); + expect( + held.length, + `the running footer must hold until the messages land\n${record}`, + ).toBeGreaterThan(0); + expect( + new Set(held.map((frame) => `${frame.footer?.top}+${frame.footer?.height}`)).size, + `the held footer must not move or resize\n${record}`, + ).toBe(1); + expect( + new Set(held.map((frame) => frame.footerText)).size, + `the held footer's content must not change\n${record}`, + ).toBe(1); + expect( + held.filter((frame) => frame.elapsed === null).map((frame) => frame.at), + `the timer must stay on screen\n${record}`, + ).toEqual([]); + const seconds = held.map(elapsedSeconds); + expect(seconds, `the timer must keep counting up\n${record}`).toEqual( + seconds.toSorted((left, right) => left - right), + ); +} + +function expectReturnLandsWhole(returned: ReturnedTurn): void { + const { frames, finalReplyLength } = returned; + const record = `frames:\n${describeFrames(frames)}`; + const firstRunning = frames.findIndex((frame) => frame.running && frame.footer !== null); + expect(firstRunning, record).toBeGreaterThanOrEqual(0); + const fromRunning = frames.slice(firstRunning); + + expect( + fromRunning.filter((frame) => frame.footer === null).map((frame) => frame.at), + `the footer must never disappear\n${record}`, + ).toEqual([]); + + const completed = fromRunning.filter((frame) => !frame.running); + expect(completed.length, record).toBeGreaterThan(0); + // The first Markdown on a page paints an empty row until the web parser loads. Only text that + // is present and short of the final reply is stale. + expect( + completed + .filter((frame) => frame.replyLength > 0 && frame.replyLength !== finalReplyLength) + .map((frame) => frame.at), + `a completed footer must never sit under stale messages\n${record}`, + ).toEqual([]); + + expectFooterUnchangedUntilMessagesLand(returned, record); +} + +async function withFirstTurnAgent( + model: string, + run: (agent: MockAgentWorkspace) => Promise, +): Promise { + const agent = await seedMockAgentWorkspace({ + repoPrefix: "turn-return-", + title: "Turn return", + model, + }); + try { + await run(agent); + } finally { + await agent.cleanup(); + } +} + +test.describe("Returning to a turn that finished while away", () => { + test("keeps the running footer until the messages land when only the prompt was on screen", async ({ + page, + }) => { + await withFirstTurnAgent("e2e-fast-stream", async (agent) => { + const prompt = "emit 1 coalesced agent stream updates"; + const returned = await returnAfterTurnFinished({ + page, + agent, + prompt, + leaveWhile: (gate) => leaveWithOnlyThePromptOnScreen(page, agent, gate, prompt), + }); + expectReturnLandsWhole(returned); + }); + }); + + test("keeps the running footer until the messages land under a partial reply", async ({ + page, + }) => { + await withFirstTurnAgent("ten-second-stream", async (agent) => { + const prompt = "Return partial prompt"; + const returned = await returnAfterTurnFinished({ + page, + agent, + prompt, + leaveWhile: (gate) => leaveWithAPartialReplyOnScreen(page, agent, gate, prompt), + }); + expectReturnLandsWhole(returned); + }); + }); +}); + test.describe("Turn end", () => { test("ends a plain reply without moving or dropping its footer", async ({ page, agent }) => { const input = { prompt: "emit 1 coalesced agent stream updates", reply: "stress-update-0" }; diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index ccae42afbe..a91f335048 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -86,7 +86,7 @@ import { useCreateFlowStore } from "@/stores/create-flow-store"; import { buildDraftStoreKey, generateDraftId } from "@/stores/draft-keys"; import { selectAgentTimelineState, - selectAgentTurnPresentation, + selectAgentTimelineTurnPresentation, type Agent, useSessionStore, } from "@/stores/session-store"; @@ -1411,7 +1411,7 @@ const AgentStreamSection = memo(function AgentStreamSection({ const turnPresentation = useSessionStore( useShallow((state) => agentId - ? selectAgentTurnPresentation(state.sessions[serverId], agentId) + ? selectAgentTimelineTurnPresentation(state.sessions[serverId], agentId) : { isActive: false, isCancelling: false, startedAt: null, turnId: null }, ), ); diff --git a/packages/app/src/runtime/directory-sync/internal/agent-store.ts b/packages/app/src/runtime/directory-sync/internal/agent-store.ts index 1b3f24bcd6..04f27e3112 100644 --- a/packages/app/src/runtime/directory-sync/internal/agent-store.ts +++ b/packages/app/src/runtime/directory-sync/internal/agent-store.ts @@ -206,6 +206,7 @@ export class AgentStoreProjection { const store = useSessionStore.getState(); const removeKey = withoutAgent(agentId); store.setAgentTimelineCursor(this.serverId, removeKey); + store.setAgentTimelineUnsettledTurn(this.serverId, agentId, null); store.setAgentAuthoritativeHistoryApplied(this.serverId, agentId, false); store.setAgentStreamTail(this.serverId, removeKey); store.clearAgentStreamHead(this.serverId, agentId); diff --git a/packages/app/src/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts index 5c90ffa5e1..a55bf51e04 100644 --- a/packages/app/src/runtime/replica-cache/index.test.ts +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -216,6 +216,19 @@ function timeline(text = "Cached") { }; } +it("keeps the unsettled turn with the cached messages", async () => { + const storage = new MemoryStorage(); + const writer = createCache(storage); + const cached = { + ...timeline(), + unsettledTurn: { startedAt: new Date("2026-08-26T09:59:00.000Z"), turnId: "turn-1" }, + }; + writer.commitTimeline(SERVER_ID, "agent-1", cached); + await writer.flush(); + + expect(await createCache(storage).readTimeline(SERVER_ID, "agent-1")).toEqual(cached); +}); + function commitDirectory( cache: ReplicaCache, serverId: string, diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index aa52a1b0c5..e9304513b9 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -20,6 +20,7 @@ import { type AgentToolCallData, type StreamItem, } from "@/types/stream"; +import type { UnsettledTurn } from "@/timeline/turn-liveness"; import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; import { clearLegacyReplicaCache } from "./legacy-cleanup"; import { @@ -67,6 +68,8 @@ export interface CachedTimeline { items: StreamItem[]; range: AgentTimelineCursorState | null; hasOlder: boolean; + /** Set when these messages were captured while a turn was open and not yet caught up. */ + unsettledTurn?: UnsettledTurn; } const PERSIST_DELAY_MS = 1_000; @@ -356,6 +359,9 @@ const StoredTimelineSchema = z.strictObject({ }) .nullable(), hasOlder: z.boolean(), + unsettledTurn: z + .strictObject({ turnId: z.string().nullable(), startedAt: IsoDateSchema.nullable() }) + .optional(), }); type StoredAgent = z.infer; @@ -412,6 +418,16 @@ function deserializeTimeline(stored: StoredTimeline | null): CachedTimeline | nu items: stored.items.map(deserializeTimelineItem), range: stored.range, hasOlder: stored.hasOlder, + ...(stored.unsettledTurn + ? { + unsettledTurn: { + turnId: stored.unsettledTurn.turnId, + startedAt: stored.unsettledTurn.startedAt + ? new Date(stored.unsettledTurn.startedAt) + : null, + }, + } + : {}), }; } @@ -812,6 +828,14 @@ function serializeTimeline(timeline: CachedTimeline): StoredTimeline | null { ? { epoch: range.epoch, startSeq: range.startSeq, endSeq: range.endSeq } : null, hasOlder: canPersistCoverage ? timeline.hasOlder : false, + ...(timeline.unsettledTurn + ? { + unsettledTurn: { + turnId: timeline.unsettledTurn.turnId, + startedAt: timeline.unsettledTurn.startedAt?.toISOString() ?? null, + }, + } + : {}), }; } diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index ee73e0ee76..9cb48d8096 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -56,10 +56,12 @@ import { type WorkspaceAgentActivity, } from "@/utils/workspace-agent-activity"; import { + resolveTimelineTurnPresentation, resolveTurnPresentation, TURN_LIVENESS_IDLE, type TurnLiveness, type TurnPresentation, + type UnsettledTurn, } from "@/timeline/turn-liveness"; export interface AgentRuntimeInfo { @@ -374,6 +376,64 @@ export function selectAgentTurnPresentation( ); } +/** Turn state for the chat's own timeline. Every other surface reads the true status above. */ +export function selectAgentTimelineTurnPresentation( + session: SessionState | undefined, + agentId: string, +): TurnPresentation { + return resolveTimelineTurnPresentation( + selectAgentTurnPresentation(session, agentId), + session?.agentTimelineUnsettledTurn.get(agentId), + ); +} + +/** The unsettled turn a chat's messages carry: the one open right now, or the one it holds. */ +export function selectAgentUnsettledTurn( + session: SessionState | undefined, + agentId: string, +): UnsettledTurn | undefined { + const turn = session?.agents.get(agentId)?.turn ?? session?.agentDetails.get(agentId)?.turn; + if (turn?.phase === "open") return { turnId: turn.turnId, startedAt: turn.startedAt }; + return session?.agentTimelineUnsettledTurn.get(agentId); +} + +function withUnsettledTurn( + current: Map, + agentId: string, + turn: UnsettledTurn, +): Map { + const held = current.get(agentId); + // COMPAT(agentTurnIdentity): added in v0.2.6, remove after 2027-01-31 once daemon floor >= v0.2.6. + // Hosts without turn ids report null for every turn, so only the start time tells two turns apart. + if (held?.turnId === turn.turnId && held?.startedAt?.getTime() === turn.startedAt?.getTime()) { + return current; + } + const next = new Map(current); + next.set(agentId, turn); + return next; +} + +function withoutUnsettledTurn( + current: Map, + agentId: string, +): Map { + if (!current.has(agentId)) return current; + const next = new Map(current); + next.delete(agentId); + return next; +} + +// The messages that bring a chat current settle its turn in this same commit. +function commitUnsettledTurn( + current: Map, + agentId: string, + commit: { synchronized: boolean; unsettledTurn?: UnsettledTurn }, +): Map { + if (commit.synchronized) return withoutUnsettledTurn(current, agentId); + if (commit.unsettledTurn) return withUnsettledTurn(current, agentId, commit.unsettledTurn); + return current; +} + function latestTasksFromStream(items: readonly StreamItem[]): TodoEntry[] { for (let index = items.length - 1; index >= 0; index -= 1) { const item = items[index]; @@ -426,6 +486,7 @@ export interface SessionState { agentTasks: Map; messageSubmissions: Map; agentTimelineCursor: Map; + agentTimelineUnsettledTurn: Map; agentTimelineHasOlder: Map; agentTimelineHasNewer: Map; agentTimelineOlderFetchInFlight: Map; @@ -569,8 +630,14 @@ interface SessionStoreActions { newer: boolean; synchronized: boolean; acknowledgedClientMessageIds: string[]; + unsettledTurn?: UnsettledTurn; }, ) => void; + setAgentTimelineUnsettledTurn: ( + serverId: string, + agentId: string, + turn: UnsettledTurn | null, + ) => void; // Initializing agents setInitializingAgents: ( @@ -674,6 +741,7 @@ function createInitialSessionState( agentTasks: new Map(), messageSubmissions: new Map(), agentTimelineCursor: new Map(), + agentTimelineUnsettledTurn: new Map(), agentTimelineHasOlder: new Map(), agentTimelineHasNewer: new Map(), agentTimelineOlderFetchInFlight: new Map(), @@ -1455,6 +1523,11 @@ export const useSessionStore = create()( nextAuthoritative.set(agentId, true); nextSyncGeneration.set(agentId, session.historySyncGeneration); } + const agentTimelineUnsettledTurn = commitUnsettledTurn( + session.agentTimelineUnsettledTurn, + agentId, + state, + ); const tasks = latestTasksFromStream([...state.items, ...state.head]); const agentTasks = new Map(session.agentTasks); if (tasks.length > 0) agentTasks.set(agentId, tasks); @@ -1472,6 +1545,7 @@ export const useSessionStore = create()( agentTimelineCursor: nextCursor, agentTimelineHasOlder: nextHasOlder, agentTimelineHasNewer: nextHasNewer, + agentTimelineUnsettledTurn, agentAuthoritativeHistoryApplied: nextAuthoritative, agentHistorySyncGeneration: nextSyncGeneration, messageSubmissions, @@ -1481,6 +1555,25 @@ export const useSessionStore = create()( }); }, + setAgentTimelineUnsettledTurn: (serverId, agentId, turn) => { + set((prev) => { + const session = prev.sessions[serverId]; + if (!session) return prev; + const current = session.agentTimelineUnsettledTurn; + const next = turn + ? withUnsettledTurn(current, agentId, turn) + : withoutUnsettledTurn(current, agentId); + if (next === current) return prev; + return { + ...prev, + sessions: { + ...prev.sessions, + [serverId]: { ...session, agentTimelineUnsettledTurn: next }, + }, + }; + }); + }, + // Initializing agents setInitializingAgents: (serverId, state) => { set((prev) => { diff --git a/packages/app/src/stores/workspace-layout-store.test.ts b/packages/app/src/stores/workspace-layout-store.test.ts index 6096a55436..75093f4b8c 100644 --- a/packages/app/src/stores/workspace-layout-store.test.ts +++ b/packages/app/src/stores/workspace-layout-store.test.ts @@ -174,6 +174,8 @@ it("feeds restored layout to timeline sync as a release signal, not a subscripti fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }), onCatchUpEnded: () => undefined, reportError: () => undefined, + onCatchUpOwed: () => undefined, + onCatchUpSettled: () => undefined, }); // The edge session-context.tsx owns. Pointing it at subscription instead would resume every // agent behind a restored tab on the daemon. diff --git a/packages/app/src/timeline/replica.test.ts b/packages/app/src/timeline/replica.test.ts index 0ce9b8ca2b..d18bae6ae3 100644 --- a/packages/app/src/timeline/replica.test.ts +++ b/packages/app/src/timeline/replica.test.ts @@ -8,15 +8,23 @@ import { import { afterEach, describe, expect, it } from "vitest"; import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages"; import type { CachedTimeline } from "@/runtime/replica-cache"; -import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store"; +import { + selectAgentTimelineState, + selectAgentTimelineTurnPresentation, + selectAgentTurnPresentation, + useSessionStore, + type Agent, +} from "@/stores/session-store"; import { useCreateFlowStore, type PendingCreateAttempt } from "@/stores/create-flow-store"; import type { StreamItem } from "@/types/stream"; -import type { TurnLivenessTransition } from "./turn-liveness"; +import { reduceTurnLiveness, type TurnLivenessTransition } from "./turn-liveness"; import { createTimelineReplica, createViewedTimelineOwner, type TimelineReplicaStorage, + type TimelineResponsePayload, type ViewedTimelineOwner, + type ViewedTimelineOwnerPorts, } from "./viewed-timeline-sync"; const SERVER_ID = "timeline-replica-host"; @@ -47,6 +55,14 @@ function createOwner( agentId: string, transitions: readonly TurnLivenessTransition[], ) => void = () => undefined, + fetchPage: ViewedTimelineOwnerPorts["fetchPage"] = async () => ({ + hasNewer: false, + endCursor: null, + }), + observe: ViewedTimelineOwnerPorts["observe"] = () => ({ + ready: Promise.resolve(), + release: async () => undefined, + }), ): ViewedTimelineOwner { const replica = createTimelineReplica({ serverId: SERVER_ID, @@ -60,9 +76,9 @@ function createOwner( drainQueuedAgentMessage: () => undefined, applyTurnLiveness, ports: { - observe: () => ({ ready: Promise.resolve(), release: async () => undefined }), + observe, readCursor: () => undefined, - fetchPage: async () => ({ hasNewer: false, endCursor: null }), + fetchPage, fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }), reportError: () => undefined, }, @@ -626,6 +642,471 @@ describe("turn boundaries", () => { }); }); +describe("the turn footer changes together with its messages", () => { + const OPEN_TURN = { turnId: "turn-1", startedAt: new Date("2026-08-26T09:59:00.000Z") }; + + function setTurn(transition: TurnLivenessTransition): void { + useSessionStore.getState().setAgents(SERVER_ID, (agents) => { + const agent = + agents.get(AGENT_ID) ?? + ({ + id: AGENT_ID, + turn: { phase: "idle", cancellationRequestId: null }, + } as Agent); + const next = new Map(agents); + next.set(AGENT_ID, { ...agent, turn: reduceTurnLiveness(agent.turn, transition) }); + return next; + }); + } + const agentIsRunning = () => setTurn({ type: "snapshot", activeTurn: OPEN_TURN }); + const idleSnapshotArrives = () => setTurn({ type: "snapshot", activeTurn: null }); + + type Footer = "running" | "settled"; + function footer(): Footer { + const session = useSessionStore.getState().sessions[SERVER_ID]; + const turn = selectAgentTimelineTurnPresentation(session, AGENT_ID); + return turn.isActive ? "running" : "settled"; + } + function timerStart(): Date | null { + const session = useSessionStore.getState().sessions[SERVER_ID]; + return selectAgentTimelineTurnPresentation(session, AGENT_ID).startedAt; + } + function replies(): string[] { + const session = useSessionStore.getState().sessions[SERVER_ID]; + return [ + ...(session?.agentStreamTail.get(AGENT_ID) ?? []), + ...(session?.agentStreamHead.get(AGENT_ID) ?? []), + ] + .filter((entry) => entry.kind === "assistant_message") + .map((entry) => entry.text); + } + + // A chat that was current when the connection dropped, holding `cachedReply` if given. + function createChat( + cachedReply: string | null, + fetchPage?: ViewedTimelineOwnerPorts["fetchPage"], + ): ViewedTimelineOwner { + useSessionStore.getState().initializeSession(SERVER_ID, null); + useSessionStore.getState().applyAgentTimelineResponseState(SERVER_ID, AGENT_ID, { + items: cachedReply ? [item("reply-1", cachedReply, 4)] : [], + head: [], + range: { epoch: "epoch-1", startSeq: 1, endSeq: 4 }, + older: "none", + newer: false, + synchronized: true, + acknowledgedClientMessageIds: [], + }); + return createOwner( + { readTimeline: async () => undefined, commitTimeline: () => undefined }, + undefined, + fetchPage, + ); + } + + async function connectAndCatchUp(owner: ViewedTimelineOwner): Promise { + owner.replaceVisibleAgentIds("test", [AGENT_ID]); + owner.setConnected(true); + await expect.poll(() => owner.getAgentTimelineStatus(AGENT_ID)).toBe("ready"); + } + + function catchUpPage(input: { + replyThroughSeq: number | null; + reply?: string; + }): TimelineResponsePayload { + const end = input.replyThroughSeq ?? 4; + return { + requestId: "catch-up", + agentId: AGENT_ID, + agent: null, + direction: "after", + projection: "projected", + reset: false, + epoch: "epoch-1", + window: { minSeq: 1, maxSeq: end, nextSeq: end + 1 }, + startCursor: input.reply ? { epoch: "epoch-1", seq: 5 } : null, + endCursor: input.reply ? { epoch: "epoch-1", seq: end } : null, + entries: input.reply + ? [ + { + provider: "mock", + item: { type: "assistant_message", text: input.reply }, + timestamp: "2026-08-26T10:00:02.000Z", + seqStart: 5, + seqEnd: end, + sourceSeqRanges: [{ startSeq: 5, endSeq: end }], + collapsed: [], + }, + ] + : [], + error: null, + hasNewer: false, + hasOlder: false, + staleCursor: false, + gap: false, + }; + } + + function recordCommits(): { commits: { footer: Footer; replies: string[] }[]; stop(): void } { + const commits: { footer: Footer; replies: string[] }[] = [ + { footer: footer(), replies: replies() }, + ]; + const stop = useSessionStore.subscribe(() => { + commits.push({ footer: footer(), replies: replies() }); + }); + return { commits, stop }; + } + + it.each([ + ["only the prompt cached", null, ["Done."]], + ["a partial reply cached", "Half", ["HalfDone."]], + ])("snapshot before messages, %s: running until messages land", async (_name, cached, final) => { + const owner = createChat(cached); + agentIsRunning(); + await connectAndCatchUp(owner); + expect(footer()).toBe("running"); + + owner.setConnected(false); + expect(footer()).toBe("running"); + owner.setConnected(true); + const recorder = recordCommits(); + + idleSnapshotArrives(); + expect(footer()).toBe("running"); + + owner.applyTimelineResponse(catchUpPage({ replyThroughSeq: 8, reply: "Done." })); + recorder.stop(); + + expect(footer()).toBe("settled"); + expect(replies()).toEqual(final); + // The messages and the settled footer arrive in one commit, with no state between. + const previous = recorder.commits.at(-2); + expect(previous?.footer).toBe("running"); + expect(previous?.replies).not.toEqual(final); + expect( + recorder.commits.filter( + (commit) => commit.footer === "settled" && commit.replies.join() !== final.join(), + ), + ).toEqual([]); + owner.dispose(); + }); + + it("messages before snapshot: running stays until the idle snapshot, then completes", async () => { + const owner = createChat("Half"); + agentIsRunning(); + await connectAndCatchUp(owner); + owner.setConnected(false); + owner.setConnected(true); + + owner.applyTimelineResponse(catchUpPage({ replyThroughSeq: 8, reply: "Done." })); + expect(replies()).toEqual(["HalfDone."]); + expect(footer()).toBe("running"); + + idleSnapshotArrives(); + expect(footer()).toBe("settled"); + owner.dispose(); + }); + + it("a turn that ended with no new messages settles when the empty catch-up lands", async () => { + const owner = createChat("Done."); + agentIsRunning(); + await connectAndCatchUp(owner); + owner.setConnected(false); + owner.setConnected(true); + idleSnapshotArrives(); + expect(footer()).toBe("running"); + + owner.applyTimelineResponse(catchUpPage({ replyThroughSeq: null })); + + expect(footer()).toBe("settled"); + expect(replies()).toEqual(["Done."]); + owner.dispose(); + }); + + it("keeps the true idle status for every surface outside the timeline", async () => { + const owner = createChat("Half"); + agentIsRunning(); + await connectAndCatchUp(owner); + owner.setConnected(false); + owner.setConnected(true); + idleSnapshotArrives(); + + expect(footer()).toBe("running"); + const session = useSessionStore.getState().sessions[SERVER_ID]; + expect(selectAgentTurnPresentation(session, AGENT_ID).isActive).toBe(false); + owner.dispose(); + }); + + it("keeps the running turn's identity and timer start when the idle status arrives", async () => { + const owner = createChat("Half"); + agentIsRunning(); + await connectAndCatchUp(owner); + owner.setConnected(false); + owner.setConnected(true); + idleSnapshotArrives(); + + const session = useSessionStore.getState().sessions[SERVER_ID]; + expect(selectAgentTimelineTurnPresentation(session, AGENT_ID)).toEqual({ + isActive: true, + isCancelling: false, + startedAt: OPEN_TURN.startedAt, + turnId: "turn-1", + }); + owner.dispose(); + }); + + it("an agent still running on return keeps its running footer", async () => { + const owner = createChat("Half"); + agentIsRunning(); + await connectAndCatchUp(owner); + owner.setConnected(false); + owner.setConnected(true); + agentIsRunning(); + + expect(footer()).toBe("running"); + owner.applyTimelineResponse(catchUpPage({ replyThroughSeq: 6, reply: " more" })); + expect(footer()).toBe("running"); + owner.dispose(); + }); + + it("applies a snapshot at once when the chat is current", async () => { + const owner = createChat("Done."); + agentIsRunning(); + await connectAndCatchUp(owner); + + idleSnapshotArrives(); + + expect(footer()).toBe("settled"); + owner.dispose(); + }); + + it("a reconnect that finds nothing changed never leaves the settled footer", async () => { + const owner = createChat("Done."); + await connectAndCatchUp(owner); + const recorder = recordCommits(); + + owner.setConnected(false); + owner.setConnected(true); + idleSnapshotArrives(); + owner.applyTimelineResponse(catchUpPage({ replyThroughSeq: null })); + recorder.stop(); + + expect(recorder.commits.map((commit) => commit.footer)).not.toContain("running"); + owner.dispose(); + }); + + it("catch-up failure gives the footer the true status", async () => { + let failNext = false; + const owner = createChat("Half", async () => { + if (failNext) throw new Error("catch-up failed"); + return { hasNewer: false, endCursor: null }; + }); + agentIsRunning(); + await connectAndCatchUp(owner); + failNext = true; + owner.setConnected(false); + owner.setConnected(true); + idleSnapshotArrives(); + + await expect.poll(() => owner.getAgentTimelineStatus(AGENT_ID)).toBe("error"); + + expect(footer()).toBe("settled"); + owner.dispose(); + }); + + it("a catch-up that completes without a page still releases the footer", async () => { + const owner = createChat("Half"); + agentIsRunning(); + await connectAndCatchUp(owner); + owner.setConnected(false); + idleSnapshotArrives(); + expect(footer()).toBe("running"); + + owner.setConnected(true); + + await expect.poll(() => owner.getAgentTimelineStatus(AGENT_ID)).toBe("ready"); + expect(footer()).toBe("settled"); + owner.dispose(); + }); + + it("a chat closed while running and reopened after it finished keeps the running footer", async () => { + const owner = createChat("Half"); + agentIsRunning(); + await connectAndCatchUp(owner); + owner.replaceVisibleAgentIds("test", []); + owner.replaceOpenTabAgentIds([]); + idleSnapshotArrives(); + + owner.replaceVisibleAgentIds("test", [AGENT_ID]); + expect(footer()).toBe("running"); + + owner.applyTimelineResponse(catchUpPage({ replyThroughSeq: 8, reply: "Done." })); + expect(footer()).toBe("settled"); + owner.dispose(); + }); + + it.each([ + { host: "that reports turn ids", olderTurnId: "turn-1", newerTurnId: "turn-2" }, + { host: "without turn ids", olderTurnId: null, newerTurnId: null }, + ])("on a host $host, a newer turn replaces the held one so the timer follows it", async (ids) => { + const newerTurn = { turnId: ids.newerTurnId, startedAt: new Date("2026-08-26T09:59:30.000Z") }; + const owner = createChat("Half"); + setTurn({ + type: "snapshot", + activeTurn: { turnId: ids.olderTurnId, startedAt: OPEN_TURN.startedAt }, + }); + await connectAndCatchUp(owner); + owner.replaceVisibleAgentIds("test", []); + owner.replaceOpenTabAgentIds([]); + idleSnapshotArrives(); + + setTurn({ type: "snapshot", activeTurn: newerTurn }); + owner.replaceVisibleAgentIds("test", [AGENT_ID]); + idleSnapshotArrives(); + + expect(footer()).toBe("running"); + expect(timerStart()).toEqual(newerTurn.startedAt); + owner.dispose(); + }); + + function paintChat(unsettledTurn: CachedTimeline["unsettledTurn"]): ViewedTimelineOwner { + useSessionStore.getState().initializeSession(SERVER_ID, null); + idleSnapshotArrives(); + return createOwner({ + readTimeline: async () => ({ + agentId: AGENT_ID, + items: [item("reply-1", "Half", 4)], + range: { epoch: "epoch-1", startSeq: 1, endSeq: 4 }, + hasOlder: false, + ...(unsettledTurn ? { unsettledTurn } : {}), + }), + commitTimeline: () => undefined, + }); + } + + it("first paint from cache: shows the running footer in the first frame when the cached messages were from an open turn", async () => { + const owner = paintChat({ turnId: "turn-1", startedAt: OPEN_TURN.startedAt }); + const recorder = recordCommits(); + + owner.replaceVisibleAgentIds("test", [AGENT_ID]); + await expect.poll(() => replies()).toEqual(["Half"]); + recorder.stop(); + + expect(recorder.commits.find((commit) => commit.replies.length > 0)?.footer).toBe("running"); + expect(timerStart()).toEqual(OPEN_TURN.startedAt); + owner.applyTimelineResponse(catchUpPage({ replyThroughSeq: 8, reply: "Done." })); + expect(footer()).toBe("settled"); + owner.dispose(); + }); + + async function subscriptionRefusedDuringCacheRead(): Promise<{ + owner: ViewedTimelineOwner; + finishCacheRead(): void; + }> { + useSessionStore.getState().initializeSession(SERVER_ID, null); + idleSnapshotArrives(); + let resolveRead: (timeline: CachedTimeline) => void = () => undefined; + const owner = createOwner( + { + readTimeline: () => new Promise((resolve) => (resolveRead = resolve)), + commitTimeline: () => undefined, + }, + undefined, + undefined, + () => ({ + ready: Promise.reject(new Error("subscription refused")), + release: async () => undefined, + }), + ); + owner.replaceVisibleAgentIds("test", [AGENT_ID]); + owner.setConnected(true); + await expect.poll(() => owner.getAgentTimelineStatus(AGENT_ID)).toBe("error"); + return { + owner, + finishCacheRead: () => + resolveRead({ + agentId: AGENT_ID, + items: [item("reply-1", "Half", 4)], + range: { epoch: "epoch-1", startSeq: 1, endSeq: 4 }, + hasOlder: false, + unsettledTurn: { turnId: "turn-1", startedAt: OPEN_TURN.startedAt }, + }), + }; + } + + it("first paint from cache: a catch-up that failed during the cache read keeps the true status", async () => { + const { owner, finishCacheRead } = await subscriptionRefusedDuringCacheRead(); + + finishCacheRead(); + await expect.poll(() => replies()).toEqual(["Half"]); + + expect(owner.getAgentTimelineStatus(AGENT_ID)).toBe("error"); + expect(footer()).toBe("settled"); + owner.dispose(); + }); + + it("first paint from cache: a catch-up owed again during the cache read keeps the running footer", async () => { + const { owner, finishCacheRead } = await subscriptionRefusedDuringCacheRead(); + owner.setConnected(false); + + finishCacheRead(); + await expect.poll(() => replies()).toEqual(["Half"]); + + expect(footer()).toBe("running"); + owner.dispose(); + }); + + it("first paint from cache: shows the settled footer at once when the cached messages were from a finished turn", async () => { + const owner = paintChat(undefined); + + owner.replaceVisibleAgentIds("test", [AGENT_ID]); + await expect.poll(() => replies()).toEqual(["Half"]); + + expect(footer()).toBe("settled"); + owner.dispose(); + }); + + it("first paint from cache: records the open turn with the messages and drops it once the turn ends", async () => { + useSessionStore.getState().initializeSession(SERVER_ID, null); + applySynced(AGENT_ID, 8); + agentIsRunning(); + const commits: CachedTimeline[] = []; + const owner = createOwner( + { + readTimeline: async () => undefined, + commitTimeline: (_serverId, _agentId, timeline) => commits.push(timeline), + }, + (_agentId, transitions) => { + for (const transition of transitions) setTurn(transition); + }, + ); + owner.applyStreamEvent(AGENT_ID, { + event: { + type: "timeline", + provider: "codex", + item: { type: "assistant_message", text: "Working" }, + } as AgentStreamEventPayload, + seq: 9, + epoch: "epoch-1", + timestamp: new Date("2026-08-26T10:00:01.000Z"), + }); + owner.flushStreamAgent(AGENT_ID); + expect(commits.at(-1)?.unsettledTurn).toEqual({ + turnId: "turn-1", + startedAt: expect.any(Date), + }); + + owner.applyStreamEvent(AGENT_ID, { + event: { type: "turn_completed", provider: "codex", turnId: "turn-1" }, + seq: undefined, + epoch: undefined, + timestamp: new Date("2026-08-26T10:00:02.000Z"), + }); + + expect(commits.at(-1)?.unsettledTurn).toBeUndefined(); + owner.dispose(); + }); +}); + describe("create handoff lifetime", () => { const DRAFT_ID = "draft-1"; diff --git a/packages/app/src/timeline/turn-liveness.ts b/packages/app/src/timeline/turn-liveness.ts index deadee830e..6502f6580a 100644 --- a/packages/app/src/timeline/turn-liveness.ts +++ b/packages/app/src/timeline/turn-liveness.ts @@ -54,6 +54,16 @@ export interface TurnPresentation { turnId: string | null; } +/** + * The turn a chat's messages were captured during, kept until the chat has caught up past it. + * While it is held and the agent reads idle, the messages on screen may be older than the turn's + * end, so the chat cannot claim the turn is finished yet. + */ +export interface UnsettledTurn { + turnId: string | null; + startedAt: Date | null; +} + export function resolveTurnPresentation( liveness: TurnLiveness, hasActiveSubmission: boolean, @@ -74,6 +84,24 @@ export function resolveTurnPresentation( }; } +/** + * What a chat shows for its turn. It differs from the true status only while the agent has + * finished and the chat's messages have not caught up: the chat keeps showing the running turn, so + * the completed footer can land with the messages it belongs to. + */ +export function resolveTimelineTurnPresentation( + turn: TurnPresentation, + unsettledTurn: UnsettledTurn | undefined, +): TurnPresentation { + if (turn.isActive || !unsettledTurn) return turn; + return { + isActive: true, + isCancelling: false, + startedAt: unsettledTurn.startedAt, + turnId: unsettledTurn.turnId, + }; +} + function closeTurn(current: TurnLiveness, turnId: string | null): TurnLiveness { // COMPAT(agentTurnIdentity): added in v0.2.6, remove after 2027-01-31 once daemon floor >= v0.2.6. const targetsDifferentIdentifiedTurn = diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index e556da612f..eeb000aefb 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -142,6 +142,8 @@ class TimelineWorld { status: this.sync.getAgentTimelineStatus(agentId), }); }, + onCatchUpOwed: () => undefined, + onCatchUpSettled: () => undefined, prepare: async (agentId) => { this.cacheRequests.push(agentId); this.cacheRequestWaiters.shift()?.(agentId); diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index 0f5369c1c4..94d4b685f8 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -6,6 +6,7 @@ import { import type { CachedTimeline } from "@/runtime/replica-cache"; import { selectAgentTimelineState, + selectAgentUnsettledTurn, useSessionStore, type AgentTimelineCursorState, type AgentTimelineState, @@ -40,6 +41,7 @@ async function prepareCachedTimeline(input: { agentId: string; storage: TimelineReplicaStorage; prepareAgent: (agentId: string) => Promise; + isTurnSettled: () => boolean; }): Promise { const before = useSessionStore.getState().sessions[input.serverId]; const beforeTimeline = selectAgentTimelineState(before, input.agentId); @@ -87,6 +89,7 @@ async function prepareCachedTimeline(input: { newer: false, synchronized: false, acknowledgedClientMessageIds: replacement.acknowledgedClientMessageIds, + unsettledTurn: input.isTurnSettled() ? undefined : stored.unsettledTurn, }); return stored; } @@ -96,11 +99,16 @@ export interface TimelineReplica { readCursor(agentId: string): { epoch: string; endSeq: number } | undefined; readRange(agentId: string): AgentTimelineCursorState | undefined; timelineUpdated(agentId: string): void; + catchUpOwed(agentId: string): void; + catchUpSettled(agentId: string): void; } class TimelineReplicaOwner implements TimelineReplica { private readonly cachedRanges = new Map(); private readonly preparations = new Map>(); + // A cache read in flight still carries the turn as it was saved. Once the catch-up has + // settled, that turn must not come back with the cached messages. + private readonly settledDuringPreparation = new Set(); constructor( private readonly serverId: string, @@ -123,6 +131,7 @@ class TimelineReplicaOwner implements TimelineReplica { const preparation = this.load(agentId).finally(() => { if (this.preparations.get(agentId) === preparation) { this.preparations.delete(agentId); + this.settledDuringPreparation.delete(agentId); } }); this.preparations.set(agentId, preparation); @@ -135,6 +144,7 @@ class TimelineReplicaOwner implements TimelineReplica { agentId, storage: this.storage, prepareAgent: this.prepareAgent, + isTurnSettled: () => this.settledDuringPreparation.has(agentId), }); if (!stored) return; if (stored.range) { @@ -161,8 +171,21 @@ class TimelineReplicaOwner implements TimelineReplica { items: [...timeline.items, ...(session?.agentStreamHead.get(agentId) ?? [])], range: timeline.status === "synced" ? timeline.range : null, hasOlder: timeline.status === "synced" && timeline.older === "available", + unsettledTurn: selectAgentUnsettledTurn(session, agentId), }); } + + catchUpOwed(agentId: string): void { + this.settledDuringPreparation.delete(agentId); + const store = useSessionStore.getState(); + const openTurn = selectAgentUnsettledTurn(store.sessions[this.serverId], agentId); + if (openTurn) store.setAgentTimelineUnsettledTurn(this.serverId, agentId, openTurn); + } + + catchUpSettled(agentId: string): void { + if (this.preparations.has(agentId)) this.settledDuringPreparation.add(agentId); + useSessionStore.getState().setAgentTimelineUnsettledTurn(this.serverId, agentId, null); + } } export function createTimelineReplica(input: { @@ -240,6 +263,7 @@ function commitProcessedTimeline(input: { }); } store.markAgentHistorySynchronized(serverId, agentId); + if (synchronized) store.setAgentTimelineUnsettledTurn(serverId, agentId, null); } function finalizeProcessedTimeline(input: { @@ -330,6 +354,10 @@ export interface ViewedTimelineSyncPorts { * demanded set. Disconnect and backgrounding keep the obligation and do not report here. */ onCatchUpEnded(agentId: string): void; + /** The chat stopped receiving every row, so a catch-up is owed before it is current again. */ + onCatchUpOwed(agentId: string): void; + /** The owed catch-up reached current or failed. The chat can take the true turn status. */ + onCatchUpSettled(agentId: string): void; reportError(error: unknown): void; } @@ -358,7 +386,7 @@ export interface ViewedTimelineSync extends ViewedTimelineUiBridge { export type ViewedTimelineOwnerPorts = Omit< ViewedTimelineSyncPorts, - "prepare" | "replaceDemandedAgentIds" | "onCatchUpEnded" + "prepare" | "replaceDemandedAgentIds" | "onCatchUpEnded" | "onCatchUpOwed" | "onCatchUpSettled" >; export interface ViewedTimelineOwner extends ViewedTimelineSync { @@ -396,6 +424,8 @@ export function createViewedTimelineOwner(input: { replaceDemandedAgentIds: input.replaceDemandedAgentIds, onCatchUpEnded: (agentId) => useCreateFlowStore.getState().clearByAgent({ serverId: input.serverId, agentId }), + onCatchUpOwed: (agentId) => input.replica.catchUpOwed(agentId), + onCatchUpSettled: (agentId) => input.replica.catchUpSettled(agentId), }); const streamQueue = createSessionAgentStreamReducerQueue({ serverId: input.serverId, @@ -423,6 +453,7 @@ export function createViewedTimelineOwner(input: { if (turnLiveness.length === 0) return; streamQueue.flushAgent(agentId); input.applyTurnLiveness(agentId, turnLiveness); + input.replica.timelineUpdated(agentId); }, flushStreamAgent(agentId) { streamQueue.flushAgent(agentId); @@ -537,6 +568,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const wasPending = visibilityCatchUpPending.delete(agentId); const hadError = visibilityCatchUpErrors.delete(agentId); const wasRetrying = manualRetries.delete(agentId); + ports.onCatchUpSettled(agentId); ports.onCatchUpEnded(agentId); if (wasPending || hadError || wasRetrying) notifyListeners(); }; @@ -545,6 +577,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const message = error instanceof Error ? error.message : String(error); let changed = false; for (const agentId of agentIds) { + ports.onCatchUpSettled(agentId); if (manualRetries.delete(agentId)) changed = true; if (visibilityCatchUpPending.delete(agentId)) changed = true; if (visibilityCatchUpErrors.get(agentId) !== message) { @@ -555,6 +588,11 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed if (changed) notifyListeners(); }; + const markCatchUpOwed = (agentId: string) => { + visibilityCatchUpPending.add(agentId); + ports.onCatchUpOwed(agentId); + }; + const cancelCatchUp = (agentId: string) => { catchUpGenerations.set(agentId, (catchUpGenerations.get(agentId) ?? 0) + 1); catchUps.delete(agentId); @@ -749,7 +787,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed if (options.resetCatchUpStatus) { for (const agentId of nextDesired) { if (!visibilityCatchUpPending.has(agentId)) { - visibilityCatchUpPending.add(agentId); + markCatchUpOwed(agentId); statusChanged = true; } if (visibilityCatchUpErrors.delete(agentId)) statusChanged = true; @@ -768,12 +806,13 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed visibilityCatchUpPending.delete(agentId); visibilityCatchUpErrors.delete(agentId); manualRetries.delete(agentId); + ports.onCatchUpOwed(agentId); released.push(agentId); } } for (const agentId of nextDesired) { if (!desired.includes(agentId)) { - visibilityCatchUpPending.add(agentId); + markCatchUpOwed(agentId); visibilityCatchUpErrors.delete(agentId); manualRetries.delete(agentId); ensureCacheLoaded(agentId); @@ -830,8 +869,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed } for (const agentId of visibleAgentIds()) { if (visibilityCatchUpErrors.has(agentId)) continue; - visibilityCatchUpPending.add(agentId); - visibilityCatchUpErrors.delete(agentId); + markCatchUpOwed(agentId); startCatchUp(agentId, { supersede: true }); } notifyListeners();