feat(sync): keep live data scoped and current

Only viewed chats receive live timeline rows, while directory state now uses one subscribed bootstrap followed by ordered deltas. Legacy clients and daemons retain their existing behavior through centralized compatibility gates.

packages/server/src/server/selective-timeline-delivery.e2e.test.ts

@@ -0,0 +1,273 @@
+import { afterEach, beforeEach, expect, test } from "vitest";
+import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities";
+import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
+import { DaemonClient } from "./test-utils/daemon-client.js";
+import { createTestPaseoDaemon, type TestPaseoDaemon } from "./test-utils/paseo-daemon.js";
+
+interface MessageWaiter {
+  predicate(message: SessionOutboundMessage): boolean;
+  resolve(message: SessionOutboundMessage): void;
+  reject(error: Error): void;
+  timeout: ReturnType<typeof setTimeout>;
+}
+
+class ConnectedClient {
+  readonly messages: SessionOutboundMessage[] = [];
+  private readonly waiters: MessageWaiter[] = [];
+  private readonly unsubscribe: () => void;
+
+  constructor(readonly client: DaemonClient) {
+    this.unsubscribe = client.subscribeRawMessages((message) => {
+      this.messages.push(message);
+      for (let waiterIndex = this.waiters.length - 1; waiterIndex >= 0; waiterIndex -= 1) {
+        const waiter = this.waiters[waiterIndex];
+        if (!waiter.predicate(message)) continue;
+        clearTimeout(waiter.timeout);
+        this.waiters.splice(waiterIndex, 1);
+        waiter.resolve(message);
+      }
+    });
+  }
+
+  clear(): void {
+    this.messages.length = 0;
+  }
+
+  next(
+    predicate: (message: SessionOutboundMessage) => boolean,
+    description: string,
+  ): Promise<SessionOutboundMessage> {
+    const existing = this.messages.find(predicate);
+    if (existing) return Promise.resolve(existing);
+    return new Promise((resolve, reject) => {
+      const timeout = setTimeout(() => {
+        const index = this.waiters.findIndex((waiter) => waiter.resolve === resolve);
+        if (index >= 0) this.waiters.splice(index, 1);
+        reject(new Error(`Timed out waiting for ${description}`));
+      }, 5_000);
+      this.waiters.push({ predicate, resolve, reject, timeout });
+    });
+  }
+
+  hasTimeline(agentId: string): boolean {
+    return this.messages.some(
+      (message) => message.type === "agent_stream" && message.payload.agentId === agentId,
+    );
+  }
+
+  async barrier(label: string): Promise<void> {
+    await this.client.ping({ requestId: `barrier-${label}` });
+  }
+
+  close(): void {
+    this.unsubscribe();
+    for (const waiter of this.waiters) {
+      clearTimeout(waiter.timeout);
+      waiter.reject(new Error("Client boundary closed"));
+    }
+    this.waiters.length = 0;
+  }
+}
+
+function isAgentStream(agentId: string) {
+  return (message: SessionOutboundMessage): boolean =>
+    message.type === "agent_stream" && message.payload.agentId === agentId;
+}
+
+function isDedicatedAttention(agentId: string) {
+  return (message: SessionOutboundMessage): boolean =>
+    message.type === "agent_attention_required" && message.payload.agentId === agentId;
+}
+
+function isLegacyAttention(agentId: string) {
+  return (message: SessionOutboundMessage): boolean =>
+    message.type === "agent_stream" &&
+    message.payload.agentId === agentId &&
+    message.payload.event.type === "attention_required";
+}
+
+function dedicatedAttentionResult(message: SessionOutboundMessage, timelineLeaked: boolean) {
+  if (message.type !== "agent_attention_required") {
+    throw new Error(`Expected agent_attention_required, received ${message.type}`);
+  }
+  return {
+    type: message.type,
+    shouldNotify: message.payload.shouldNotify,
+    timelineLeaked,
+  };
+}
+
+function legacyAttentionResult(message: SessionOutboundMessage) {
+  if (message.type !== "agent_stream" || message.payload.event.type !== "attention_required") {
+    throw new Error(`Expected legacy attention_required agent_stream, received ${message.type}`);
+  }
+  return {
+    type: message.type,
+    eventType: message.payload.event.type,
+    agentId: message.payload.agentId,
+  };
+}
+
+let daemon: TestPaseoDaemon;
+const clients: ConnectedClient[] = [];
+
+beforeEach(async () => {
+  daemon = await createTestPaseoDaemon();
+});
+
+afterEach(async () => {
+  for (const connected of clients) {
+    connected.close();
+    await connected.client.close().catch(() => undefined);
+  }
+  clients.length = 0;
+  await daemon.close();
+}, 30_000);
+
+async function connect(input: { clientId: string; selective: boolean }): Promise<ConnectedClient> {
+  const client = new DaemonClient({
+    url: `ws://127.0.0.1:${daemon.port}/ws`,
+    clientId: input.clientId,
+    capabilities: { [CLIENT_CAPS.selectiveAgentTimeline]: input.selective },
+    reconnect: { enabled: false },
+  });
+  await client.connect();
+  const connected = new ConnectedClient(client);
+  clients.push(connected);
+  return connected;
+}
+
+test("real WebSocket sessions enforce selective delivery, retained resets, downgrade, and dedicated attention", async () => {
+  const legacy = await connect({ clientId: "legacy-client", selective: false });
+  let capable = await connect({ clientId: "capable-client", selective: true });
+  const agents = await Promise.all(
+    ["A", "B", "C"].map((title) =>
+      legacy.client.createAgent({
+        provider: "codex",
+        cwd: "/tmp",
+        title: `Selective ${title}`,
+        modeId: "full-access",
+      }),
+    ),
+  );
+  const [agentA, agentB, agentC] = agents;
+  legacy.clear();
+  capable.clear();
+
+  await daemon.daemon.agentManager.emitLiveTimelineItem(agentC.id, {
+    type: "assistant_message",
+    text: "before membership",
+  });
+  await legacy.next(isAgentStream(agentC.id), "legacy global delivery before membership");
+  await capable.barrier("before-membership");
+  expect(capable.hasTimeline(agentC.id)).toBe(false);
+
+  await capable.client.setAgentTimelineSubscription([agentA.id, agentB.id]);
+  legacy.clear();
+  capable.clear();
+  await daemon.daemon.agentManager.emitLiveTimelineItem(agentA.id, {
+    type: "assistant_message",
+    text: "viewed A",
+  });
+  await daemon.daemon.agentManager.emitLiveTimelineItem(agentB.id, {
+    type: "assistant_message",
+    text: "viewed B",
+  });
+  await daemon.daemon.agentManager.emitLiveTimelineItem(agentC.id, {
+    type: "assistant_message",
+    text: "unviewed C",
+  });
+  await Promise.all([
+    capable.next(isAgentStream(agentA.id), "capable A delivery"),
+    capable.next(isAgentStream(agentB.id), "capable B delivery"),
+    legacy.next(isAgentStream(agentC.id), "legacy C delivery"),
+  ]);
+  await capable.barrier("unviewed-c");
+  expect(capable.hasTimeline(agentC.id)).toBe(false);
+
+  await capable.client.setAgentTimelineSubscription([agentB.id]);
+  legacy.clear();
+  capable.clear();
+  await daemon.daemon.agentManager.emitLiveTimelineItem(agentA.id, {
+    type: "assistant_message",
+    text: "removed A",
+  });
+  await daemon.daemon.agentManager.emitLiveTimelineItem(agentB.id, {
+    type: "assistant_message",
+    text: "retained B",
+  });
+  await Promise.all([
+    legacy.next(isAgentStream(agentA.id), "legacy removed A delivery"),
+    capable.next(isAgentStream(agentB.id), "capable retained B delivery"),
+  ]);
+  await capable.barrier("removed-a");
+  expect(capable.hasTimeline(agentA.id)).toBe(false);
+
+  capable.close();
+  await capable.client.close();
+  clients.splice(clients.indexOf(capable), 1);
+  capable = await connect({ clientId: "capable-client", selective: true });
+  legacy.clear();
+  capable.clear();
+  await daemon.daemon.agentManager.emitLiveTimelineItem(agentB.id, {
+    type: "assistant_message",
+    text: "after capable resume",
+  });
+  await legacy.next(isAgentStream(agentB.id), "legacy delivery after capable resume");
+  await capable.barrier("resumed-membership-reset");
+  expect(capable.hasTimeline(agentB.id)).toBe(false);
+
+  capable.client.sendHeartbeat({
+    deviceType: "mobile",
+    focusedAgentId: null,
+    lastActivityAt: new Date().toISOString(),
+    appVisible: true,
+  });
+  legacy.clear();
+  capable.clear();
+  const attention = capable.next(
+    isDedicatedAttention(agentC.id),
+    "capable dedicated attention notification",
+  );
+  const legacyAttention = legacy.next(
+    isLegacyAttention(agentC.id),
+    "legacy attention stream notification",
+  );
+  await legacy.client.sendMessage(agentC.id, "finish attention boundary test");
+  const [attentionMessage, legacyAttentionMessage] = await Promise.all([
+    attention,
+    legacyAttention,
+  ]);
+  await capable.barrier("attention-delivery");
+  expect({
+    capable: dedicatedAttentionResult(attentionMessage, capable.hasTimeline(agentC.id)),
+    legacy: legacyAttentionResult(legacyAttentionMessage),
+  }).toEqual({
+    capable: {
+      type: "agent_attention_required",
+      shouldNotify: true,
+      timelineLeaked: false,
+    },
+    legacy: {
+      type: "agent_stream",
+      eventType: "attention_required",
+      agentId: agentC.id,
+    },
+  });
+
+  capable.close();
+  await capable.client.close();
+  clients.splice(clients.indexOf(capable), 1);
+  const downgraded = await connect({ clientId: "capable-client", selective: false });
+  downgraded.clear();
+  await daemon.daemon.agentManager.emitLiveTimelineItem(agentA.id, {
+    type: "assistant_message",
+    text: "after downgrade",
+  });
+  const downgradedDelivery = await downgraded.next(
+    isAgentStream(agentA.id),
+    "legacy global delivery after capability downgrade",
+  );
+
+  expect(downgradedDelivery.type).toBe("agent_stream");
+}, 30_000);

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

