fix(app): preserve background sync state across races

packages/app/src/contexts/session-context.tsx

@@ -120,7 +120,6 @@ interface BufferedAudioChunk {
 interface WorkspaceHydrationSnapshot {
   workspaces: Map<string, WorkspaceDescriptor>;
   emptyProjects: Map<string, EmptyProjectDescriptor>;
-  deltas: readonly WorkspaceUpdatePayload[];
 }
 
 interface WorkspaceHydrationTransaction {
@@ -130,6 +129,10 @@ interface WorkspaceHydrationTransaction {
   deltas: WorkspaceUpdatePayload[];
 }
 
+function isWorkspaceHydrationCancelled(isCancelled?: () => boolean): boolean {
+  return isCancelled?.() ?? false;
+}
+
 type WorkspaceUpdatePayload = Extract<
   SessionOutboundMessage,
   { type: "workspace_update" }
@@ -181,7 +184,6 @@ async function fetchWorkspaceHydrationSnapshot(input: {
   return {
     workspaces: new Map(input.transaction.workspaces),
     emptyProjects,
-    deltas: [...input.transaction.deltas],
   };
 }
 
@@ -406,11 +408,6 @@ function finalizeTimelineApplication(input: {
   if (shouldMarkAuthoritativeHistoryApplied) {
     setAgentAuthoritativeHistoryApplied(serverId, agentId, true);
     useCreateFlowStore.getState().clearByAgent({ serverId, agentId });
-  }
-  if (result.initResolution === "resolve") {
-    resolveInitDeferred(initKey);
-  }
-  if (result.clearInitializing) {
     markAgentHistorySynchronized(serverId, agentId);
     const session = useSessionStore.getState().sessions[serverId];
     const agent = session?.agents.get(agentId) ?? session?.agentDetails.get(agentId);
@@ -418,6 +415,9 @@ function finalizeTimelineApplication(input: {
       getHostRuntimeStore().drainQueuedAgentMessage(serverId, agentId);
     }
   }
+  if (result.initResolution === "resolve") {
+    resolveInitDeferred(initKey);
+  }
 }
 
 function applyToolResultToMessages(
@@ -603,7 +603,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
         }
         throw error;
       }
-      if (!snapshot || options?.isCancelled?.()) {
+      if (!snapshot || isWorkspaceHydrationCancelled(options?.isCancelled)) {
         if (workspaceHydrationRef.current === transaction) workspaceHydrationRef.current = null;
         return;
       }
@@ -620,18 +620,25 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
         return;
       }
 
+      if (
+        workspaceHydrationRef.current !== transaction ||
+        isWorkspaceHydrationCancelled(options?.isCancelled)
+      ) {
+        return;
+      }
+      const deltas = [...transaction.deltas];
+      workspaceHydrationRef.current = null;
       setWorkspaces(
         serverId,
         reconcileWorkspaceDirectory({
           serverId,
           snapshot: snapshot.workspaces,
-          deltas: snapshot.deltas,
+          deltas,
         }),
       );
       setEmptyProjects(serverId, snapshot.emptyProjects.values());
       setHasHydratedWorkspaces(serverId, true);
-      for (const delta of snapshot.deltas) applyWorkspaceUpdatePayload(delta);
-      if (workspaceHydrationRef.current === transaction) workspaceHydrationRef.current = null;
+      for (const delta of deltas) applyWorkspaceUpdatePayload(delta);
     },
     [
       applyWorkspaceUpdatePayload,

packages/app/src/panels/agent-panel.tsx

@@ -13,6 +13,7 @@ import { useStoreWithEqualityFn } from "zustand/traditional";
 import { AgentStreamView, type AgentStreamViewHandle } from "@/agent-stream/view";
 import { ArchivedAgentCallout } from "@/components/archived-agent-callout";
 import { FileDropZone } from "@/components/file-drop/file-drop-zone";
+import { useRetainedPanelActive } from "@/components/retained-panel";
 import { Composer } from "@/composer";
 import { AgentModeControl } from "@/composer/agent-controls/mode-control";
 import { RewindComposerRestoreProvider } from "@/components/rewind/composer-restore";
@@ -701,6 +702,7 @@ function ChatAgentContent({
   onOpenWorkspaceFile?: (request: WorkspaceFileOpenRequest) => void;
 }) {
   const { t } = useTranslation();
+  const isPaneVisible = useRetainedPanelActive();
   const { api: toastApi, toast: toastState, dismiss: dismissToast } = useToastHost();
   const { isArchivingAgent } = useArchiveAgent();
   const streamViewRef = useRef<AgentStreamViewHandle>(null);
@@ -912,7 +914,7 @@ function ChatAgentContent({
       }
       return;
     }
-    if (!isPaneFocused || !isConnected || !hasSession) {
+    if (!isPaneVisible || !isConnected || !hasSession) {
       return;
     }
     if (
@@ -973,7 +975,7 @@ function ChatAgentContent({
     ensureAgentIsInitialized,
     hasSession,
     isConnected,
-    isPaneFocused,
+    isPaneVisible,
     missingAgentState.kind,
     serverId,
   ]);

packages/app/src/runtime/host-runtime.test.ts

@@ -34,6 +34,7 @@ class FakeDaemonClient {
     Awaited<ReturnType<DaemonClient["fetchAgents"]>> | ReturnType<DaemonClient["fetchAgents"]>
   > = [];
   public sentAgentMessages: Array<Parameters<DaemonClient["sendAgentMessage"]>> = [];
+  public sendAgentMessageFailures: Error[] = [];
   private agentUpdateListeners = new Set<
     (message: Extract<SessionOutboundMessage, { type: "agent_update" }>) => void
   >();
@@ -80,6 +81,8 @@ class FakeDaemonClient {
   async sendAgentMessage(...args: Parameters<DaemonClient["sendAgentMessage"]>): Promise<void> {
     this.sentAgentMessages.push(args);
     for (const waiter of this.sentMessageWaiters) waiter();
+    const failure = this.sendAgentMessageFailures.shift();
+    if (failure) throw failure;
   }
 
   async waitForSentMessages(count: number): Promise<void> {
@@ -2105,6 +2108,51 @@ describe("HostRuntimeStore", () => {
     useSessionStore.getState().clearSession(host.serverId);
   });
 
+  it("restores an automatically drained message when sending fails", async () => {
+    const host = makeHost({ serverId: "srv_failed_queue_drain" });
+    const fakeClient = new FakeDaemonClient();
+    fakeClient.sendAgentMessageFailures.push(new Error("connection lost"));
+    const store = new HostRuntimeStore({
+      deps: {
+        createClient: () => fakeClient as unknown as DaemonClient,
+        connectToDaemon: async () => ({
+          client: fakeClient as unknown as DaemonClient,
+          serverId: host.serverId,
+          hostname: null,
+        }),
+        getClientId: async () => "cid_failed_queue_drain",
+      },
+    });
+    const sessionStore = useSessionStore.getState();
+    sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1);
+    sessionStore.setQueuedMessages(
+      host.serverId,
+      new Map([
+        [
+          "agent",
+          [
+            { id: "first", text: "retry me", attachments: [] },
+            { id: "second", text: "keep me behind", attachments: [] },
+          ],
+        ],
+      ]),
+    );
+
+    store.drainQueuedAgentMessage(host.serverId, "agent");
+
+    await vi.waitFor(() => {
+      expect(fakeClient.sentAgentMessages).toHaveLength(1);
+      expect(
+        useSessionStore.getState().sessions[host.serverId]?.queuedMessages.get("agent"),
+      ).toEqual([
+        { id: "first", text: "retry me", attachments: [] },
+        { id: "second", text: "keep me behind", attachments: [] },
+      ]);
+    });
+
+    useSessionStore.getState().clearSession(host.serverId);
+  });
+
   it("applies buffered stale side effects from the accepted page agent", async () => {
     const host = makeHost({ serverId: "srv_buffered_stale_side_effects" });
     const fakeClient = new FakeDaemonClient();

packages/app/src/runtime/host-runtime.ts

@@ -61,6 +61,7 @@ import {
 } from "@/data/push-router";
 import { mountBrowserAutomationDaemonClientHandler } from "@/browser-automation/handler";
 import { schedulesQueryBaseKey } from "@/schedules/aggregated-schedules";
+import { sendQueuedComposerMessageNow } from "@/composer/actions";
 import { splitComposerAttachmentsForSubmit } from "@/composer/attachments/submit";
 import { encodeImages } from "@/utils/encode-images";
 
@@ -2126,31 +2127,38 @@ export class HostRuntimeStore {
     const store = useSessionStore.getState();
     const session = store.sessions[serverId];
     const queue = session?.queuedMessages.get(agentId);
-    if (!session?.client || !queue?.length || session.initializingAgents.get(agentId) === true) {
+    const client = session?.client;
+    if (!client || !queue?.length || session.initializingAgents.get(agentId) === true) {
       return;
     }
-    const [next, ...rest] = queue;
-    const wirePayload = splitComposerAttachmentsForSubmit(next.attachments);
-    store.setQueuedMessages(serverId, (current) => {
-      const updated = new Map(current);
-      updated.set(agentId, rest);
-      return updated;
-    });
-    void encodeImages(wirePayload.images)
-      .then((images) =>
-        session.client?.sendAgentMessage(agentId, next.text, {
+    const next = queue[0];
+    void sendQueuedComposerMessageNow({
+      agentId,
+      messageId: next.id,
+      queue: {
+        read: (queuedAgentId) =>
+          useSessionStore.getState().sessions[serverId]?.queuedMessages.get(queuedAgentId) ?? [],
+        write: (update) => useSessionStore.getState().setQueuedMessages(serverId, update),
+      },
+      submitMessage: async ({ text, attachments }) => {
+        const wirePayload = splitComposerAttachmentsForSubmit(attachments);
+        const images = await encodeImages(wirePayload.images);
+        await client.sendAgentMessage(agentId, text, {
           messageId: next.id,
           ...(images && images.length > 0 ? { images } : {}),
           attachments: wirePayload.attachments,
-        }),
-      )
-      .catch((error) => {
+        });
+      },
+    }).then((result) => {
+      if (result.status === "failed") {
         console.error("[HostRuntime] failed to drain queued agent message", {
           serverId,
           agentId,
-          error: toErrorMessage(error),
+          error: result.errorMessage,
         });
-      });
+      }
+      return result;
+    });
   }
 
   getSnapshot(serverId: string): HostRuntimeSnapshot | null {