fix(client): keep daemon connections lean by default (#4470)

Move supported protocol capabilities into client defaults and make session traffic subscription-driven. Preserve timeline readiness and reconnect recovery across SDK, CLI, and plugin consumers.

packages/server/src/server/authorization/operation-permissions.ts

@@ -20,6 +20,7 @@ const INBOUND_PERMISSION = {
   "agent.skills.uninstall.request": "daemon.manage",
   "agent.timeline.list_prompts.request": "workspace.read",
   "agent.timeline.append.request": "workspace.write",
+  "session.events.set_subscription.request": ["workspace.read", "daemon.read", "hub.execute"],
   "agent.timeline.set_subscription.request": ["workspace.read", "hub.execute"],
   agent_permission_response: "workspace.write",
   archive_agent_request: ["workspace.write", "hub.execute"],
@@ -220,6 +221,7 @@ const OUTBOUND_PERMISSION = {
   "agent.timeline.list_prompts.response": "workspace.read",
   "agent.timeline.append.response": "workspace.write",
   "agent.timeline.replacement": ["workspace.read", "hub.execute"],
+  "session.events.set_subscription.response": ["workspace.read", "daemon.read", "hub.execute"],
   "agent.timeline.set_subscription.response": ["workspace.read", "hub.execute"],
   agent_archived: ["workspace.read", "hub.execute"],
   agent_attention_required: ["workspace.read", "hub.execute"],

packages/server/src/server/client-activity.e2e.test.ts

@@ -84,14 +84,11 @@ describe("client activity tracking", () => {
         reject(new Error(`Timeout waiting for attention_required (${timeout}ms)`));
       }, timeout);
 
-      const cleanup = client.on("agent_stream", (msg) => {
-        if (msg.type !== "agent_stream") return;
-        if (msg.payload.agentId !== agentId) return;
-        if (msg.payload.event.type !== "attention_required") return;
-
+      const cleanup = client.onAgentAttentionRequired((notification) => {
+        if (notification.agentId !== agentId) return;
         clearTimeout(timer);
         cleanup();
-        resolve(msg.payload.event);
+        resolve({ type: "attention_required", ...notification });
       });
     });
   }
@@ -107,7 +104,7 @@ describe("client activity tracking", () => {
         reject(new Error(`Timeout waiting for assistant timeline (${timeout}ms)`));
       }, timeout);
 
-      const cleanup = client.on("agent_stream", (msg) => {
+      const cleanup = client.subscribeAgentTimeline(agentId, (msg) => {
         if (msg.type !== "agent_stream") return;
         if (msg.payload.agentId !== agentId) return;
         if (msg.payload.event.type !== "timeline") return;

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

@@ -1473,7 +1473,7 @@ test("creates agent and exercises lifecycle", async () => {
       sawAssistantMessage = true;
     }
   });
-  const unsubscribeRawStream = ctx.client.on("agent_stream", (message) => {
+  const unsubscribeRawStream = ctx.client.subscribeAgentTimeline(agent.id, (message) => {
     if (message.type !== "agent_stream") {
       return;
     }

packages/server/src/server/daemon-e2e/user-message-contract.real.e2e.test.ts

@@ -61,7 +61,7 @@ function tmpCwd(provider: ContractProvider): string {
 
 function collectUserMessageEvents(client: DaemonClient, agentId: string): AgentStreamEvent[] {
   const events: AgentStreamEvent[] = [];
-  client.on("agent_stream", (message) => {
+  client.subscribeAgentTimeline(agentId, (message) => {
     if (message.type !== "agent_stream" || message.payload.agentId !== agentId) {
       return;
     }

packages/server/src/server/plugins/connection-demand.e2e.test.ts

@@ -0,0 +1,91 @@
+import { mkdtemp, writeFile, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { expect, test } from "vitest";
+import { createPaseoApi } from "@getpaseo/client";
+import { DaemonClient } from "../test-utils/daemon-client.js";
+import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+
+test("an RPC-only plugin receives no agent, project or provider data", async () => {
+  const directory = await mkdtemp(path.join(tmpdir(), "paseo-quiet-plugin-"));
+  await writeFile(path.join(directory, "paseo-plugin.json"), JSON.stringify({ id: "quiet" }));
+  await writeFile(
+    path.join(directory, "index.server.ts"),
+    `
+import { defineRpc } from "@getpaseo/plugin";
+import { z } from "zod";
+export default function contribute(server) {
+  const counts = {};
+  const observe = message => {
+    if (message.type !== "paseo_frame" || typeof message.data !== "string") return;
+    const frame = JSON.parse(message.data);
+    const type = frame.message?.type;
+    if (type) counts[type] = (counts[type] ?? 0) + 1;
+  };
+  process.on("message", observe);
+  server.handle(defineRpc({name:"counts", input:z.object({}), output:z.record(z.string(),z.number())}), () => counts);
+  return () => process.off("message", observe);
+}`,
+  );
+  const daemon = await createTestPaseoDaemon();
+  const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  const idle = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  const legacy = new DaemonClient({
+    url: `ws://127.0.0.1:${daemon.port}/ws`,
+    capabilities: {
+      selective_agent_timeline: false,
+      explicit_event_subscriptions: false,
+      provider_snapshot_references: false,
+    },
+  });
+  const legacyTypes: string[] = [];
+  legacy.subscribeRawMessages((message) => legacyTypes.push(message.type));
+  const idleTypes: string[] = [];
+  idle.subscribeRawMessages((message) => idleTypes.push(message.type));
+  try {
+    await client.connect();
+    await idle.connect();
+    await legacy.connect();
+    await client.patchDaemonConfig({ pluginsEnabled: true });
+    await client.installDirectoryPlugin(directory);
+    const agent = await client.createAgent({ provider: "claude", cwd: directory });
+    const api = createPaseoApi(client);
+    const received: string[] = [];
+    const release = api.agents
+      .ref(agent.id)
+      .timeline.subscribe((event) => received.push(event.event.type));
+    await release.ready;
+    await daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, {
+      type: "assistant_message",
+      id: "row",
+      text: "only the subscriber sees this",
+    });
+    await expect.poll(() => received).toEqual(["timeline"]);
+    await client.sendMessage(agent.id, "Say hello");
+    await client.waitForFinish(agent.id);
+    await client.refreshProvidersSnapshot({ cwd: directory });
+    const counts = await client.invokePluginRpc("quiet", "counts", {});
+    expect(counts).toEqual({ status: 1 });
+    await legacy.getDaemonConfig();
+    expect(legacyTypes).toEqual(
+      expect.arrayContaining(["project.update", "providers_snapshot_update", "agent_stream"]),
+    );
+    expect(idleTypes.filter((type) => type !== "status")).toEqual([]);
+    const deliveredBeforeRelease = received.length;
+    release();
+    await client.getDaemonConfig();
+    await daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, {
+      type: "assistant_message",
+      id: "row-2",
+      text: "unsubscribed",
+    });
+    await client.getDaemonConfig();
+    expect(received).toHaveLength(deliveredBeforeRelease);
+  } finally {
+    await legacy.close();
+    await idle.close();
+    await client.close();
+    await daemon.close();
+    await rm(directory, { recursive: true, force: true });
+  }
+}, 60_000);

packages/server/src/server/session.ts

@@ -1,3 +1,4 @@
+import type { SessionEventSubscription } from "@getpaseo/protocol/messages";
 import type { AgentRequests } from "./agent/requests/index.js";
 import equal from "fast-deep-equal";
 import { v4 as uuidv4 } from "uuid";
@@ -715,6 +716,8 @@ export class Session {
     unsubscribe: () => void;
   } | null = null;
   private projectSyncEnabled = false;
+  private readonly defaultEventSubscriptionSource = {};
+  private readonly eventSubscriptions = new Map<object, Set<SessionEventSubscription>>();
   private readonly workspaceUpdateTails = new Map<string, Promise<void>>();
   private clientActivity: {
     deviceType: "web" | "mobile";
@@ -925,6 +928,7 @@ export class Session {
         isProviderVisibleToClient: (provider) => this.isProviderVisibleToClient(provider),
         supportsCustomModeIcons: () => this.supports(CLIENT_CAPS.customModeIcons),
         supportsCompactProviderSnapshots: () => this.supports(CLIENT_CAPS.compactProviderSnapshots),
+        wantsSnapshotChanges: () => this.wantsEvent("providers_snapshot_update"),
         supportsProviderSnapshotReferences: () =>
           this.supports(CLIENT_CAPS.providerSnapshotReferences),
         listProviderAvailability: () => this.agentManager.listProviderAvailability(),
@@ -1135,6 +1139,7 @@ export class Session {
   updateClientCapabilities(capabilities: Record<string, unknown> | null, source?: object): void {
     this.clientCapabilities = parseClientCapabilities(capabilities);
     if (source) {
+      this.eventSubscriptions.delete(source);
       this.clientCapabilitiesBySource.set(source, this.clientCapabilities);
     }
     if (!source && !this.supports(CLIENT_CAPS.selectiveAgentTimeline)) {
@@ -1145,6 +1150,7 @@ export class Session {
 
   clearAgentTimelineSubscription(source: object): void {
     this.clientCapabilitiesBySource.delete(source);
+    this.eventSubscriptions.delete(source);
     if (this.viewedTimelineAgentIdsBySource.delete(source)) {
       this.rebuildViewedTimelineAgentIds();
     }
@@ -1223,6 +1229,7 @@ export class Session {
         continue;
       const supportsSelectiveDelivery = capabilities.has(CLIENT_CAPS.selectiveAgentTimeline);
       if (supportsSelectiveDelivery && serializedEvent.type === "attention_required") {
+        if (!this.wantsEvent("agent_attention_required", source)) continue;
         this.onMessageToSource(source, {
           type: "agent_attention_required",
           payload: {
@@ -1267,6 +1274,7 @@ export class Session {
   }
 
   private async publishProjectUpdate(update: ProjectUpdate): Promise<void> {
+    if (!this.wantsEvent("project.update")) return;
     const projectedPayload =
       update.kind === "upsert"
         ? { kind: "upsert" as const, project: await this.buildProjectDescriptor(update.project) }
@@ -1275,15 +1283,7 @@ export class Session {
       type: "project.update",
       payload: this.directorySync.sequenceProjectUpdate(projectedPayload, this.projectSyncEnabled),
     };
-    if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) {
-      if (this.supports(CLIENT_CAPS.projectUpdates)) this.emit(message);
-      return;
-    }
-    for (const [source, capabilities] of this.clientCapabilitiesBySource) {
-      if (capabilities.has(CLIENT_CAPS.projectUpdates)) {
-        this.onMessageToSource(source, message);
-      }
-    }
+    this.emit(message);
   }
 
   async syncWorkspaceGitObserverForWorkspace(workspace: PersistedWorkspaceRecord): Promise<void> {
@@ -2333,6 +2333,20 @@ export class Session {
         return this.handleProviderSubagentListRequest(msg);
       case "agent.provider_subagents.timeline.get.request":
         return this.handleProviderSubagentTimelineRequest(msg, source);
+      case "session.events.set_subscription.request": {
+        this.eventSubscriptions.set(
+          source ?? this.defaultEventSubscriptionSource,
+          new Set(msg.events),
+        );
+        this.emitForSource(
+          {
+            type: "session.events.set_subscription.response",
+            payload: { requestId: msg.requestId },
+          },
+          source,
+        );
+        return undefined;
+      }
       case "agent.timeline.set_subscription.request": {
         const agentIds = [...new Set(msg.agentIds)].sort();
         if (
@@ -7703,10 +7717,43 @@ export class Session {
   /**
    * Emit a message to the client
    */
+  // COMPAT(explicitEventSubscriptions): added in v0.8.0, remove legacy broadcasts after 2027-03-08.
+  private wantsEvent(event: SessionEventSubscription, source?: object): boolean {
+    if (!source && this.clientCapabilitiesBySource.size > 0) {
+      return [...this.clientCapabilitiesBySource.keys()].some((candidate) =>
+        this.wantsEvent(event, candidate),
+      );
+    }
+    const capabilities = source
+      ? this.clientCapabilitiesBySource.get(source)!
+      : this.clientCapabilities;
+    if (event === "project.update" && !capabilities.has(CLIENT_CAPS.projectUpdates)) return false;
+    return (
+      !capabilities.has(CLIENT_CAPS.explicitEventSubscriptions) ||
+      this.eventSubscriptions.get(source ?? this.defaultEventSubscriptionSource)?.has(event) ===
+        true
+    );
+  }
+
   private emit(msg: SessionOutboundMessage): void {
     if (!this.authorization.allowsOutbound(msg)) {
       return;
     }
+    if (
+      msg.type === "project.update" ||
+      msg.type === "providers_snapshot_update" ||
+      msg.type === "agent_attention_required" ||
+      msg.type === "agent_permission_request" ||
+      msg.type === "agent_permission_resolved"
+    ) {
+      if (this.clientCapabilitiesBySource.size > 0 && this.onMessageToSource) {
+        for (const source of this.clientCapabilitiesBySource.keys()) {
+          if (this.wantsEvent(msg.type, source)) this.onMessageToSource(source, msg);
+        }
+        return;
+      }
+      if (!this.wantsEvent(msg.type)) return;
+    }
     // JSON.stringify(msg) is only computed when trace is enabled — it runs for
     // every outbound message otherwise, and trace is disabled by default.
     // Optional-chained because test logger stubs don't implement isLevelEnabled.

packages/server/src/server/session/agent-updates/agent-updates-service.test.ts

@@ -681,3 +681,13 @@ describe("subscription lifecycle", () => {
     expect(h.agentUpdates()).toEqual([{ kind: "remove", agentId: "a" }]);
   });
 });
+
+test("an idle session skips agent hydration while preserving workspace updates", async () => {
+  const h = buildHarness();
+  const agent = { ...h.managed("unobserved"), workspaceId: "workspace" };
+  // No payload is registered: attempting to hydrate this agent would fail.
+  await h.service.forwardLiveAgent(agent);
+  expect(h.loggedErrors).toEqual([]);
+  expect(h.agentUpdates()).toEqual([]);
+  expect(h.workspaceUpdates).toEqual([agent.workspaceId]);
+});

packages/server/src/server/session/agent-updates/agent-updates-service.ts

@@ -390,6 +390,18 @@ export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentU
   }
 
   function forwardLiveAgent(agent: ManagedAgent): Promise<void> {
+    if (!subscription) {
+      const workspaceId = agent.workspaceId;
+      return workspaceId
+        ? enqueueAgentUpdate(agent.id, async () => {
+            try {
+              await deps.emitWorkspaceUpdateForWorkspaceId(workspaceId);
+            } catch (error) {
+              deps.logger.error({ err: error }, "Failed to emit workspace update");
+            }
+          })
+        : Promise.resolve();
+    }
     const payload = toAgentPayload(agent);
     return enqueueAgentUpdate(payload.id, () => emitLiveAgentUpdate(payload));
   }

packages/server/src/server/session/provider/provider-catalog-session.test.ts

@@ -54,6 +54,7 @@ function makeSubsystem(options: MakeOptions = {}) {
     isProviderVisibleToClient: (provider) => visible.has(provider),
     supportsCustomModeIcons: () => options.supportsCustomModeIcons ?? false,
     supportsProviderSnapshotReferences: () => false,
+    wantsSnapshotChanges: () => true,
     supportsCompactProviderSnapshots: () => options.supportsCompactProviderSnapshots ?? false,
     listProviderAvailability: async () => [],
     listDraftFeatures: async () => [],
@@ -395,6 +396,7 @@ it("announces shared content without retransmitting models or hashing discovery
         isProviderVisibleToClient: () => true,
         supportsCustomModeIcons: () => true,
         supportsCompactProviderSnapshots: () => true,
+        wantsSnapshotChanges: () => true,
         supportsProviderSnapshotReferences: () => references,
         listProviderAvailability: async () => [],
         listDraftFeatures: async () => [],

packages/server/src/server/session/provider/provider-catalog-session.ts

@@ -48,6 +48,7 @@ export interface ProviderCatalogSessionHost {
   supportsCustomModeIcons(): boolean;
   supportsCompactProviderSnapshots(): boolean;
   supportsProviderSnapshotReferences(): boolean;
+  wantsSnapshotChanges(): boolean;
   listProviderAvailability(): Promise<ProviderAvailability[]>;
   listDraftFeatures(config: AgentSessionConfig): Promise<AgentFeature[]>;
 }
@@ -82,6 +83,7 @@ export class ProviderCatalogSession {
 
   start(): void {
     const handleProviderSnapshotChange = (transition: ProviderSnapshotTransition) => {
+      if (!this.host.wantsSnapshotChanges()) return;
       const previous = this.visibleSnapshot(transition.previous);
       const current = this.visibleSnapshot(transition.current);
       if (sameSnapshotRecords(previous.records, current.records)) return;

packages/server/src/server/voice-local-agent.e2e.test.ts

@@ -32,7 +32,7 @@ function makeErrorHandler(reject: (error: Error) => void) {
 }
 
 function makeSpeakToolHandler(resolve: (value: string) => void) {
-  return (msg: SessionMessage<"agent_stream">) => {
+  return (msg: SessionMessage<"agent_stream" | "agent.timeline.replacement">) => {
     if (msg.type !== "agent_stream") return;
     if (msg.payload.event.type !== "timeline") return;
     const item = msg.payload.event.item;
@@ -131,7 +131,10 @@ function waitForSignal<T>(
     });
 
     const speakToolPromise = waitForSignal<string>(120000, (resolve, reject) => {
-      const offStream = ctx.client.on("agent_stream", makeSpeakToolHandler(resolve));
+      const offStream = ctx.client.subscribeAgentTimeline(
+        targetAgent.id,
+        makeSpeakToolHandler(resolve),
+      );
       const offError = ctx.client.on("activity_log", makeErrorHandler(reject));
       return () => {
         offStream();

packages/server/src/server/voice-roundtrip.e2e.test.ts

@@ -225,7 +225,7 @@ for (const targetProvider of [
       const timelineToolAgentIds = new Set<string>();
       const activityErrors: string[] = [];
 
-      const offStream = ctx.client.on("agent_stream", (message) => {
+      const offStream = ctx.client.subscribeAgentTimeline(voiceAgentId, (message) => {
         if (message.type !== "agent_stream") {
           return;
         }

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

@@ -1755,6 +1755,7 @@ export class VoiceAssistantWebSocketServer {
         forgeProviders: true,
         // COMPAT(selectiveAgentTimeline): added in v0.1.106, remove after 2027-01-12.
         selectiveAgentTimeline: true,
+        explicitEventSubscriptions: true,
         // COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30.
         canonicalSubmittedPrompts: true,
         // COMPAT(stableProjectIdentity): added in v0.1.109, remove gate after 2027-01-15.