@@ -4554,6 +4554,28 @@ describe("chat/schedule/loop dispatch routing (behavior preservation)", () => {
   });
 });
 
+test("replaces a capable session's complete viewed timeline set", async () => {
+  const messages: SessionOutboundMessage[] = [];
+  const session = createSessionForTest({ messages });
+  session.updateClientCapabilities({ selective_agent_timeline: true });
+
+  await session.handleMessage({
+    type: "agent.timeline.set_subscription.request",
+    agentIds: ["agent-b", "agent-a", "agent-a"],
+    requestId: "timeline-subscription-1",
+  });
+
+  expect(messages).toEqual([
+    {
+      type: "agent.timeline.set_subscription.response",
+      payload: {
+        agentIds: ["agent-a", "agent-b"],
+        requestId: "timeline-subscription-1",
+      },
+    },
+  ]);
+});
+
 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

@@ -583,6 +583,7 @@ export class Session {
   private readonly daemonConfigStore: DaemonConfigStore;
   private readonly pushTokenStore: PushTokenStore;
   private unsubscribeAgentEvents: (() => void) | null = null;
+  private viewedTimelineAgentIds = new Set<string>();
   private unsubscribeTerminalWorkspaceContributionEvents: (() => void) | null = null;
   private readonly agentUpdates: AgentUpdatesService;
   private workspaceUpdatesSubscription: WorkspaceUpdatesSubscriptionState | null = null;
@@ -965,6 +966,9 @@ export class Session {
 
   updateClientCapabilities(capabilities: Record<string, unknown> | null): void {
     this.clientCapabilities = parseClientCapabilities(capabilities);
+    if (this.supports(CLIENT_CAPS.selectiveAgentTimeline)) {
+      this.viewedTimelineAgentIds.clear();
+    }
   }
 
   supports(capability: ClientCapability): boolean {
@@ -979,6 +983,16 @@ export class Session {
     await this.emitWorkspaceUpdatesForWorkspaceIds([workspaceId], { skipReconcile: true });
   }
 
+  private async emitCreatedWorkspaceUpdate(workspace: WorkspaceDescriptorPayload): Promise<void> {
+    if (this.workspaceUpdatesSubscription) {
+      await this.emitWorkspaceUpdateForWorkspaceId(workspace.id);
+      return;
+    }
+    // COMPAT(workspaceCreateCausalUpdate): added in v0.1.106, remove after 2027-01-12.
+    // Older clients create before subscribing and require the causal update beside the response.
+    this.emit({ type: "workspace_update", payload: { kind: "upsert", workspace } });
+  }
+
   async archiveWorkspaceRecordForExternalMutation(workspaceId: string): Promise<void> {
     await this.archiveWorkspaceRecord(workspaceId);
   }
@@ -1271,10 +1285,31 @@ export class Session {
           "agent.session.forward_stream",
         );
 
-        this.emit({
-          type: "agent_stream",
-          payload: this.buildAgentStreamPayload(event, serializedEvent),
-        });
+        if (
+          this.supports(CLIENT_CAPS.selectiveAgentTimeline) &&
+          serializedEvent.type === "attention_required"
+        ) {
+          this.emit({
+            type: "agent_attention_required",
+            payload: {
+              agentId: event.agentId,
+              reason: serializedEvent.reason,
+              timestamp: serializedEvent.timestamp,
+              shouldNotify: serializedEvent.shouldNotify,
+              ...(serializedEvent.notification
+                ? { notification: serializedEvent.notification }
+                : {}),
+            },
+          });
+        } else if (
+          !this.supports(CLIENT_CAPS.selectiveAgentTimeline) ||
+          this.viewedTimelineAgentIds.has(event.agentId)
+        ) {
+          this.emit({
+            type: "agent_stream",
+            payload: this.buildAgentStreamPayload(event, serializedEvent),
+          });
+        }
 
         if (event.event.type === "permission_requested") {
           this.emit({
@@ -1524,6 +1559,17 @@ export class Session {
         return this.handleProviderSubagentListRequest(msg);
       case "agent.provider_subagents.timeline.get.request":
         return this.handleProviderSubagentTimelineRequest(msg);
+      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.emit({
+          type: "agent.timeline.set_subscription.response",
+          payload: { agentIds, requestId: msg.requestId },
+        });
+        return undefined;
+      }
       case "agent.fork_context.request":
         return this.handleAgentForkContextRequest(msg);
       default:
@@ -4259,6 +4305,9 @@ export class Session {
       this.workspaceGitObserver.recordDescriptorState(workspaceId, nextWorkspace);
 
       if (!nextWorkspace) {
+        if (workspace && !subscription.lastEmittedByWorkspaceId.has(workspaceId)) {
+          continue;
+        }
         subscription.lastEmittedByWorkspaceId.delete(workspaceId);
         this.bufferOrEmitWorkspaceUpdate(
           subscription,
@@ -4661,7 +4710,7 @@ export class Session {
         error: null,
       },
     });
-    await this.emitWorkspaceUpdateForWorkspaceId(workspace.workspaceId);
+    await this.emitCreatedWorkspaceUpdate(descriptor);
     void this.workspaceGitService
       .getSnapshot(workspace.cwd, { force: true, includeForge: true, reason: "open_project" })
       .catch((error) => {
@@ -4746,10 +4795,7 @@ export class Session {
         error: null,
       },
     });
-    this.emit({
-      type: "workspace_update",
-      payload: { kind: "upsert", workspace: descriptor },
-    });
+    await this.emitCreatedWorkspaceUpdate(descriptor);
   }
 
   private async resolveWorktreeSourceCwd(input: {

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

@@ -7827,4 +7827,75 @@ test("workspace.create.response persists the first prompt as the initial title",
   expect(workspaceId).toBeDefined();
   const persisted = await session.workspaceRegistry.get(workspaceId as string);
   expect(persisted?.title).toBe("Add retries to the payments flow");
+  expect(filterByType(emitted, "workspace_update")).toHaveLength(1);
+});
+
+test("workspace create emits through a matching workspace subscription", async () => {
+  const emitted: SessionOutboundMessage[] = [];
+  const workspaces = new Map<string, ReturnType<typeof createPersistedWorkspaceRecord>>();
+  const session = createSessionForWorkspaceTests({
+    onMessage: (message) => emitted.push(message),
+    workspaceRegistry: {
+      initialize: async () => {},
+      existsOnDisk: async () => true,
+      list: async () => Array.from(workspaces.values()),
+      get: async (workspaceId) => workspaces.get(workspaceId) ?? null,
+      upsert: async (workspace) => {
+        workspaces.set(workspace.workspaceId, workspace);
+      },
+      archive: async () => {},
+      remove: async () => {},
+    },
+  });
+  session.listAgentPayloads = async () => [];
+
+  await session.handleMessage({
+    type: "fetch_workspaces_request",
+    requestId: "req-subscribe-create-match",
+    filter: { query: "repo" },
+    subscribe: { subscriptionId: "sub-create-match" },
+  });
+  emitted.length = 0;
+  await session.handleMessage({
+    type: "workspace.create.request",
+    requestId: "req-create-match",
+    source: { kind: "directory", path: REPO_CWD },
+  });
+
+  expect(filterByType(emitted, "workspace_update")).toHaveLength(1);
+});
+
+test("workspace create stays out of a non-matching workspace subscription", async () => {
+  const emitted: SessionOutboundMessage[] = [];
+  const workspaces = new Map<string, ReturnType<typeof createPersistedWorkspaceRecord>>();
+  const session = createSessionForWorkspaceTests({
+    onMessage: (message) => emitted.push(message),
+    workspaceRegistry: {
+      initialize: async () => {},
+      existsOnDisk: async () => true,
+      list: async () => Array.from(workspaces.values()),
+      get: async (workspaceId) => workspaces.get(workspaceId) ?? null,
+      upsert: async (workspace) => {
+        workspaces.set(workspace.workspaceId, workspace);
+      },
+      archive: async () => {},
+      remove: async () => {},
+    },
+  });
+  session.listAgentPayloads = async () => [];
+
+  await session.handleMessage({
+    type: "fetch_workspaces_request",
+    requestId: "req-subscribe-create-filtered",
+    filter: { query: "definitely-not-this-workspace" },
+    subscribe: { subscriptionId: "sub-create-filtered" },
+  });
+  emitted.length = 0;
+  await session.handleMessage({
+    type: "workspace.create.request",
+    requestId: "req-create-filtered",
+    source: { kind: "directory", path: REPO_CWD },
+  });
+
+  expect(filterByType(emitted, "workspace_update")).toEqual([]);
 });

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

@@ -173,6 +173,7 @@ function createSessionWithActivity(
 ) {
   return {
     getClientActivity: vi.fn(() => activity),
+    supports: () => false,
   };
 }
 

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

@@ -1163,12 +1163,15 @@ export class VoiceAssistantWebSocketServer {
         existing.session.updateAppVersion(newAppVersion);
       }
       const newClientCapabilities = message.capabilities ?? null;
+      // COMPAT(selectiveAgentTimeline): added in v0.1.106. Every capable resumed
+      // hello resets membership before server_info so stale retained-session
+      // state cannot leak. Remove after 2027-01-12.
+      existing.session.updateClientCapabilities(newClientCapabilities);
       if (
         JSON.stringify(existing.clientCapabilities ?? null) !==
         JSON.stringify(newClientCapabilities ?? null)
       ) {
         existing.clientCapabilities = newClientCapabilities;
-        existing.session.updateClientCapabilities(newClientCapabilities);
         this.syncBrowserToolsClientRegistration(existing);
       }
       existing.sockets.add(ws);
@@ -1281,6 +1284,8 @@ export class VoiceAssistantWebSocketServer {
         importSessionWorkspaceTarget: true,
         // COMPAT(forgeProviders): added in v0.1.106, drop the gate when daemon floor >= v0.1.106.
         forgeProviders: true,
+        // COMPAT(selectiveAgentTimeline): added in v0.1.106, remove after 2027-01-12.
+        selectiveAgentTimeline: true,
       },
     };
   }
@@ -2009,21 +2014,36 @@ export class VoiceAssistantWebSocketServer {
     for (const [clientIndex, { ws }] of clientEntries.entries()) {
       const shouldNotify = clientIndex === plan.inAppRecipientIndex;
       const timestamp = new Date().toISOString();
-      const message = wrapSessionMessage({
-        type: "agent_stream",
-        payload: {
-          agentId: params.agentId,
-          event: {
-            type: "attention_required",
-            provider: params.provider,
-            reason: params.reason,
-            timestamp,
-            shouldNotify,
-            notification,
-          },
-          timestamp,
-        },
-      });
+      const connection = this.sessions.get(ws);
+      const attentionPayload = {
+        agentId: params.agentId,
+        reason: params.reason,
+        timestamp,
+        shouldNotify,
+        notification,
+      };
+      const message = wrapSessionMessage(
+        connection?.session.supports(CLIENT_CAPS.selectiveAgentTimeline)
+          ? {
+              type: "agent_attention_required",
+              payload: attentionPayload,
+            }
+          : {
+              type: "agent_stream",
+              payload: {
+                agentId: params.agentId,
+                event: {
+                  type: "attention_required",
+                  provider: params.provider,
+                  reason: params.reason,
+                  timestamp,
+                  shouldNotify,
+                  notification,
+                },
+                timestamp,
+              },
+            },
+      );
 
       this.sendToClient(ws, message);
     }