fix(sync): finish paged history before advancing

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)));

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;
     }
 

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) => {

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<string, { epoch: string; startSeq: number; endSeq: number }>();
+  private readonly authoritativeHistory = new Set<string>();
   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<MembershipRequest> {
@@ -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);

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<void>;
   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);
   };
 

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",