diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 6abe6bfec..71bda619a 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -85,16 +85,22 @@ recomposition while the runtime still owns the same directory snapshot and timel Removing the host from the registry is the destructive boundary: it stops the runtime and clears the session and host-scoped setup state together. -The durable replica cache is a display cache, not a synchronization checkpoint. Its timeline record -contains only the focused `agentId` and a truncated item tail. It never persists a cursor, epoch, -older-history availability, authority status, or sync generation because those facts would describe -the complete source dataset rather than the truncated display dataset. +The durable replica cache persists synchronization authority only when it can store the complete +current canonical window losslessly. The stored range describes those exact items: `startSeq` drives +older pagination and `endSeq` drives forward catch-up. Restore paints the items immediately, requests +`after endSeq`, and requests `before startSeq` when the user loads older history. -Restoring that cache produces a painted timeline: the items may render immediately, but the first -daemon timeline request is still `tail`. A successful tail response atomically establishes canonical -items, range, and older-history availability. Live rows received between cache paint and that tail -response stay in the separate live head, do not advance a cursor or trigger gap recovery, and are -reconciled with the authoritative tail and subsequent catch-up. +The first resume request is bounded. If it reports more newer history, fetch one latest bounded tail +instead of replaying every missed page. Live gap recovery still pages forward until current. + +If the canonical window exceeds the cache item limit, contains a discontiguous retained range, has a +live head, or includes presentation data the cache cannot encode losslessly, persistence drops the +range and keeps a display-only tail. Restore then uses the ordinary bounded `tail` bootstrap. Never +slice items while retaining the pre-slice range; that falsely certifies discarded source rows. + +Live rows received between cache paint and catch-up stay in the separate live head and reconcile with +the authoritative range through the existing forward-page path. The cache does not persist sync +generation or unreconciled local submissions. Every daemon-derived live item carries its timeline epoch and sequence position. Bootstrap replacement keeps only positioned rows newer than the page it installs, while unresolved local diff --git a/packages/app/e2e/browser/agent-message-submission.spec.ts b/packages/app/e2e/browser/agent-message-submission.spec.ts index 528a9343f..ed2bc794a 100644 --- a/packages/app/e2e/browser/agent-message-submission.spec.ts +++ b/packages/app/e2e/browser/agent-message-submission.spec.ts @@ -30,6 +30,12 @@ import { buildHostWorkspaceRoute } from "@/utils/host-routes"; import { delayBrowserAgentCreatedStatus } from "../support/helpers/new-workspace"; import { installDaemonWebSocketGate } from "../support/helpers/daemon-websocket-gate"; import { selectModel } from "../support/helpers/app"; +import { observeTimelineSubscriptions } from "../support/helpers/timeline-delivery"; +import { + expectResumeOverflowFallsBackToOneTail, + rememberTimelineRequestCounts, +} from "../support/helpers/timeline-resume"; +import { workspaceDeckEntryLocator } from "../support/helpers/workspace-ui"; import { scrollTimelineToNewestLoadedEdge, scrollTimelineUntilOlderHistoryIsReachable, @@ -336,6 +342,64 @@ async function expectInterruptedTurnOrderAfterReconnect( } } +async function expectHiddenStreamingSubmissionOrderAfterWorkspaceEviction( + page: Page, + testInfo: { workerIndex: number }, +): Promise { + const subscriptions = observeTimelineSubscriptions(page); + const gate = await installDaemonWebSocketGate(page); + const target = await seedMockAgentWorkspace({ + repoPrefix: `submission-hidden-stream-${testInfo.workerIndex}-`, + title: "Hidden streaming submission", + model: "ten-second-stream", + }); + const evictionAgents = await Promise.all( + Array.from({ length: 3 }, (_unused, index) => + seedMockAgentWorkspace({ + repoPrefix: `submission-workspace-eviction-${testInfo.workerIndex}-${index}-`, + title: `Workspace eviction ${index + 1}`, + }), + ), + ); + const prompt = "Keep this hidden image prompt before its streaming output."; + const targetDeckEntry = workspaceDeckEntryLocator(page, getServerId(), target.workspaceId); + + try { + await openAgentRoute(page, target); + await expectComposerVisible(page); + await subscriptions.waitForSubscribedAgents([target.agentId]); + + const userMessageCount = gate.getAgentStreamItemCount("user_message"); + gate.setAgentStreamSuppressed(true); + const promptRow = await submitMessageWithImage(page, prompt); + await gate.waitForAgentStreamItem("user_message", userMessageCount + 1); + + for (const evictionAgent of evictionAgents) { + await openAgentRoute(page, evictionAgent); + await expectComposerVisible(page); + } + await expect(targetDeckEntry).toHaveCount(0); + await subscriptions.waitForSubscribedAgents([evictionAgents[2]!.agentId]); + gate.setAgentStreamSuppressed(false); + + await target.client.waitForFinish(target.agentId, 30_000); + const requestsBeforeReturn = rememberTimelineRequestCounts(gate); + await openAgentRoute(page, target); + await expectComposerVisible(page); + await subscriptions.waitForSubscribedAgents([target.agentId]); + + const response = page.getByText("(end of synthetic stream)", { exact: true }).last(); + await expect(promptRow).toBeVisible(); + await expect(response).toBeVisible(); + await expectRenderedBefore(promptRow, response); + expectResumeOverflowFallsBackToOneTail(gate, requestsBeforeReturn); + } finally { + gate.setAgentStreamSuppressed(false); + gate.restore(); + await Promise.all([...evictionAgents.map((agent) => agent.cleanup()), target.cleanup()]); + } +} + async function expectCompletedSubmissionClearsAfterMissedRunningTransition( page: Page, testInfo: { workerIndex: number }, @@ -1013,6 +1077,13 @@ test.describe("Agent message submission", () => { await expectInterruptedTurnOrderAfterReconnect(page, testInfo); }); + test("keeps a streaming hidden submission before its output after workspace eviction", async ({ + page, + }, testInfo) => { + test.setTimeout(120_000); + await expectHiddenStreamingSubmissionOrderAfterWorkspaceEviction(page, testInfo); + }); + test("clears an attachment-only submission when canonical history arrives after a missed running transition", async ({ page, }, testInfo) => { diff --git a/packages/app/e2e/browser/agent-timeline-pagination.spec.ts b/packages/app/e2e/browser/agent-timeline-pagination.spec.ts index 72e7491de..af6b0009c 100644 --- a/packages/app/e2e/browser/agent-timeline-pagination.spec.ts +++ b/packages/app/e2e/browser/agent-timeline-pagination.spec.ts @@ -26,9 +26,41 @@ import { expectTimelineViewportAnchoredAfterPrepend, userNavigatesTimelineToHistoryStartWithKeyboard, userScrollsTimelineToHistoryStart, + waitForPersistedCanonicalTimelineRange, } from "../support/helpers/timeline-pagination"; +import { trackAgentTimelineRequests } from "../support/helpers/agent-timeline-gate"; +import { seedMockAgentWorkspace } from "../support/helpers/mock-agent"; test.describe("Agent timeline pagination", () => { + test("resumes a persisted canonical timeline after its exact end", async ({ page }) => { + const prompt = "timeline persisted range resumes without tail replay"; + const agent = await seedMockAgentWorkspace({ + repoPrefix: "timeline-persisted-range-", + title: "Persisted timeline range", + initialPrompt: prompt, + }); + try { + await agent.client.waitForFinish(agent.agentId, 15_000); + await openAgentTimeline(page, agent); + await expectTimelinePromptVisible(page, prompt); + const range = await waitForPersistedCanonicalTimelineRange(page, agent.agentId); + + const tracker = await trackAgentTimelineRequests(page, agent.agentId); + await page.reload(); + await expectTimelinePromptVisible(page, prompt); + + const request = await tracker.nextRequest(); + expect(request).toEqual({ + direction: "after", + cursor: { epoch: range.epoch, seq: range.endSeq }, + }); + await tracker.waitForResponse(); + expect(tracker.requests().some((entry) => entry.direction === "tail")).toBe(false); + } finally { + await agent.cleanup(); + } + }); + test("keeps a fixed history-start gutter before, during, and after pagination", async ({ page, }) => { diff --git a/packages/app/e2e/browser/agent-timeline-resume.spec.ts b/packages/app/e2e/browser/agent-timeline-resume.spec.ts index 7691f6543..8a093feff 100644 --- a/packages/app/e2e/browser/agent-timeline-resume.spec.ts +++ b/packages/app/e2e/browser/agent-timeline-resume.spec.ts @@ -13,9 +13,10 @@ import { import { commitTimelineTurnsWhileDisconnected, disconnectViewedTimeline, - expectOneTailCheckWithoutCatchUpPages, + expectOneResumeCheckWithoutTail, + expectResumeOverflowFallsBackToOneTail, rememberTimelineRequestCounts, - restoreViewedTimelineWithHeldTail, + restoreViewedTimelineWithHeldResponse, } from "../support/helpers/timeline-resume"; test.describe("Agent timeline resume", () => { @@ -29,11 +30,11 @@ test.describe("Agent timeline resume", () => { const requests = rememberTimelineRequestCounts(gate); await disconnectViewedTimeline(page, gate); const background = await commitTimelineTurnsWhileDisconnected(agent, 24); - await restoreViewedTimelineWithHeldTail(page, gate); + await restoreViewedTimelineWithHeldResponse(page, gate); await expectTimelinePromptVisible(page, background.newestPrompt); await expectTimelinePromptNotMounted(page, background.firstPrompt); - expectOneTailCheckWithoutCatchUpPages(gate, requests); + expectResumeOverflowFallsBackToOneTail(gate, requests); await scrollTimelineUntilOlderHistoryIsReachable(page, background.firstPrompt); } finally { gate.restore(); @@ -53,10 +54,10 @@ test.describe("Agent timeline resume", () => { const requests = rememberTimelineRequestCounts(gate); await disconnectViewedTimeline(page, gate); - await restoreViewedTimelineWithHeldTail(page, gate); + await restoreViewedTimelineWithHeldResponse(page, gate); await expectTimelinePresentationUnchanged(page, presentation); - expectOneTailCheckWithoutCatchUpPages(gate, requests); + expectOneResumeCheckWithoutTail(gate, requests); } finally { gate.restore(); await agent.cleanup(); diff --git a/packages/app/e2e/support/helpers/agent-timeline-gate.ts b/packages/app/e2e/support/helpers/agent-timeline-gate.ts index ae5b883aa..eca355698 100644 --- a/packages/app/e2e/support/helpers/agent-timeline-gate.ts +++ b/packages/app/e2e/support/helpers/agent-timeline-gate.ts @@ -31,6 +31,82 @@ export interface PromptJumpRequestTracker { requests(): Array<{ cursorSeq: number | null; limit: number | null; mergeWindow: boolean }>; } +export interface TimelineRequestTracker { + nextRequest(): Promise<{ + direction: string | null; + cursor: { epoch: string; seq: number } | null; + }>; + requests(): Array<{ + direction: string | null; + cursor: { epoch: string; seq: number } | null; + }>; + waitForResponse(): Promise; +} + +export async function trackAgentTimelineRequests( + page: Page, + agentId: string, +): Promise { + type Request = ReturnType[number]; + const seen: Request[] = []; + const waiters: Array<(request: Request) => void> = []; + let responseSeen = false; + let resolveResponse: (() => void) | null = null; + const response = new Promise((resolve) => { + resolveResponse = resolve; + }); + await page.routeWebSocket(daemonWsRoutePattern(), (ws) => { + const server = ws.connectToServer(); + ws.onMessage((message) => { + const sessionMessage = getSessionMessage(message); + if ( + sessionMessage?.type === "fetch_agent_timeline_request" && + sessionMessage.agentId === agentId + ) { + const rawCursor = sessionMessage.cursor; + const cursor = + rawCursor && + typeof rawCursor === "object" && + typeof (rawCursor as { epoch?: unknown }).epoch === "string" && + typeof (rawCursor as { seq?: unknown }).seq === "number" + ? { + epoch: (rawCursor as { epoch: string }).epoch, + seq: (rawCursor as { seq: number }).seq, + } + : null; + const request = { + direction: typeof sessionMessage.direction === "string" ? sessionMessage.direction : null, + cursor, + }; + seen.push(request); + waiters.shift()?.(request); + } + server.send(message); + }); + server.onMessage((message) => { + const sessionMessage = getSessionMessage(message); + const payload = sessionMessage ? getPayload(sessionMessage) : null; + if ( + sessionMessage?.type === "fetch_agent_timeline_response" && + payload?.agentId === agentId + ) { + responseSeen = true; + resolveResponse?.(); + } + ws.send(message); + }); + }); + return { + nextRequest() { + const request = seen[0]; + if (request) return Promise.resolve(request); + return new Promise((resolve) => waiters.push(resolve)); + }, + requests: () => [...seen], + waitForResponse: () => (responseSeen ? Promise.resolve() : response), + }; +} + export async function trackPromptJumpRequests( page: Page, agentId: string, diff --git a/packages/app/e2e/support/helpers/replica-cache-storage.ts b/packages/app/e2e/support/helpers/replica-cache-storage.ts index 8d42be348..78df9b538 100644 --- a/packages/app/e2e/support/helpers/replica-cache-storage.ts +++ b/packages/app/e2e/support/helpers/replica-cache-storage.ts @@ -5,10 +5,16 @@ const STORE_NAME = "key-value"; const STORAGE_KEY = "@paseo:replica-cache"; export interface ReplicaCacheRecord { + version?: number; hosts?: Array<{ timeline?: { agentId?: string; items?: Array>; + range?: { + endSeq?: number; + epoch?: string; + startSeq?: number; + } | null; } | null; }>; } diff --git a/packages/app/e2e/support/helpers/timeline-pagination.ts b/packages/app/e2e/support/helpers/timeline-pagination.ts index f7c972295..cf02760e9 100644 --- a/packages/app/e2e/support/helpers/timeline-pagination.ts +++ b/packages/app/e2e/support/helpers/timeline-pagination.ts @@ -40,6 +40,12 @@ interface TimelineViewportSnapshot { scrollTop: number; } +interface PersistedCanonicalTimelineRange { + epoch: string; + startSeq: number; + endSeq: number; +} + export interface TimelinePresentationSnapshot { marker: string; position: TimelinePromptPositionSnapshot; @@ -298,6 +304,32 @@ export async function reloadAgentTimelineFromPersistedReplica( await expectTimelinePromptVisible(page, agent.newestPrompt); } +export async function waitForPersistedCanonicalTimelineRange( + page: Page, + agentId: string, +): Promise { + const readRange = async () => { + const cache = await readReplicaCache(page); + if (cache?.version !== 6) return null; + const range = cache.hosts?.find((host) => host.timeline?.agentId === agentId)?.timeline?.range; + if ( + typeof range?.epoch !== "string" || + typeof range.startSeq !== "number" || + typeof range.endSeq !== "number" + ) { + return null; + } + return { epoch: range.epoch, startSeq: range.startSeq, endSeq: range.endSeq }; + }; + + await expect.poll(readRange).not.toBeNull(); + const range = await readRange(); + if (!range) { + throw new Error(`Persisted canonical timeline range is missing for ${agentId}`); + } + return range; +} + export async function holdNextOlderTimelinePage( page: Page, agent: LongTimelineAgent, diff --git a/packages/app/e2e/support/helpers/timeline-resume.ts b/packages/app/e2e/support/helpers/timeline-resume.ts index c4a44713c..73c921776 100644 --- a/packages/app/e2e/support/helpers/timeline-resume.ts +++ b/packages/app/e2e/support/helpers/timeline-resume.ts @@ -40,12 +40,20 @@ export function rememberTimelineRequestCounts(gate: DaemonWebSocketGate): Timeli }; } -export function expectOneTailCheckWithoutCatchUpPages( +export function expectOneResumeCheckWithoutTail( gate: DaemonWebSocketGate, before: TimelineRequestCounts, ): void { + expect(gate.getTimelineRequestCount("after") - before.after).toBe(1); + expect(gate.getTimelineRequestCount("tail") - before.tail).toBe(0); +} + +export function expectResumeOverflowFallsBackToOneTail( + gate: DaemonWebSocketGate, + before: TimelineRequestCounts, +): void { + expect(gate.getTimelineRequestCount("after") - before.after).toBe(1); expect(gate.getTimelineRequestCount("tail") - before.tail).toBe(1); - expect(gate.getTimelineRequestCount("after") - before.after).toBe(0); } export async function disconnectViewedTimeline( @@ -56,7 +64,7 @@ export async function disconnectViewedTimeline( await expectReconnectingToastVisible(page); } -export async function restoreViewedTimelineWithHeldTail( +export async function restoreViewedTimelineWithHeldResponse( page: Page, gate: DaemonWebSocketGate, ): Promise { diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 2de5bc523..6cf80708e 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -19,7 +19,10 @@ import { type TimelineReducerSideEffect, } from "@/timeline/session-stream-reducers"; import { useCreateFlowStore } from "@/stores/create-flow-store"; -import { isTimelineResumeSnapshotAuthoritative } from "@/timeline/timeline-sync-plan"; +import { + isTimelineResumeSnapshotAuthoritative, + planTimelineTailFetch, +} from "@/timeline/timeline-sync-plan"; import { createViewedTimelineSync, type TimelineDeliveryMode, @@ -33,6 +36,7 @@ import { type AgentAttentionNotificationPayload, type NotificationPermissionRequest, } from "@getpaseo/protocol/agent-attention-notification"; + import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; import type { AgentSessionConfig } from "@getpaseo/protocol/agent-types"; import type { GitSetupOptions } from "@getpaseo/protocol/messages"; @@ -67,6 +71,20 @@ import { applyCheckoutStatusUpdateFromEvent } from "@/git/checkout-status-cache" import { useProviderSubagentStore } from "@/subagents/provider-store"; import { revalidateSessionAfterResume } from "@/contexts/session-resume-revalidation"; +type TimelineResponsePayload = Extract< + SessionOutboundMessage, + { type: "fetch_agent_timeline_response" } +>["payload"]; + +function consumeForcedTimelineTailReplacement( + payload: TimelineResponsePayload, + replacements: Set, +): TimelineResponsePayload { + if (payload.direction !== "tail") return payload; + if (!replacements.delete(payload.agentId)) return payload; + return { ...payload, reset: true }; +} + // Re-export types from session-store and draft-store for backward compatibility export type { DraftInput } from "@/stores/draft-store"; export type { @@ -350,6 +368,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const _sessionStateTimeoutRef = useRef | null>(null); const attentionNotifiedRef = useRef>(new Map()); const appStateRef = useRef(AppState.currentState); + const forcedTimelineTailReplacements = useRef(new Set()); const viewedTimelineSyncRef = useRef(null); const audioOutputBuffersRef = useRef>(new Map()); const activeAudioGroupsRef = useRef>(new Set()); @@ -528,12 +547,11 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider ); const applyTimelineResponse = useCallback( - ( - payload: Extract< - SessionOutboundMessage, - { type: "fetch_agent_timeline_response" } - >["payload"], - ) => { + (receivedPayload: TimelineResponsePayload) => { + const payload = consumeForcedTimelineTailReplacement( + receivedPayload, + forcedTimelineTailReplacements.current, + ); const agentId = payload.agentId; const initKey = getInitKey(serverId, agentId); const shouldMarkAuthoritativeHistoryApplied = isTimelineResumeSnapshotAuthoritative({ @@ -642,6 +660,15 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const sync = createViewedTimelineSync({ initialDeliveryMode, setSubscription: (agentIds) => client.setAgentTimelineSubscription(agentIds), + readCursor: (agentId) => { + const timeline = selectAgentTimelineState( + useSessionStore.getState().sessions[serverId], + agentId, + ); + return timeline.status === "synced" && timeline.range + ? { epoch: timeline.range.epoch, endSeq: timeline.range.endSeq } + : undefined; + }, fetchPage: async (agentId, request) => { const session = useSessionStore.getState().sessions[serverId]; const initKey = getInitKey(serverId, agentId); @@ -672,6 +699,18 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider throw error; } }, + fetchLatestTail: async (agentId) => { + forcedTimelineTailReplacements.current.add(agentId); + try { + return await getHostRuntimeStore().fetchAgentTimeline( + serverId, + agentId, + planTimelineTailFetch(), + ); + } finally { + forcedTimelineTailReplacements.current.delete(agentId); + } + }, reportError: (error) => { console.warn("[Session] viewed timeline synchronization failed", { serverId, error }); }, diff --git a/packages/app/src/hooks/use-agent-initialization.test.ts b/packages/app/src/hooks/use-agent-initialization.test.ts index a4a918de5..8786e3ac0 100644 --- a/packages/app/src/hooks/use-agent-initialization.test.ts +++ b/packages/app/src/hooks/use-agent-initialization.test.ts @@ -49,7 +49,7 @@ afterEach(() => { }); describe("ensureAgentIsInitialized", () => { - it("requests a bounded projected tail when authoritative history is loaded", () => { + it("catches up after loaded authoritative history", () => { const client = new FakeDaemonClient(); const runtime = new FakeTimelineRuntime(); useSessionStore.getState().initializeSession(serverId, client as never); @@ -74,13 +74,14 @@ describe("ensureAgentIsInitialized", () => { serverId, agentId, request: { - direction: "tail", + direction: "after", + cursor: { epoch: "epoch-1", seq: 42 }, limit: TIMELINE_FETCH_PAGE_SIZE, projection: "projected", }, }, ]); - expect(getInitDeferred(getInitKey(serverId, agentId))?.requestDirection).toBe("tail"); + expect(getInitDeferred(getInitKey(serverId, agentId))?.requestDirection).toBe("after"); }); it("requests a bounded projected tail when no authoritative cursor is available", () => { @@ -127,6 +128,8 @@ describe("ensureAgentIsInitialized", () => { timestamp: new Date("2026-07-27T10:00:00.000Z"), }, ], + range: null, + hasOlder: false, }, }); @@ -151,6 +154,52 @@ describe("ensureAgentIsInitialized", () => { ]); }); + it("catches up after the restored canonical replica range", () => { + const client = new FakeDaemonClient(); + const runtime = new FakeTimelineRuntime(); + useSessionStore.getState().restoreSessionReplica(serverId, { + agents: new Map(), + workspaces: new Map(), + projects: new Map(), + timeline: { + agentId, + items: [ + { + kind: "assistant_message", + id: "canonical-item", + text: "Canonical before restart", + timelineCursor: { epoch: "epoch-1", seq: 12 }, + timestamp: new Date("2026-07-27T10:00:00.000Z"), + }, + ], + range: { epoch: "epoch-1", startSeq: 10, endSeq: 12 }, + hasOlder: true, + }, + }); + + void ensureAgentIsInitialized({ + serverId, + agentId, + client: client as never, + runtime, + setAgentInitializing: bindSetAgentInitializing(), + }); + + expect(runtime.requests).toEqual([ + { + serverId, + agentId, + request: { + direction: "after", + cursor: { epoch: "epoch-1", seq: 12 }, + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "projected", + }, + }, + ]); + expect(getInitDeferred(getInitKey(serverId, agentId))?.requestDirection).toBe("after"); + }); + it("times out initialization after 65 seconds", async () => { vi.useFakeTimers(); const client = new FakeDaemonClient(); diff --git a/packages/app/src/hooks/use-agent-initialization.ts b/packages/app/src/hooks/use-agent-initialization.ts index c44621c0f..a19f36bf2 100644 --- a/packages/app/src/hooks/use-agent-initialization.ts +++ b/packages/app/src/hooks/use-agent-initialization.ts @@ -1,7 +1,7 @@ import { useCallback, useMemo } from "react"; import { useTranslation } from "react-i18next"; import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; -import { useSessionStore } from "@/stores/session-store"; +import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store"; import { createInitDeferred, getInitDeferred, @@ -11,7 +11,7 @@ import { refreshInitTimeout, } from "@/utils/agent-initialization"; import { getHostRuntimeStore, type HostRuntimeStore } from "@/runtime/host-runtime"; -import { planTimelineTailFetch } from "@/timeline/timeline-sync-plan"; +import { planTimelineResumeFetch, planTimelineTailFetch } from "@/timeline/timeline-sync-plan"; import { i18n } from "@/i18n/i18next"; export type SetAgentInitializing = (agentId: string, initializing: boolean) => void; @@ -51,7 +51,10 @@ export function ensureAgentIsInitialized(input: EnsureAgentIsInitializedInput): return existing.promise; } - const timelineRequest = planTimelineTailFetch(); + const timeline = selectAgentTimelineState(useSessionStore.getState().sessions[serverId], agentId); + const timelineRequest = planTimelineResumeFetch( + timeline.status === "synced" ? timeline.range : null, + ); const deferred = createInitDeferred(key, timelineRequest.direction); refreshAgentInitializationTimeout({ key, agentId, setAgentInitializing }); diff --git a/packages/app/src/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts index ee008f2ab..301d3e1c9 100644 --- a/packages/app/src/runtime/replica-cache/index.test.ts +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -111,6 +111,31 @@ function message(id: string, text: string): StreamItem { }; } +function toolCall(): StreamItem { + return { + kind: "tool_call", + id: "tool-1", + timelineCursor: { epoch: "epoch-1", seq: 12 }, + timestamp: new Date("2026-07-18T08:02:00.000Z"), + payload: { + source: "agent", + data: { + provider: "codex", + callId: "call-1", + name: "shell", + status: "completed", + error: null, + detail: { + type: "shell", + command: "npm test", + output: "passed", + exitCode: 0, + }, + }, + }, + }; +} + function seedSession(): void { const store = useSessionStore.getState(); store.initializeSession(SERVER_ID, null); @@ -209,7 +234,7 @@ describe("ReplicaCache", () => { cache.setHosts([]); }); - it("restores a displayable stale replica without claiming remote hydration", async () => { + it("restores the exact persisted canonical timeline window", async () => { const storage = new MemoryStorage(); const writer = new ReplicaCache(storage); writer.setHosts([SERVER_ID]); @@ -237,11 +262,65 @@ describe("ReplicaCache", () => { expect(session.workspaces.get("workspace-1")?.statusEnteredAt).toBeInstanceOf(Date); expect(session.workspaces.get("workspace-1")?.worktreeSlug).toBe("owned-worktree"); expect(session.agentStreamTail.get("agent-1")).toEqual([message("message-1", "Cached")]); - expect(session.agentAuthoritativeHistoryApplied).toEqual(new Map()); - expect(session.agentTimelineCursor).toEqual(new Map()); - expect(session.agentTimelineHasOlder).toEqual(new Map()); - expect(session.agentTimelineHasNewer).toEqual(new Map()); + expect(session.agentAuthoritativeHistoryApplied).toEqual(new Map([["agent-1", true]])); + expect(session.agentTimelineCursor).toEqual( + new Map([["agent-1", { epoch: "epoch-1", startSeq: 1, endSeq: 12 }]]), + ); + expect(session.agentTimelineHasOlder).toEqual(new Map([["agent-1", true]])); + expect(session.agentTimelineHasNewer).toEqual(new Map([["agent-1", false]])); expect(session.agentHistorySyncGeneration).toEqual(new Map()); + expect(selectAgentTimelineState(session, "agent-1")).toEqual({ + status: "synced", + items: [message("message-1", "Cached")], + range: { epoch: "epoch-1", startSeq: 1, endSeq: 12 }, + older: "available", + newer: "none", + }); + }); + + it("restores tool calls inside an authoritative cached window", async () => { + const storage = new MemoryStorage(); + const writer = new ReplicaCache(storage); + writer.setHosts([SERVER_ID]); + seedSession(); + useSessionStore.getState().setAgentStreamTail(SERVER_ID, new Map([["agent-1", [toolCall()]]])); + await writer.flush(); + + useSessionStore.getState().clearSession(SERVER_ID); + const reader = new ReplicaCache(storage); + reader.setHosts([SERVER_ID]); + await reader.restore(); + + const session = useSessionStore.getState().sessions[SERVER_ID]; + expect(session?.agentStreamTail.get("agent-1")).toEqual([toolCall()]); + expect(session?.agentTimelineCursor.get("agent-1")).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 12, + }); + }); + + it("restores display-only when retained items do not reach the stored range end", async () => { + const storage = new MemoryStorage(); + const writer = new ReplicaCache(storage); + writer.setHosts([SERVER_ID]); + seedSession(); + useSessionStore + .getState() + .setAgentTimelineCursor( + SERVER_ID, + new Map([["agent-1", { epoch: "epoch-1", startSeq: 1, endSeq: 13 }]]), + ); + await writer.flush(); + + useSessionStore.getState().clearSession(SERVER_ID); + const reader = new ReplicaCache(storage); + reader.setHosts([SERVER_ID]); + await reader.restore(); + + const session = useSessionStore.getState().sessions[SERVER_ID]; + expect(session?.agentStreamTail.get("agent-1")).toEqual([message("message-1", "Cached")]); + expect(session?.agentTimelineCursor).toEqual(new Map()); expect(selectAgentTimelineState(session, "agent-1")).toEqual({ status: "painted", items: [message("message-1", "Cached")], @@ -264,9 +343,10 @@ describe("ReplicaCache", () => { normalizeWorkspaceDescriptor(workspace("workspace-2", "project-2", "/repo/other")), ), ); - const secondTimeline = Array.from({ length: 60 }, (_, index) => - message(`message-${index}`, `Second ${index}`), - ); + const secondTimeline = Array.from({ length: 60 }, (_, index) => ({ + ...message(`message-${index}`, `Second ${index}`), + timelineCursor: { epoch: "epoch-2", seq: index + 1 }, + })); store.setAgentStreamTail( SERVER_ID, new Map([ @@ -274,6 +354,12 @@ describe("ReplicaCache", () => { ["agent-2", secondTimeline], ]), ); + store.setAgentTimelineCursor( + SERVER_ID, + new Map([["agent-2", { epoch: "epoch-2", startSeq: 1, endSeq: 60 }]]), + ); + store.setAgentTimelineHasOlder(SERVER_ID, new Map([["agent-2", true]])); + store.setAgentAuthoritativeHistoryApplied(SERVER_ID, "agent-2", true); store.setFocusedAgentId(SERVER_ID, "agent-2"); await cache.flush(); @@ -293,13 +379,23 @@ describe("ReplicaCache", () => { ]); expect(Array.from(timelines?.keys() ?? [])).toEqual(["agent-2"]); expect(timelines?.get("agent-2")).toEqual(secondTimeline.slice(-50)); + expect(session?.agentTimelineCursor.has("agent-2")).toBe(false); + expect(selectAgentTimelineState(session, "agent-2")).toEqual({ + status: "painted", + items: secondTimeline.slice(-50), + }); const persisted = JSON.parse(storage.values.get("@paseo:replica-cache") ?? "null") as { version: number; hosts: Array<{ timeline: Record | null }>; }; - expect(persisted.version).toBe(5); - expect(Object.keys(persisted.hosts[0]?.timeline ?? {}).sort()).toEqual(["agentId", "items"]); + expect(persisted.version).toBe(6); + expect(Object.keys(persisted.hosts[0]?.timeline ?? {}).sort()).toEqual([ + "agentId", + "hasOlder", + "items", + "range", + ]); }); it("persists reconciled rows without caching unreconciled local presentations", async () => { @@ -404,12 +500,12 @@ describe("ReplicaCache", () => { expect(Object.keys(useSessionStore.getState().sessions).sort()).toEqual(["host-a", "host-c"]); }); - it("rejects and clears version 4 cache data before overwriting it on flush", async () => { + it("rejects and clears version 5 cache data before overwriting it on flush", async () => { const storage = new MemoryStorage(); storage.values.set( "@paseo:replica-cache", JSON.stringify({ - version: 4, + version: 5, hosts: [ { serverId: SERVER_ID, @@ -435,7 +531,7 @@ describe("ReplicaCache", () => { expect(useSessionStore.getState().sessions[SERVER_ID]).toBeUndefined(); expect(JSON.parse(storage.values.get("@paseo:replica-cache") ?? "null")).toEqual({ - version: 5, + version: 6, hosts: [], }); }); diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index 35ddf09fe..961930fe2 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -1,6 +1,6 @@ import { Buffer } from "buffer"; import { z } from "zod"; -import { AgentStatusSchema } from "@getpaseo/protocol/messages"; +import { AgentStatusSchema, AgentTimelineItemPayloadSchema } from "@getpaseo/protocol/messages"; import { AgentProviderSchema } from "@getpaseo/protocol/provider-manifest"; import { normalizeProjectDescriptor, @@ -9,6 +9,7 @@ import { useSessionStore, type Agent, type SessionReplica, + type SessionReplicaTimeline, type SessionState, type ProjectDescriptor, type WorkspaceDescriptor, @@ -17,7 +18,7 @@ import { isUnreconciledLocalUserMessage, type StreamItem } from "@/types/stream" import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; const STORAGE_KEY = "@paseo:replica-cache"; -const CACHE_VERSION = 5; +const CACHE_VERSION = 6; const PERSIST_DELAY_MS = 750; const MAX_TIMELINE_ITEMS = 50; const MAX_CACHE_BYTES = 32 * 1024 * 1024; @@ -91,6 +92,12 @@ const StoredTimelineItemSchema = z.discriminatedUnion("kind", [ trigger: z.enum(["auto", "manual"]).optional(), preTokens: z.number().nonnegative().optional(), }), + z.strictObject({ + ...TimelineItemBaseShape, + kind: z.literal("tool_call"), + provider: AgentProviderSchema, + item: AgentTimelineItemPayloadSchema.refine((item) => item.type === "tool_call"), + }), ]); const AgentCapabilitiesSchema = z.strictObject({ @@ -246,6 +253,14 @@ const StoredProjectSchema = z.strictObject({ const StoredTimelineSchema = z.strictObject({ agentId: z.string(), items: z.array(StoredTimelineItemSchema), + range: z + .strictObject({ + epoch: z.string(), + startSeq: z.number().int().nonnegative(), + endSeq: z.number().int().nonnegative(), + }) + .nullable(), + hasOlder: z.boolean(), }); const StoredHostSchema = z.strictObject({ @@ -275,6 +290,8 @@ interface ReplicaInput { projects: ReadonlyMap; focusedAgent: Agent | undefined; timelineItems: StreamItem[] | undefined; + timelineRange: SessionReplicaTimeline["range"]; + timelineHasOlder: boolean; } export interface ReplicaCacheStorage { @@ -294,6 +311,8 @@ function deserializeTimeline(stored: StoredHost["timeline"]): SessionReplica["ti return { agentId: stored.agentId, items: stored.items.map(deserializeTimelineItem), + range: stored.range, + hasOlder: stored.hasOlder, }; } @@ -351,7 +370,23 @@ function serializeTimelineItem(item: StreamItem): StoredTimelineItem | null { ...(item.preTokens !== undefined ? { preTokens: item.preTokens } : {}), }; case "tool_call": - return null; + if (item.payload.source !== "agent") return null; + const storedTool = AgentTimelineItemPayloadSchema.safeParse({ + type: "tool_call", + callId: item.payload.data.callId, + name: item.payload.data.name, + status: item.payload.data.status, + error: item.payload.data.error, + detail: item.payload.data.detail, + ...(item.payload.data.metadata ? { metadata: item.payload.data.metadata } : {}), + }); + if (!storedTool.success || storedTool.data.type !== "tool_call") return null; + return { + ...base, + kind: item.kind, + provider: item.payload.data.provider, + item: storedTool.data, + }; } } @@ -404,6 +439,28 @@ function deserializeTimelineItem(item: StoredTimelineItem): StreamItem { ...(item.trigger ? { trigger: item.trigger } : {}), ...(item.preTokens !== undefined ? { preTokens: item.preTokens } : {}), }; + case "tool_call": { + const tool = item.item; + if (tool.type !== "tool_call") { + throw new Error("Stored tool call contains a non-tool timeline item"); + } + return { + ...base, + kind: item.kind, + payload: { + source: "agent", + data: { + provider: item.provider, + callId: tool.callId, + name: tool.name, + status: tool.status, + error: tool.error, + detail: tool.detail, + ...(tool.metadata ? { metadata: tool.metadata } : {}), + }, + }, + }; + } } } @@ -529,13 +586,28 @@ function serializeProject(project: ProjectDescriptor): StoredProject { }; } +function isTimelineItemStoredLosslessly(item: StreamItem): boolean { + switch (item.kind) { + case "user_message": + return (item.images?.length ?? 0) === 0 && (item.attachments?.length ?? 0) === 0; + case "activity_log": + return item.metadata === undefined; + case "tool_call": + return item.payload.source === "agent"; + default: + return true; + } +} + function replicaInputsEqual(left: ReplicaInput, right: ReplicaInput): boolean { return ( left.agents === right.agents && left.workspaces === right.workspaces && left.projects === right.projects && left.focusedAgent === right.focusedAgent && - left.timelineItems === right.timelineItems + left.timelineItems === right.timelineItems && + left.timelineRange === right.timelineRange && + left.timelineHasOlder === right.timelineHasOlder ); } @@ -550,14 +622,33 @@ function selectReplicaInput(session: SessionState, agentId: string | null): Repl projects: session.projects, focusedAgent: agent, timelineItems: timeline.status === "cold" ? undefined : timeline.items, + timelineRange: timeline.status === "synced" ? timeline.range : null, + timelineHasOlder: timeline.status === "synced" && timeline.older === "available", }; } function serializeHost(serverId: string, input: ReplicaInput, directorySync?: unknown): StoredHost { - const items = input.timelineItems - ?.filter((item) => item.kind !== "user_message" || !isUnreconciledLocalUserMessage(item)) - .map(serializeTimelineItem) - .filter((item) => item !== null); + const canonicalItems = input.timelineItems?.filter( + (item) => item.kind !== "user_message" || !isUnreconciledLocalUserMessage(item), + ); + const items = canonicalItems + ? canonicalItems.map(serializeTimelineItem).filter((item) => item !== null) + : undefined; + const range = input.timelineRange; + const canPersistCoverage = + range !== null && + range.retainedRanges === undefined && + canonicalItems !== undefined && + canonicalItems.length <= MAX_TIMELINE_ITEMS && + items?.length === canonicalItems.length && + canonicalItems.every( + (item) => + isTimelineItemStoredLosslessly(item) && + item.timelineCursor?.epoch === range.epoch && + item.timelineCursor.seq >= range.startSeq && + item.timelineCursor.seq <= range.endSeq, + ) && + canonicalItems.some((item) => item.timelineCursor?.seq === range.endSeq); return { serverId, agents: Array.from(input.agents.values(), serializeAgent), @@ -569,6 +660,10 @@ function serializeHost(serverId: string, input: ReplicaInput, directorySync?: un ? { agentId: input.focusedAgent.id, items: items.slice(-MAX_TIMELINE_ITEMS), + range: canPersistCoverage + ? { epoch: range.epoch, startSeq: range.startSeq, endSeq: range.endSeq } + : null, + hasOlder: canPersistCoverage ? input.timelineHasOlder : false, } : null, ...(directorySync ? { directorySync } : {}), @@ -783,8 +878,21 @@ export class ReplicaCache { if (session.focusedAgentId) { this.lastFocusedAgentIds.set(serverId, session.focusedAgentId); } - const input = selectReplicaInput(session, this.lastFocusedAgentIds.get(serverId) ?? null); + const focusedAgentId = this.lastFocusedAgentIds.get(serverId) ?? null; const previous = this.capturedInputs.get(serverId); + const selected = selectReplicaInput(session, focusedAgentId); + const hasTimelineHead = + focusedAgentId !== null && (session.agentStreamHead.get(focusedAgentId)?.length ?? 0) > 0; + let input = selected; + if (hasTimelineHead && previous && selected.timelineItems === previous.timelineItems) { + input = { + ...selected, + timelineRange: previous.timelineRange, + timelineHasOlder: previous.timelineHasOlder, + }; + } else if (hasTimelineHead) { + input = { ...selected, timelineRange: null, timelineHasOlder: false }; + } if (previous && replicaInputsEqual(previous, input)) return false; this.capturedInputs.set(serverId, input); diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 6045ce5f4..aeed4991a 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -308,6 +308,8 @@ export interface AgentTimelineCursorState { export interface SessionReplicaTimeline { agentId: string; items: StreamItem[]; + range: AgentTimelineCursorState | null; + hasOlder: boolean; } export interface SessionReplica { @@ -789,6 +791,16 @@ export const useSessionStore = create()( const tasks = latestTasksFromStream(timeline.items); if (tasks.length > 0) agentTasks.set(timeline.agentId, tasks); } + const agentTimelineCursor = new Map(); + const agentTimelineHasOlder = new Map(); + const agentTimelineHasNewer = new Map(); + const agentAuthoritativeHistoryApplied = new Map(); + if (timeline?.range) { + agentTimelineCursor.set(timeline.agentId, timeline.range); + agentTimelineHasOlder.set(timeline.agentId, timeline.hasOlder); + agentTimelineHasNewer.set(timeline.agentId, false); + agentAuthoritativeHistoryApplied.set(timeline.agentId, true); + } const agentLastActivity = new Map(prev.agentLastActivity); for (const agent of replica.agents.values()) { agentLastActivity.set(agent.id, agent.lastActivityAt); @@ -806,6 +818,10 @@ export const useSessionStore = create()( hasWorkspaceDirectorySnapshot: true, agentStreamTail, agentTasks, + agentTimelineCursor, + agentTimelineHasOlder, + agentTimelineHasNewer, + agentAuthoritativeHistoryApplied, }, }, agentLastActivity, diff --git a/packages/app/src/timeline/timeline-sync-plan.test.ts b/packages/app/src/timeline/timeline-sync-plan.test.ts index 99a987a48..9135b5fb8 100644 --- a/packages/app/src/timeline/timeline-sync-plan.test.ts +++ b/packages/app/src/timeline/timeline-sync-plan.test.ts @@ -5,6 +5,7 @@ import { isTimelineResumeSnapshotAuthoritative, planTimelineOlderFetch, planTimelinePromptJump, + planTimelineResumeFetch, planTimelineTailFetch, } from "./timeline-sync-plan"; @@ -19,6 +20,15 @@ describe("timeline sync planning", () => { }); }); + test("a restored canonical range catches up after its exact end", () => { + expect(planTimelineResumeFetch({ epoch: "epoch-1", endSeq: 49 })).toEqual({ + direction: "after", + cursor: { epoch: "epoch-1", seq: 49 }, + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "projected", + }); + }); + test("older history loads one bounded page before the start cursor", () => { const plan = planTimelineOlderFetch({ epoch: "epoch-1", seq: 25 }); diff --git a/packages/app/src/timeline/timeline-sync-plan.ts b/packages/app/src/timeline/timeline-sync-plan.ts index 800cbc007..e5e8b2fb9 100644 --- a/packages/app/src/timeline/timeline-sync-plan.ts +++ b/packages/app/src/timeline/timeline-sync-plan.ts @@ -51,6 +51,14 @@ export function planTimelineTailFetch() { } as const; } +export function planTimelineResumeFetch( + range: { epoch: string; endSeq: number } | null | undefined, +): ProjectedTimelineForwardFetchPlan { + return range + ? planTimelineCatchUpAfter({ epoch: range.epoch, seq: range.endSeq }) + : planTimelineTailFetch(); +} + export function planTimelineOlderFetch(cursor: TimelineSyncCursor) { return { direction: "before", diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index 55bfa8a00..a114ea2af 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -33,6 +33,7 @@ interface TimelineFetch { class TimelineWorld { readonly errors: string[] = []; + readonly cursors = new Map(); readonly sync = createViewedTimelineSync({ initialDeliveryMode: "selective", setSubscription: async (agentIds) => { @@ -45,6 +46,7 @@ class TimelineWorld { this.releaseMembershipWaiter(); return result.promise; }, + readCursor: (agentId) => this.cursors.get(agentId), fetchPage: async (agentId, request) => { const result = deferred<{ hasNewer: boolean; @@ -63,6 +65,12 @@ class TimelineWorld { this.releaseFetchWaiters(); return result.promise; }, + fetchLatestTail: (agentId) => + this.fetchTimeline(agentId, { + direction: "tail", + limit: 40, + projection: "projected", + }), reportError: (error) => { this.errors.push(error instanceof Error ? error.message : String(error)); const waiter = this.errorWaiters.shift(); @@ -94,6 +102,25 @@ class TimelineWorld { private readonly scheduled: Array<{ task: () => void; delayMs: number }> = []; private readonly retryWaiters: Array<(retry: () => void) => void> = []; + private fetchTimeline( + agentId: string, + request: ProjectedTimelineForwardFetchPlan, + ): Promise<{ hasNewer: boolean; endCursor: { epoch: string; seq: number } | null }> { + const result = deferred<{ + hasNewer: boolean; + endCursor: { epoch: string; seq: number } | null; + }>(); + this.fetches.push({ + agentId, + request, + respond: ({ hasNewer, seq = 1 }) => + result.resolve({ hasNewer, endCursor: { epoch: `epoch-${agentId}`, seq } }), + fail: (message) => result.reject(new Error(message)), + }); + this.releaseFetchWaiters(); + return result.promise; + } + nextMembership(): Promise { const request = this.memberships.shift(); if (request) return Promise.resolve(request); @@ -166,6 +193,41 @@ test("uses a tail fetch when an agent becomes visible", async () => { fetch.respond({ hasNewer: false }); }); +test("catches up after the restored cursor when an agent becomes visible", async () => { + const world = new TimelineWorld(); + world.cursors.set("agent-a", { epoch: "epoch-agent-a", endSeq: 42 }); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + const membership = await world.nextMembership(); + membership.succeed(); + + const fetch = await world.nextFetch("agent-a"); + expect(fetch.request).toEqual({ + direction: "after", + cursor: { epoch: "epoch-agent-a", seq: 42 }, + limit: 40, + projection: "projected", + }); + fetch.respond({ hasNewer: false }); +}); + +test("falls back to the latest tail when a restored cursor has more than one catch-up page", async () => { + const world = new TimelineWorld(); + world.cursors.set("agent-a", { epoch: "epoch-agent-a", endSeq: 42 }); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + const membership = await world.nextMembership(); + membership.succeed(); + + const probe = await world.nextFetch("agent-a"); + expect(probe.request.direction).toBe("after"); + probe.respond({ hasNewer: true, seq: 82 }); + + const fallback = await world.nextFetch("agent-a"); + expect(fallback.request).toEqual({ direction: "tail", limit: 40, projection: "projected" }); + fallback.respond({ hasNewer: false }); +}); + test("a gap absorbed by a running tail is recovered after the tail completes", async () => { const world = new TimelineWorld(); world.sync.setConnected(true); diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index dce143d15..4feed346b 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -1,6 +1,6 @@ import { planTimelineCatchUpAfter, - planTimelineTailFetch, + planTimelineResumeFetch, type ProjectedTimelineForwardFetchPlan, } from "./timeline-sync-plan"; @@ -12,10 +12,12 @@ interface TimelinePageResult { interface ViewedTimelineSyncPorts { initialDeliveryMode: TimelineDeliveryMode; setSubscription(agentIds: string[]): Promise; + readCursor(agentId: string): { epoch: string; endSeq: number } | undefined; fetchPage( agentId: string, request: ProjectedTimelineForwardFetchPlan, ): Promise; + fetchLatestTail(agentId: string): Promise; reportError(error: unknown): void; schedule(task: () => void, delayMs: number): () => void; } @@ -136,6 +138,12 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const isAcknowledged = (agentId: string) => acknowledged.includes(agentId); const isDesired = (agentId: string) => desired.includes(agentId); + const ownsCatchUp = (agentId: string, generation: number) => + !disposed && + connected && + isDesired(agentId) && + isAcknowledged(agentId) && + catchUps.get(agentId)?.generation === generation; const notifyListeners = () => { for (const listener of listeners) listener(); @@ -168,30 +176,26 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed agentId: string, generation: number, request: ProjectedTimelineForwardFetchPlan, + fallbackToLatestTailOnOverflow: boolean, ): Promise => { - if ( - disposed || - !connected || - !isDesired(agentId) || - !isAcknowledged(agentId) || - catchUps.get(agentId)?.generation !== generation - ) { - return; - } + if (!ownsCatchUp(agentId, generation)) return; try { const page = await ports.fetchPage(agentId, request); - if ( - disposed || - !connected || - !isDesired(agentId) || - !isAcknowledged(agentId) || - catchUps.get(agentId)?.generation !== generation - ) { - return; - } + if (!ownsCatchUp(agentId, generation)) return; if (page.hasNewer && page.endCursor) { - await fetchUntilCurrent(agentId, generation, planTimelineCatchUpAfter(page.endCursor)); + if (fallbackToLatestTailOnOverflow) { + await ports.fetchLatestTail(agentId); + catchUps.set(agentId, { generation, status: "complete" }); + setVisibilityCatchUpReady(agentId); + return; + } + await fetchUntilCurrent( + agentId, + generation, + planTimelineCatchUpAfter(page.endCursor), + false, + ); return; } if (page.hasNewer) { @@ -230,7 +234,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed if (request) pendingCatchUps.set(agentId, request); return; } - const nextRequest = request ?? planTimelineTailFetch(); + const nextRequest = request ?? planTimelineResumeFetch(ports.readCursor(agentId)); const current = catchUps.get(agentId); const decision = decideCatchUp({ current, request: nextRequest, supersede }); if (decision === "keep-and-park") { @@ -245,7 +249,12 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed catchUpGenerations.set(agentId, generation); catchUps.set(agentId, { generation, status: "running", request: nextRequest }); pendingCatchUps.delete(agentId); - void fetchUntilCurrent(agentId, generation, nextRequest); + void fetchUntilCurrent( + agentId, + generation, + nextRequest, + request === undefined && nextRequest.direction === "after", + ); }; const startAcknowledgedCatchUps = () => {