fix(sync): preserve selective delivery boundaries

Keep selective timeline capability app-owned, union viewed sets across shared sockets, and reconcile directory side effects from accepted state. Visibility and archive suppression now follow their existing authoritative boundaries.

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

@@ -4,6 +4,7 @@ import { AppState } from "react-native";
 import { useQueryClient } from "@tanstack/react-query";
 import { useTranslation } from "react-i18next";
 import { useClientActivity } from "@/hooks/use-client-activity";
+import { useAppVisible } from "@/hooks/use-app-visible";
 import { usePushTokenRegistration } from "@/hooks/use-push-token-registration";
 import { clearArchiveAgentPending } from "@/hooks/use-archive-agent";
 import {
@@ -524,6 +525,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
   const audioOutputBuffersRef = useRef<Map<string, BufferedAudioChunk[]>>(new Map());
   const activeAudioGroupsRef = useRef<Set<string>>(new Set());
   const workspaceHydrationRef = useRef<WorkspaceHydrationTransaction | null>(null);
+  const isAppVisible = useAppVisible();
 
   const applyWorkspaceUpdatePayload = useCallback(
     (payload: WorkspaceUpdatePayload) => {
@@ -555,7 +557,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
   useEffect(() => {
     const subscription = AppState.addEventListener("change", (nextState) => {
       appStateRef.current = nextState;
-      viewedTimelineSyncRef.current?.setActive(getIsAppActivelyVisible(nextState));
     });
 
     return () => {
@@ -563,6 +564,10 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
     };
   }, []);
 
+  useEffect(() => {
+    viewedTimelineSyncRef.current?.setActive(isAppVisible);
+  }, [isAppVisible]);
+
   const hydrateWorkspaces = useCallback(
     async (options?: { subscribe?: boolean; isCancelled?: () => boolean }) => {
       if (!client || !isConnected) {
@@ -609,7 +614,11 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
 
       setWorkspaces(
         serverId,
-        reconcileWorkspaceDirectory({ snapshot: snapshot.workspaces, deltas: snapshot.deltas }),
+        reconcileWorkspaceDirectory({
+          serverId,
+          snapshot: snapshot.workspaces,
+          deltas: snapshot.deltas,
+        }),
       );
       setEmptyProjects(serverId, snapshot.emptyProjects.values());
       setHasHydratedWorkspaces(serverId, true);

packages/app/src/contexts/workspace-directory-reconciliation.test.ts

@@ -1,8 +1,14 @@
 import { expect, it } from "vitest";
 import type { WorkspaceDescriptorPayload } from "@getpaseo/protocol/messages";
 import { normalizeWorkspaceDescriptor } from "@/stores/session-store";
+import {
+  clearWorkspaceArchivePending,
+  markWorkspaceArchivePending,
+} from "./session-workspace-upserts";
 import { reconcileWorkspaceDirectory } from "./workspace-directory-reconciliation";
 
+const SERVER_ID = "workspace-directory-reconciliation";
+
 function workspace(id: string, title: string): WorkspaceDescriptorPayload {
   return {
     id,
@@ -25,6 +31,7 @@ function workspace(id: string, title: string): WorkspaceDescriptorPayload {
 
 it("keeps workspace upserts and removals received during later pages", () => {
   const result = reconcileWorkspaceDirectory({
+    serverId: SERVER_ID,
     snapshot: new Map([
       ["updated", normalizeWorkspaceDescriptor(workspace("updated", "snapshot"))],
       ["removed", normalizeWorkspaceDescriptor(workspace("removed", "snapshot"))],
@@ -39,3 +46,18 @@ it("keeps workspace upserts and removals received during later pages", () => {
     ["updated", "live"],
   ]);
 });
+
+it("does not restore a locally archiving workspace from a buffered upsert", () => {
+  markWorkspaceArchivePending({ serverId: SERVER_ID, workspaceId: "archiving" });
+  try {
+    const result = reconcileWorkspaceDirectory({
+      serverId: SERVER_ID,
+      snapshot: new Map(),
+      deltas: [{ kind: "upsert", workspace: workspace("archiving", "live") }],
+    });
+
+    expect(result.has("archiving")).toBe(false);
+  } finally {
+    clearWorkspaceArchivePending({ serverId: SERVER_ID, workspaceId: "archiving" });
+  }
+});

packages/app/src/contexts/workspace-directory-reconciliation.ts

@@ -1,9 +1,11 @@
 import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
 import { normalizeWorkspaceDescriptor, type WorkspaceDescriptor } from "@/stores/session-store";
+import { shouldSuppressWorkspaceForLocalArchive } from "./session-workspace-upserts";
 
 type WorkspaceDelta = Extract<SessionOutboundMessage, { type: "workspace_update" }>["payload"];
 
 export function reconcileWorkspaceDirectory(input: {
+  serverId: string;
   snapshot: ReadonlyMap<string, WorkspaceDescriptor>;
   deltas: readonly WorkspaceDelta[];
 }): Map<string, WorkspaceDescriptor> {
@@ -13,7 +15,11 @@ export function reconcileWorkspaceDirectory(input: {
       workspaces.delete(delta.id);
     } else {
       const workspace = normalizeWorkspaceDescriptor(delta.workspace);
-      workspaces.set(workspace.id, workspace);
+      if (shouldSuppressWorkspaceForLocalArchive({ serverId: input.serverId, workspace })) {
+        workspaces.delete(workspace.id);
+      } else {
+        workspaces.set(workspace.id, workspace);
+      }
     }
   }
   return workspaces;

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

@@ -1761,6 +1761,12 @@ describe("HostRuntimeStore", () => {
       updatedAt: "2026-07-12T10:00:00.000Z",
       title: "snapshot",
     });
+    const recoveredAgent = makeFetchAgentsEntry({
+      id: "agent-recovered",
+      cwd: "/repo",
+      updatedAt: "2026-07-12T10:00:00.000Z",
+      title: "before refresh",
+    });
     fakeClient.fetchAgentsResponses.push(
       makeFetchAgentsPayload({
         entries: [pageOneAgent],
@@ -1781,9 +1787,16 @@ describe("HostRuntimeStore", () => {
         getClientId: async () => "cid_paged_delta",
       },
     });
-    useSessionStore
-      .getState()
-      .initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1);
+    const sessionStore = useSessionStore.getState();
+    sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1);
+    sessionStore.setAgents(
+      host.serverId,
+      new Map([[recoveredAgent.agent.id, replicaAgent(recoveredAgent.agent, host.serverId)]]),
+    );
+    sessionStore.setAgentTimelineCursor(
+      host.serverId,
+      new Map([[recoveredAgent.agent.id, { epoch: "epoch", startSeq: 10, endSeq: 20 }]]),
+    );
     store.syncHosts([host]);
     await fakeClient.waitForFetches(2);
 
@@ -1792,6 +1805,11 @@ describe("HostRuntimeStore", () => {
       agent: { ...pageOneAgent.agent, title: "live" },
       project: pageOneAgent.project,
     });
+    fakeClient.agentUpdate({
+      kind: "upsert",
+      agent: { ...recoveredAgent.agent, title: "recovered by delta" },
+      project: recoveredAgent.project,
+    });
     finishPageTwo(
       makeFetchAgentsPayload({
         entries: [
@@ -1812,7 +1830,13 @@ describe("HostRuntimeStore", () => {
     ).toEqual([
       ["agent-a", "live"],
       ["agent-b", null],
+      ["agent-recovered", "recovered by delta"],
     ]);
+    expect(
+      useSessionStore
+        .getState()
+        .sessions[host.serverId]?.agentTimelineCursor.get(recoveredAgent.agent.id),
+    ).toEqual({ epoch: "epoch", startSeq: 10, endSeq: 20 });
 
     const agentB = makeFetchAgentsEntry({
       id: "agent-b",

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

@@ -554,6 +554,10 @@ function createDefaultDeps(): HostRuntimeControllerDeps {
         },
       }
     : undefined;
+  const appCapabilities = {
+    [CLIENT_CAPS.selectiveAgentTimeline]: true,
+    ...browserAutomationCapabilities,
+  };
 
   return {
     createClient: ({ host, connection, clientId, runtimeGeneration }) => {
@@ -564,7 +568,7 @@ function createDefaultDeps(): HostRuntimeControllerDeps {
         clientType: "mobile" as const,
         appVersion: resolveAppVersion() ?? undefined,
         runtimeGeneration,
-        ...(browserAutomationCapabilities ? { capabilities: browserAutomationCapabilities } : {}),
+        capabilities: appCapabilities,
       };
       if (connection.type === "directSocket" || connection.type === "directPipe") {
         return new DaemonClient({
@@ -602,7 +606,7 @@ function createDefaultDeps(): HostRuntimeControllerDeps {
       connectToDaemon(connection, {
         ...(host.serverId ? { serverId: host.serverId } : {}),
         ...(timeoutMs !== undefined ? { timeoutMs } : {}),
-        ...(browserAutomationCapabilities ? { capabilities: browserAutomationCapabilities } : {}),
+        capabilities: appCapabilities,
       }),
     getClientId: () => getOrCreateClientId(),
     mountClientHandlers: ({ client, host }) => {
@@ -2327,6 +2331,7 @@ export class HostRuntimeStore {
     this.applyAgentDirectoryCommitSideEffects({
       serverId: input.serverId,
       previous,
+      committedEntries: reconciled.entries,
       directory: {
         entries: stampedSnapshot,
         deltas: transaction.deltas,
@@ -2379,6 +2384,7 @@ export class HostRuntimeStore {
     this.applyAgentDirectoryCommitSideEffects({
       serverId: input.serverId,
       previous,
+      committedEntries: reconciled.entries,
       directory,
       stoppedRunningAgentIds: reconciled.stoppedRunningAgentIds,
     });
@@ -2390,10 +2396,11 @@ export class HostRuntimeStore {
   private applyAgentDirectoryCommitSideEffects(input: {
     serverId: string;
     previous: ReadonlyMap<string, Agent>;
+    committedEntries: FetchAgentsEntry[];
     directory: AgentDirectoryFetchResult;
     stoppedRunningAgentIds: string[];
   }): void {
-    const snapshotAgentIds = new Set(input.directory.entries.map((entry) => entry.agent.id));
+    const snapshotAgentIds = new Set(input.committedEntries.map((entry) => entry.agent.id));
     for (const agentId of input.previous.keys()) {
       if (!snapshotAgentIds.has(agentId)) {
         applyAgentDirectoryDelta({ serverId: input.serverId, delta: { kind: "remove", agentId } });

packages/client/src/daemon-client.test.ts

@@ -186,6 +186,7 @@ test("does not infer browser automation capabilities from Electron runtime", asy
     })
     .parse(JSON.parse(assertStr(mock.sent[0])));
   expect(hello.capabilities[CLIENT_CAPS.browserHost]).toBeUndefined();
+  expect(hello.capabilities[CLIENT_CAPS.selectiveAgentTimeline]).toBeUndefined();
 });
 
 test("advertises consumer-provided browser automation capabilities", async () => {
@@ -647,7 +648,6 @@ test("advertises client capabilities in hello", async () => {
       custom_mode_icons: true,
       provider_subagents: true,
       reasoning_merge_enum: true,
-      selective_agent_timeline: true,
       terminal_reflowable_snapshot: true,
       browser_host: {
         supportedCommands: ["list_tabs"],

packages/client/src/daemon-client.ts

@@ -4857,7 +4857,6 @@ export class DaemonClient {
           protocolVersion: 1,
           capabilities: {
             [CLIENT_CAPS.customModeIcons]: true,
-            [CLIENT_CAPS.selectiveAgentTimeline]: true,
             [CLIENT_CAPS.reasoningMergeEnum]: true,
             [CLIENT_CAPS.terminalReflowableSnapshot]: true,
             [CLIENT_CAPS.providerSubagents]: true,

packages/server/src/server/agent/providers/opencode-agent.test.ts

@@ -3003,11 +3003,11 @@ describe("OpenCode provider subagent contract", () => {
         event: {
           type: "timeline",
           id: "ses_child_with_history",
-          item: {
+          item: expect.objectContaining({
             type: "assistant_message",
             text: "Persisted child result.",
             messageId: "msg_child_history",
-          },
+          }),
           timestamp: "1970-01-01T00:00:02.000Z",
         },
       }),
@@ -3081,11 +3081,11 @@ describe("OpenCode provider subagent contract", () => {
           event: expect.objectContaining({
             type: "timeline",
             id: "ses_child_hydrating",
-            item: {
+            item: expect.objectContaining({
               type: "assistant_message",
               text: "Hydration did not lose this.",
               messageId: "msg_child_hydrating",
-            },
+            }),
           }),
         }),
       ),

packages/server/src/server/session.test.ts

@@ -17,6 +17,7 @@ import { Session } from "./session.js";
 import { DownloadTokenStore } from "./file-download/token-store.js";
 import { StructuredAgentFallbackError } from "./agent/agent-response-loop.js";
 import type { StoredAgentRecord } from "./agent/agent-storage.js";
+import type { AgentManagerEvent } from "./agent/agent-manager.js";
 import type { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js";
 import type { SessionOptions } from "./session.js";
 import type { SessionInboundMessage, SessionOutboundMessage } from "./messages.js";
@@ -4184,6 +4185,91 @@ test("replaces a capable session's complete viewed timeline set", async () => {
   ]);
 });
 
+test("unions viewed timelines across socket sources and removes detached sources", async () => {
+  const messages: SessionOutboundMessage[] = [];
+  const agentEventListeners: Array<(event: AgentManagerEvent) => void> = [];
+  const session = createSessionForTest({
+    messages,
+    agentManager: {
+      subscribe: vi.fn((listener: (event: AgentManagerEvent) => void) => {
+        agentEventListeners.push(listener);
+        return () => {};
+      }),
+    },
+  });
+  session.updateClientCapabilities({ selective_agent_timeline: true });
+  const firstSocket = {};
+  const secondSocket = {};
+
+  await session.handleMessage(
+    {
+      type: "agent.timeline.set_subscription.request",
+      agentIds: ["agent-a"],
+      requestId: "timeline-subscription-a",
+    },
+    firstSocket,
+  );
+  await session.handleMessage(
+    {
+      type: "agent.timeline.set_subscription.request",
+      agentIds: ["agent-b"],
+      requestId: "timeline-subscription-b",
+    },
+    secondSocket,
+  );
+  messages.length = 0;
+
+  if (agentEventListeners.length === 0) throw new Error("Agent event listener was not installed");
+  const forward = (event: AgentManagerEvent) => {
+    for (const listener of agentEventListeners) listener(event);
+  };
+  forward({
+    type: "agent_stream",
+    agentId: "agent-a",
+    event: {
+      type: "timeline",
+      provider: "mock",
+      item: { type: "assistant_message", messageId: "message-a", text: "A" },
+    },
+  });
+  forward({
+    type: "agent_stream",
+    agentId: "agent-b",
+    event: {
+      type: "timeline",
+      provider: "mock",
+      item: { type: "assistant_message", messageId: "message-b", text: "B" },
+    },
+  });
+  expect(messages.filter((message) => message.type === "agent_stream")).toHaveLength(2);
+
+  session.clearAgentTimelineSubscription(firstSocket);
+  messages.length = 0;
+  forward({
+    type: "agent_stream",
+    agentId: "agent-a",
+    event: {
+      type: "timeline",
+      provider: "mock",
+      item: { type: "assistant_message", messageId: "message-a-2", text: "detached A" },
+    },
+  });
+  forward({
+    type: "agent_stream",
+    agentId: "agent-b",
+    event: {
+      type: "timeline",
+      provider: "mock",
+      item: { type: "assistant_message", messageId: "message-b-2", text: "retained B" },
+    },
+  });
+  expect(
+    messages.flatMap((message) =>
+      message.type === "agent_stream" ? [message.payload.agentId] : [],
+    ),
+  ).toEqual(["agent-b"]);
+});
+
 describe("agent config setters", () => {
   test("set_agent_mode_request: success emits accepted response carrying the notice", async () => {
     const messages: SessionOutboundMessage[] = [];

packages/server/src/server/session.ts

@@ -554,6 +554,8 @@ export class Session {
   private readonly pushTokenStore: PushTokenStore;
   private unsubscribeAgentEvents: (() => void) | null = null;
   private viewedTimelineAgentIds = new Set<string>();
+  private readonly viewedTimelineAgentIdsBySource = new Map<object, Set<string>>();
+  private readonly defaultTimelineSubscriptionSource = {};
   private unsubscribeTerminalWorkspaceContributionEvents: (() => void) | null = null;
   private readonly agentUpdates: AgentUpdatesService;
   private workspaceUpdatesSubscription: WorkspaceUpdatesSubscriptionState | null = null;
@@ -925,11 +927,32 @@ export class Session {
 
   updateClientCapabilities(capabilities: Record<string, unknown> | null): void {
     this.clientCapabilities = parseClientCapabilities(capabilities);
-    if (this.supports(CLIENT_CAPS.selectiveAgentTimeline)) {
+    if (!this.supports(CLIENT_CAPS.selectiveAgentTimeline)) {
+      this.viewedTimelineAgentIdsBySource.clear();
       this.viewedTimelineAgentIds.clear();
     }
   }
 
+  clearAgentTimelineSubscription(source: object): void {
+    if (!this.viewedTimelineAgentIdsBySource.delete(source)) return;
+    this.rebuildViewedTimelineAgentIds();
+  }
+
+  private replaceAgentTimelineSubscription(source: object | undefined, agentIds: string[]): void {
+    const subscriptionSource = source ?? this.defaultTimelineSubscriptionSource;
+    if (agentIds.length === 0) this.viewedTimelineAgentIdsBySource.delete(subscriptionSource);
+    else this.viewedTimelineAgentIdsBySource.set(subscriptionSource, new Set(agentIds));
+    this.rebuildViewedTimelineAgentIds();
+  }
+
+  private rebuildViewedTimelineAgentIds(): void {
+    const viewedAgentIds = new Set<string>();
+    for (const agentIds of this.viewedTimelineAgentIdsBySource.values()) {
+      for (const agentId of agentIds) viewedAgentIds.add(agentId);
+    }
+    this.viewedTimelineAgentIds = viewedAgentIds;
+  }
+
   supports(capability: ClientCapability): boolean {
     return this.clientCapabilities.has(capability);
   }
@@ -1376,7 +1399,7 @@ export class Session {
   /**
    * Main entry point for processing session messages
    */
-  public async handleMessage(msg: SessionInboundMessage): Promise<void> {
+  public async handleMessage(msg: SessionInboundMessage, source?: object): Promise<void> {
     this.inflightRequests++;
     if (this.inflightRequests > this.peakInflightRequests) {
       this.peakInflightRequests = this.inflightRequests;
@@ -1390,7 +1413,7 @@ export class Session {
         "agent.session.inbound",
       );
       try {
-        await this.dispatchInboundMessage(msg);
+        await this.dispatchInboundMessage(msg, source);
       } catch (error) {
         const err = error instanceof Error ? error : new Error(String(error));
         this.sessionLogger.error({ err }, "Error handling message");
@@ -1428,12 +1451,12 @@ export class Session {
     }
   }
 
-  private async dispatchInboundMessage(msg: SessionInboundMessage): Promise<void> {
+  private async dispatchInboundMessage(msg: SessionInboundMessage, source?: object): Promise<void> {
     const promise =
       this.dispatchVoiceAndControlMessage(msg) ??
       this.dispatchAgentRewindMessage(msg) ??
       this.dispatchAgentRelationshipMessage(msg) ??
-      this.dispatchAgentTimelineMessage(msg) ??
+      this.dispatchAgentTimelineMessage(msg, source) ??
       this.dispatchAgentLifecycleMessage(msg) ??
       this.dispatchAgentConfigMessage(msg) ??
       this.dispatchCheckoutMessage(msg) ??
@@ -1514,7 +1537,10 @@ export class Session {
     }
   }
 
-  private dispatchAgentTimelineMessage(msg: SessionInboundMessage): Promise<void> | undefined {
+  private dispatchAgentTimelineMessage(
+    msg: SessionInboundMessage,
+    source?: object,
+  ): Promise<void> | undefined {
     switch (msg.type) {
       case "fetch_agent_timeline_request":
         return this.handleFetchAgentTimelineRequest(msg);
@@ -1525,7 +1551,7 @@ export class Session {
       case "agent.timeline.set_subscription.request": {
         const agentIds = [...new Set(msg.agentIds)].sort();
         if (this.supports(CLIENT_CAPS.selectiveAgentTimeline)) {
-          this.viewedTimelineAgentIds = new Set(agentIds);
+          this.replaceAgentTimelineSubscription(source, agentIds);
         }
         this.emit({
           type: "agent.timeline.set_subscription.response",

packages/server/src/server/websocket-server.relay-reconnect.test.ts

@@ -52,6 +52,10 @@ const sessionMock = vi.hoisted(() => {
     handleMessage = vi.fn(async () => {});
     handleBinaryFrame = vi.fn((_frame: unknown) => {});
     supports = vi.fn((capability: string) => this.args.clientCapabilities?.[capability] === true);
+    updateClientCapabilities = vi.fn((capabilities: Record<string, unknown> | null) => {
+      this.args.clientCapabilities = capabilities;
+    });
+    clearAgentTimelineSubscription = vi.fn();
     getClientActivity = vi.fn(() => null);
     getSessionId = vi.fn(() => "mock-session-id");
     resetPeakInflight = vi.fn(() => {});

packages/server/src/server/websocket-server.ts

@@ -1380,6 +1380,7 @@ export class VoiceAssistantWebSocketServer {
 
     this.sessions.delete(ws);
     connection.sockets.delete(ws);
+    connection.session.clearAgentTimelineSubscription(ws);
     this.socketIdentities.delete(ws);
 
     if (connection.sockets.size === 0) {
@@ -1748,7 +1749,7 @@ export class VoiceAssistantWebSocketServer {
     }
 
     const startMs = performance.now();
-    await activeConnection.session.handleMessage(message.message);
+    await activeConnection.session.handleMessage(message.message, ws);
     const durationMs = performance.now() - startMs;
     this.recordRequestLatency(message.message.type, durationMs);