From f9254c10664fa7ed2e69a9d0d66a3c653e013d00 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 12 Jul 2026 23:37:09 +0200 Subject: [PATCH] fix(sync): finish paged history before advancing --- packages/app/src/contexts/session-context.tsx | 20 +++++++++++++------ packages/app/src/panels/agent-panel.tsx | 6 +++++- packages/app/src/runtime/host-runtime.ts | 6 ++++-- .../src/timeline/viewed-timeline-sync.test.ts | 20 +++++++++++++++++++ .../app/src/timeline/viewed-timeline-sync.ts | 9 ++++++++- packages/server/src/server/session.ts | 7 +++++-- 6 files changed, 56 insertions(+), 12 deletions(-) diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index f749718a5..74e6d3366 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -412,6 +412,7 @@ function finalizeTimelineApplication(input: { } if (result.clearInitializing) { markAgentHistorySynchronized(serverId, agentId); + getHostRuntimeStore().drainQueuedAgentMessage(serverId, agentId); } } @@ -1028,14 +1029,17 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider setSubscription: (agentIds) => client.setAgentTimelineSubscription(agentIds), readCursor: (agentId) => useSessionStore.getState().sessions[serverId]?.agentTimelineCursor.get(agentId), + hasAuthoritativeHistory: (agentId) => + useSessionStore + .getState() + .sessions[serverId]?.agentAuthoritativeHistoryApplied.get(agentId) === true, fetchPage: async (agentId, request) => { const session = useSessionStore.getState().sessions[serverId]; const initKey = getInitKey(serverId, agentId); - if ( - session?.agentAuthoritativeHistoryApplied.get(agentId) !== true && - !getInitDeferred(initKey) - ) { - createInitDeferred(initKey, request.direction ?? "tail"); + if (session?.agentAuthoritativeHistoryApplied.get(agentId) !== true) { + if (!getInitDeferred(initKey)) { + createInitDeferred(initKey, request.direction ?? "tail"); + } refreshAgentInitializationTimeout({ key: initKey, agentId, @@ -1044,7 +1048,11 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider setAgentInitializing(agentId, true); } try { - return await client.fetchAgentTimeline(agentId, request); + const page = await client.fetchAgentTimeline(agentId, request); + if (getInitDeferred(initKey)) { + refreshAgentInitializationTimeout({ key: initKey, agentId, setAgentInitializing }); + } + return page; } catch (error) { setAgentInitializing(agentId, false); rejectInitDeferred(initKey, error instanceof Error ? error : new Error(String(error))); diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index 6752396c4..1eaaacccc 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -911,7 +911,11 @@ function ChatAgentContent({ if (!isConnected || !hasSession) { return; } - if (missingAgentState.kind === "resolving" || missingAgentState.kind === "not_found") { + if ( + missingAgentState.kind === "resolving" || + missingAgentState.kind === "not_found" || + missingAgentState.kind === "error" + ) { return; } diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 4b38b48dd..083fecb01 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -2122,11 +2122,13 @@ export class HostRuntimeStore { } } - private drainQueuedAgentMessage(serverId: string, agentId: string): void { + drainQueuedAgentMessage(serverId: string, agentId: string): void { const store = useSessionStore.getState(); const session = store.sessions[serverId]; const queue = session?.queuedMessages.get(agentId); - if (!session?.client || !queue?.length) return; + if (!session?.client || !queue?.length || session.initializingAgents.get(agentId) === true) { + return; + } const [next, ...rest] = queue; const wirePayload = splitComposerAttachmentsForSubmit(next.attachments); store.setQueuedMessages(serverId, (current) => { diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index af5aa8f23..367675710 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -45,6 +45,7 @@ class TimelineWorld { return result.promise; }, readCursor: (agentId) => this.cursors.get(agentId), + hasAuthoritativeHistory: (agentId) => this.authoritativeHistory.has(agentId), fetchPage: async (agentId, request) => { const result = deferred<{ hasNewer: boolean; @@ -87,12 +88,18 @@ class TimelineWorld { resolve(fetch: TimelineFetch): void; }> = []; private readonly cursors = new Map(); + private readonly authoritativeHistory = new Set(); private readonly errorWaiters: Array<(message: string) => void> = []; private readonly retries: Array<() => void> = []; private readonly retryWaiters: Array<(retry: () => void) => void> = []; setCursor(agentId: string, endSeq: number): void { this.cursors.set(agentId, { epoch: `epoch-${agentId}`, startSeq: 1, endSeq }); + this.authoritativeHistory.add(agentId); + } + + setLiveCursor(agentId: string, endSeq: number): void { + this.cursors.set(agentId, { epoch: `epoch-${agentId}`, startSeq: 1, endSeq }); } nextMembership(): Promise { @@ -145,6 +152,19 @@ class TimelineWorld { } } +test("uses a tail fetch when a live cursor is not authoritative", async () => { + const world = new TimelineWorld(); + world.setLiveCursor("agent-a", 9); + 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: "tail", limit: 100, projection: "projected" }); + fetch.respond({ hasNewer: false }); +}); + test("unchanged visible-set publication does not cancel paged catch-up", 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 58e4b5767..3a40069b5 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -1,5 +1,6 @@ import type { AgentTimelineCursorState } from "@/stores/session-store"; import { + planInitialAgentTimelineSync, planResumeTimelineSync, planTimelineCatchUpAfter, type ProjectedTimelineForwardFetchPlan, @@ -13,6 +14,7 @@ interface TimelinePageResult { interface ViewedTimelineSyncPorts { setSubscription(agentIds: string[]): Promise; readCursor(agentId: string): AgentTimelineCursorState | undefined; + hasAuthoritativeHistory(agentId: string): boolean; fetchPage( agentId: string, request: ProjectedTimelineForwardFetchPlan, @@ -144,7 +146,12 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed catchUpGenerations.set(agentId, generation); catchUps.set(agentId, { generation, status: "running" }); pendingGaps.delete(agentId); - const nextRequest = request ?? planResumeTimelineSync({ cursor: ports.readCursor(agentId) }); + const cursor = ports.readCursor(agentId); + const nextRequest = + request ?? + (ports.hasAuthoritativeHistory(agentId) + ? planResumeTimelineSync({ cursor }) + : planInitialAgentTimelineSync({ cursor, hasAuthoritativeHistory: false })); void fetchUntilCurrent(agentId, generation, nextRequest); }; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 7389fe0ff..41cc5a7d2 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -4472,7 +4472,7 @@ export class Session { "fetch_workspaces_response_ready", ); const snapshot = this.buildBootstrapSnapshot(payload.entries); - this.seedWorkspaceSubscriptionSnapshot(subscriptionId, payload.entries); + this.seedWorkspaceSubscriptionSnapshot(subscriptionId, request.filter, payload.entries); this.emit({ type: "fetch_workspaces_response", @@ -4535,10 +4535,13 @@ export class Session { private seedWorkspaceSubscriptionSnapshot( subscriptionId: string | null, + filter: FetchWorkspacesRequestFilter | undefined, entries: FetchWorkspacesResponseEntry[], ): void { const subscription = this.workspaceUpdatesSubscription; - if (!subscriptionId || subscription?.subscriptionId !== subscriptionId) return; + if (!subscription) return; + if (subscriptionId && subscription.subscriptionId !== subscriptionId) return; + if (!subscriptionId && !equal(subscription.filter, filter)) return; for (const entry of entries) { subscription.lastEmittedByWorkspaceId.set(entry.id, { kind: "upsert",