Reduce workspace, agent, and chat sync traffic (#2028)

* 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. * 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. * test(app): align browser fixtures with runtime contracts * fix(sync): preserve mixed-client delivery guarantees * fix(sync): finish paged history before advancing * fix(sync): replay live deltas after refresh failures * test(server): model socket capability lookup * test(app): use platform shortcut for split-pane coverage * fix(app): keep hidden timelines dormant * test(app): normalize split-pane shortcut setup * fix(app): preserve sync state across retries * fix(app): preserve background sync state across races * fix directory bootstrap reconciliation edge cases * fix selective timeline compatibility races * fix superseded directory sync races * fix(sync): centralize directory replica ordering Own agent and workspace refresh transactions in HostRuntime so reconnect epochs, buffered updates, and lifecycle invalidation share one ordering boundary. Route timeline metadata and mixed-capability stream delivery through their authoritative runtime sources. * fix(app): keep timeline requests runtime-scoped * fix(app): close directory bootstrap races
2026-07-17 17:03:34 +02:00
parent 6c99efae52
commit 1977d330ed
57 changed files with 5783 additions and 1343 deletions
@@ -3033,11 +3033,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",
        },
      }),
@@ -3111,11 +3111,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",
-            },
+            }),
          }),
        }),
      ),
@@ -0,0 +1,287 @@
+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("subscription acknowledgements stay on the requesting socket of a retained session", async () => {
+  const legacy = await connect({ clientId: "shared-client", selective: false });
+  const capable = await connect({ clientId: "shared-client", selective: true });
+  legacy.clear();
+  capable.clear();
+
+  await capable.client.setAgentTimelineSubscription(["agent-a"]);
+  await capable.barrier("targeted-subscription-ack");
+
+  expect(
+    legacy.messages.some((message) => message.type === "agent.timeline.set_subscription.response"),
+  ).toBe(false);
+});
+
+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);
@@ -10,6 +10,7 @@ import {
  assertPullRequestAutoMergeEnableReady,
 } from "../services/github-service.js";
 import { PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels";
+import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities";
 import type { WorkspaceDescriptorPayload } from "@getpaseo/protocol/messages";
 import {
  decodeFileTransferFrame,
@@ -21,6 +22,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";
@@ -309,6 +311,7 @@ interface SessionForTestOptions {
  daemonRuntimeConfig?: SessionOptions["daemonRuntimeConfig"];
  downloadTokenStore?: SessionOptions["downloadTokenStore"];
  messages?: unknown[];
+  targetedMessages?: Array<{ source: object; message: SessionOutboundMessage }>;
  binaryMessages?: Uint8Array[];
 }

@@ -346,6 +349,12 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
  return new Session({
    clientId: "test-client",
    onMessage: (message) => messages.push(message),
+    ...(options.targetedMessages
+      ? {
+          onMessageToSource: (source: object, message: SessionOutboundMessage) =>
+            options.targetedMessages?.push({ source, message }),
+        }
+      : {}),
    onBinaryMessage: createBinaryMessageHandler(options.binaryMessages),
    logger,
    downloadTokenStore: options.downloadTokenStore ?? asDownloadTokenStore(),
@@ -4554,6 +4563,216 @@ 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",
+      },
+    },
+  ]);
+});
+
+test("acknowledges a timeline subscription only to its socket source", async () => {
+  const messages: SessionOutboundMessage[] = [];
+  const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = [];
+  const session = createSessionForTest({ messages, targetedMessages });
+  const capableSocket = {};
+  session.updateClientCapabilities({ selective_agent_timeline: true }, capableSocket);
+
+  await session.handleMessage(
+    {
+      type: "agent.timeline.set_subscription.request",
+      agentIds: ["agent-a"],
+      requestId: "timeline-subscription-targeted",
+    },
+    capableSocket,
+  );
+
+  expect(messages).toEqual([]);
+  expect(targetedMessages).toEqual([
+    {
+      source: capableSocket,
+      message: {
+        type: "agent.timeline.set_subscription.response",
+        payload: {
+          agentIds: ["agent-a"],
+          requestId: "timeline-subscription-targeted",
+        },
+      },
+    },
+  ]);
+});
+
+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 = {};
+  session.updateClientCapabilities({ selective_agent_timeline: true }, firstSocket);
+  session.updateClientCapabilities({ selective_agent_timeline: true }, 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);
+
+  const legacySocket = {};
+  session.updateClientCapabilities(null, legacySocket);
+  expect(session.supportsForSource(CLIENT_CAPS.selectiveAgentTimeline, legacySocket)).toBe(false);
+  expect(session.supportsForSource(CLIENT_CAPS.selectiveAgentTimeline, firstSocket)).toBe(true);
+  messages.length = 0;
+  forward({
+    type: "agent_stream",
+    agentId: "agent-not-viewed",
+    event: {
+      type: "timeline",
+      provider: "mock",
+      item: { type: "assistant_message", messageId: "message-legacy", text: "legacy" },
+    },
+  });
+  expect(messages.some((message) => message.type === "agent_stream")).toBe(true);
+
+  session.clearAgentTimelineSubscription(legacySocket);
+
+  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"]);
+});
+
+test("keeps selective delivery scoped per socket when a retained session also has a legacy socket", async () => {
+  const messages: SessionOutboundMessage[] = [];
+  const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = [];
+  const agentEventListeners: Array<(event: AgentManagerEvent) => void> = [];
+  const session = createSessionForTest({
+    messages,
+    targetedMessages,
+    agentManager: {
+      subscribe: vi.fn((listener: (event: AgentManagerEvent) => void) => {
+        agentEventListeners.push(listener);
+        return () => {};
+      }),
+    },
+  });
+  const legacySocket = {};
+  const selectiveSocket = {};
+  session.updateClientCapabilities(null, legacySocket);
+  session.updateClientCapabilities({ selective_agent_timeline: true }, selectiveSocket);
+  await session.handleMessage(
+    {
+      type: "agent.timeline.set_subscription.request",
+      agentIds: ["viewed-agent"],
+      requestId: "timeline-subscription-selective",
+    },
+    selectiveSocket,
+  );
+  targetedMessages.length = 0;
+
+  const listener = agentEventListeners[0];
+  if (!listener) throw new Error("Agent event listener was not installed");
+  listener({
+    type: "agent_stream",
+    agentId: "not-viewed-agent",
+    event: {
+      type: "timeline",
+      provider: "mock",
+      item: { type: "assistant_message", messageId: "message-global", text: "global" },
+    },
+  });
+
+  expect(messages).toEqual([]);
+  expect(targetedMessages).toEqual([
+    {
+      source: legacySocket,
+      message: expect.objectContaining({
+        type: "agent_stream",
+        payload: expect.objectContaining({ agentId: "not-viewed-agent" }),
+      }),
+    },
+  ]);
+});
+
 describe("agent config setters", () => {
  test("set_agent_mode_request: success emits accepted response carrying the notice", async () => {
    const messages: SessionOutboundMessage[] = [];
@@ -429,6 +429,7 @@ export interface SessionOptions {
  appVersion?: string | null;
  clientCapabilities?: Record<string, unknown> | null;
  onMessage: (msg: SessionOutboundMessage) => void;
+  onMessageToSource?: (source: object, msg: SessionOutboundMessage) => void;
  onBinaryMessage?: (frame: Uint8Array) => void;
  getTransportBufferedAmount?: () => number | null;
  onLifecycleIntent?: (intent: SessionLifecycleIntent) => void;
@@ -558,6 +559,9 @@ export class Session {
  private clientCapabilities: ReadonlySet<ClientCapability>;
  private readonly sessionId: string;
  private readonly onMessage: (msg: SessionOutboundMessage) => void;
+  private readonly onMessageToSource:
+    | ((source: object, msg: SessionOutboundMessage) => void)
+    | null;
  private readonly onBinaryMessage: ((frame: Uint8Array) => void) | null;
  private readonly getTransportBufferedAmount: () => number | null;
  private readonly onLifecycleIntent: ((intent: SessionLifecycleIntent) => void) | null;
@@ -583,6 +587,10 @@ export class Session {
  private readonly daemonConfigStore: DaemonConfigStore;
  private readonly pushTokenStore: PushTokenStore;
  private unsubscribeAgentEvents: (() => void) | null = null;
+  private viewedTimelineAgentIds = new Set<string>();
+  private readonly viewedTimelineAgentIdsBySource = new Map<object, Set<string>>();
+  private readonly selectiveTimelineCapabilityBySource = new Map<object, boolean>();
+  private readonly defaultTimelineSubscriptionSource = {};
  private unsubscribeTerminalWorkspaceContributionEvents: (() => void) | null = null;
  private readonly agentUpdates: AgentUpdatesService;
  private workspaceUpdatesSubscription: WorkspaceUpdatesSubscriptionState | null = null;
@@ -625,6 +633,7 @@ export class Session {
      appVersion,
      clientCapabilities,
      onMessage,
+      onMessageToSource,
      onBinaryMessage,
      getTransportBufferedAmount,
      onLifecycleIntent,
@@ -675,6 +684,7 @@ export class Session {
    this.clientCapabilities = parseClientCapabilities(clientCapabilities);
    this.sessionId = uuidv4();
    this.onMessage = onMessage;
+    this.onMessageToSource = onMessageToSource ?? null;
    this.onBinaryMessage = onBinaryMessage ?? null;
    this.getTransportBufferedAmount = getTransportBufferedAmount ?? (() => 0);
    this.onLifecycleIntent = onLifecycleIntent ?? null;
@@ -963,14 +973,118 @@ export class Session {
    }
  }

-  updateClientCapabilities(capabilities: Record<string, unknown> | null): void {
+  updateClientCapabilities(capabilities: Record<string, unknown> | null, source?: object): void {
    this.clientCapabilities = parseClientCapabilities(capabilities);
+    if (source) {
+      this.selectiveTimelineCapabilityBySource.set(
+        source,
+        this.supports(CLIENT_CAPS.selectiveAgentTimeline),
+      );
+    }
+    if (!source && !this.supports(CLIENT_CAPS.selectiveAgentTimeline)) {
+      this.viewedTimelineAgentIdsBySource.clear();
+      this.viewedTimelineAgentIds.clear();
+    }
+  }
+
+  clearAgentTimelineSubscription(source: object): void {
+    this.selectiveTimelineCapabilityBySource.delete(source);
+    if (this.viewedTimelineAgentIdsBySource.delete(source)) {
+      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;
+  }
+
+  private usesSelectiveTimelineDelivery(): boolean {
+    if (this.selectiveTimelineCapabilityBySource.size === 0) {
+      return this.supports(CLIENT_CAPS.selectiveAgentTimeline);
+    }
+    for (const capable of this.selectiveTimelineCapabilityBySource.values()) {
+      if (!capable) return false;
+    }
+    return true;
+  }
+
+  private forwardAgentStream(
+    event: Extract<AgentManagerEvent, { type: "agent_stream" }>,
+    serializedEvent: Extract<SessionOutboundMessage, { type: "agent_stream" }>["payload"]["event"],
+  ): void {
+    if (this.selectiveTimelineCapabilityBySource.size === 0 || !this.onMessageToSource) {
+      if (this.usesSelectiveTimelineDelivery() && 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.usesSelectiveTimelineDelivery() ||
+        this.viewedTimelineAgentIds.has(event.agentId)
+      ) {
+        this.emit({
+          type: "agent_stream",
+          payload: this.buildAgentStreamPayload(event, serializedEvent),
+        });
+      }
+      return;
+    }
+
+    for (const [source, supportsSelectiveDelivery] of this.selectiveTimelineCapabilityBySource) {
+      if (supportsSelectiveDelivery && serializedEvent.type === "attention_required") {
+        this.onMessageToSource(source, {
+          type: "agent_attention_required",
+          payload: {
+            agentId: event.agentId,
+            reason: serializedEvent.reason,
+            timestamp: serializedEvent.timestamp,
+            shouldNotify: serializedEvent.shouldNotify,
+            ...(serializedEvent.notification ? { notification: serializedEvent.notification } : {}),
+          },
+        });
+        continue;
+      }
+      if (
+        supportsSelectiveDelivery &&
+        !this.viewedTimelineAgentIdsBySource.get(source)?.has(event.agentId)
+      ) {
+        continue;
+      }
+      this.onMessageToSource(source, {
+        type: "agent_stream",
+        payload: this.buildAgentStreamPayload(event, serializedEvent),
+      });
+    }
  }

  supports(capability: ClientCapability): boolean {
    return this.clientCapabilities.has(capability);
  }

+  supportsForSource(capability: ClientCapability, source: object): boolean {
+    if (capability === CLIENT_CAPS.selectiveAgentTimeline) {
+      return this.selectiveTimelineCapabilityBySource.get(source) ?? this.supports(capability);
+    }
+    return this.supports(capability);
+  }
+
  async syncWorkspaceGitObserverForWorkspace(workspace: PersistedWorkspaceRecord): Promise<void> {
    await this.workspaceGitObserver.syncObserverForWorkspace(workspace);
  }
@@ -979,6 +1093,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 +1395,7 @@ export class Session {
          "agent.session.forward_stream",
        );

-        this.emit({
-          type: "agent_stream",
-          payload: this.buildAgentStreamPayload(event, serializedEvent),
-        });
+        this.forwardAgentStream(event, serializedEvent);

        if (event.event.type === "permission_requested") {
          this.emit({
@@ -1377,7 +1498,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;
@@ -1391,7 +1512,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");
@@ -1429,12 +1550,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) ??
@@ -1516,7 +1637,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);
@@ -1524,6 +1648,24 @@ 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 (
+          source
+            ? (this.selectiveTimelineCapabilityBySource.get(source) ??
+              this.supports(CLIENT_CAPS.selectiveAgentTimeline))
+            : this.supports(CLIENT_CAPS.selectiveAgentTimeline)
+        ) {
+          this.replaceAgentTimelineSubscription(source, agentIds);
+        }
+        const response: SessionOutboundMessage = {
+          type: "agent.timeline.set_subscription.response",
+          payload: { agentIds, requestId: msg.requestId },
+        };
+        if (source && this.onMessageToSource) this.onMessageToSource(source, response);
+        else this.emit(response);
+        return undefined;
+      }
      case "agent.fork_context.request":
        return this.handleAgentForkContextRequest(msg);
      default:
@@ -3967,6 +4109,8 @@ export class Session {
          continue;
        }
      }
+      const workspaceId = payload.kind === "upsert" ? payload.workspace.id : payload.id;
+      subscription.lastEmittedByWorkspaceId.set(workspaceId, payload);
      this.emit({
        type: "workspace_update",
        payload,
@@ -4259,6 +4403,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,
@@ -4515,6 +4662,7 @@ export class Session {
        "fetch_workspaces_response_ready",
      );
      const snapshot = this.buildBootstrapSnapshot(payload.entries);
+      this.seedWorkspaceSubscriptionSnapshot(subscriptionId, request.filter, payload.entries);

      this.emit({
        type: "fetch_workspaces_response",
@@ -4575,6 +4723,23 @@ export class Session {
    return { snapshotByWorkspaceId };
  }

+  private seedWorkspaceSubscriptionSnapshot(
+    subscriptionId: string | null,
+    filter: FetchWorkspacesRequestFilter | undefined,
+    entries: FetchWorkspacesResponseEntry[],
+  ): void {
+    const subscription = this.workspaceUpdatesSubscription;
+    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",
+        workspace: entry,
+      });
+    }
+  }
+
  private async registerWorkspaceForImportedAgent(
    workspace: PersistedWorkspaceRecord,
  ): Promise<void> {
@@ -4661,7 +4826,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 +4911,7 @@ export class Session {
        error: null,
      },
    });
-    this.emit({
-      type: "workspace_update",
-      payload: { kind: "upsert", workspace: descriptor },
-    });
+    await this.emitCreatedWorkspaceUpdate(descriptor);
  }

  private async resolveWorktreeSourceCwd(input: {
@@ -6568,6 +6568,64 @@ test("subscribed fetch_workspaces includes git enrichment in the initial snapsho
  );
 });

+test("a workspace leaving a filtered subscription after bootstrap emits a removal", async () => {
+  const emitted: SessionOutboundMessage[] = [];
+  const session = asTestSession(
+    createSessionForWorkspaceTests({ onMessage: (message) => emitted.push(message) }),
+  );
+  const descriptor = {
+    id: "ws-buffered",
+    projectId: "proj-buffered",
+    projectDisplayName: "repo",
+    projectRootPath: REPO_CWD,
+    workspaceDirectory: REPO_CWD,
+    projectKind: "git" as const,
+    workspaceKind: "local_checkout" as const,
+    name: "repo work",
+    status: "done" as const,
+    activityAt: null,
+    diffStat: null,
+  };
+  let currentDescriptor: typeof descriptor | null = descriptor;
+  let finishListing: (result: ListFetchResult) => void = () => {};
+  const listing = new Promise<ListFetchResult>((resolve) => {
+    finishListing = resolve;
+  });
+  session.listFetchWorkspacesEntries = async () => listing;
+  session.buildWorkspaceDescriptorMap = async () =>
+    new Map(currentDescriptor ? [[currentDescriptor.id, currentDescriptor]] : []);
+  session.reconcileAndEmitWorkspaceUpdates = async () => undefined;
+
+  const bootstrap = session.handleMessage({
+    type: "fetch_workspaces_request",
+    requestId: "req-buffered-filter",
+    filter: { query: "repo" },
+    subscribe: { subscriptionId: "sub-buffered-filter" },
+  });
+  await session.emitWorkspaceUpdatesForWorkspaceIds([descriptor.id], { skipReconcile: true });
+  finishListing({
+    entries: [],
+    emptyProjects: [],
+    pageInfo: { nextCursor: null, prevCursor: null, hasMore: false },
+  });
+  await bootstrap;
+
+  expect(filterByType(emitted, "workspace_update")).toEqual([
+    { type: "workspace_update", payload: { kind: "upsert", workspace: descriptor } },
+  ]);
+
+  emitted.length = 0;
+  currentDescriptor = { ...descriptor, name: "other work" };
+  await session.emitWorkspaceUpdatesForWorkspaceIds([descriptor.id], { skipReconcile: true });
+
+  expect(filterByType(emitted, "workspace_update")).toEqual([
+    {
+      type: "workspace_update",
+      payload: { kind: "remove", id: descriptor.id },
+    },
+  ]);
+});
+
 test("project.rename.request stores customName and emits an updated workspace descriptor", async () => {
  const emitted: SessionOutboundMessage[] = [];
  const session = asTestSession(
@@ -7827,4 +7885,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([]);
 });
@@ -173,6 +173,8 @@ function createSessionWithActivity(
 ) {
  return {
    getClientActivity: vi.fn(() => activity),
+    supports: () => false,
+    supportsForSource: () => false,
  };
 }

@@ -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(() => {});
@@ -988,6 +988,12 @@ export class VoiceAssistantWebSocketServer {
        }
        this.sendToConnection(connection, wrapSessionMessage(msg));
      },
+      onMessageToSource: (source, msg) => {
+        if (!connection || !connection.sockets.has(source as WebSocketLike)) {
+          return;
+        }
+        this.sendToClient(source as WebSocketLike, wrapSessionMessage(msg));
+      },
      onBinaryMessage: (frame) => {
        if (!connection) {
          return;
@@ -1093,6 +1099,7 @@ export class VoiceAssistantWebSocketServer {
      sockets: new Set([ws]),
      externalDisconnectCleanupTimeout: null,
    };
+    session.updateClientCapabilities(clientCapabilities, ws);
    return connection;
  }

@@ -1163,12 +1170,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, ws);
      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 +1291,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,
      },
    };
  }
@@ -1390,6 +1402,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) {
@@ -1758,7 +1771,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);

@@ -2009,21 +2022,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.supportsForSource(CLIENT_CAPS.selectiveAgentTimeline, ws)
+          ? {
+              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);
    }