fix(subscriptions): isolate observers and silence unrequested feeds

Filtered plugin observations shared a replaceable directory subscription with the app. Give each observation independent source ownership and route all application feeds through explicit demand, preserving combined listing and subscription.

packages/server/scripts/benchmark-startup-git-real-home.ts

@@ -140,14 +140,15 @@ async function main(): Promise<void> {
     await client.connect();
     const seen = new Set<string>();
     client.on("checkout_status_update", (message) => seen.add(message.payload.cwd));
+    await client.observeEvents(["checkout_status_update"]).ready;
     const workspaces = await client.fetchWorkspaces({
-      subscribe: { subscriptionId: `startup-git-${scenario}` },
+      subscribe: {},
       sort: [{ key: "activity_at", direction: "desc" }],
       page: { limit: 200 },
     });
     await client.fetchAgents({
       scope: "active",
-      subscribe: { subscriptionId: `startup-agents-${scenario}` },
+      subscribe: {},
       page: { limit: 200 },
     });
 

packages/server/scripts/test-mcp-inject.ts

@@ -151,7 +151,7 @@ async function main(): Promise<void> {
 
     await client.connect();
     await client.fetchAgents({
-      subscribe: { subscriptionId: "mcp-inject-real-claude" },
+      subscribe: {},
     });
 
     const results: ProviderRunResult[] = [];

packages/server/scripts/voice-main-daemon-e2e.ts

@@ -47,7 +47,7 @@ async function runVoiceRoundTrip(params: {
   const client = new DaemonClient({ url: `${params.daemonUrl}/ws` });
   await client.connect();
   await client.fetchAgents({
-    subscribe: { subscriptionId: `voice-e2e-${randomUUID()}` },
+    subscribe: {},
   });
 
   const mode = await client.setVoiceMode(true, params.voiceAgentId);
@@ -77,7 +77,7 @@ async function runVoiceRoundTrip(params: {
     });
   });
 
-  const offStream = client.on("agent_stream", (msg) => {
+  const offStream = client.subscribeAgentTimeline(activeVoiceAgentId, (msg) => {
     if (msg.type !== "agent_stream") return;
     if (msg.payload.agentId !== activeVoiceAgentId) return;
     if (msg.payload.event.type !== "timeline") return;
@@ -88,6 +88,8 @@ async function runVoiceRoundTrip(params: {
     }
   });
 
+  await offStream.ready;
+
   const chunkBytes = 4800;
   for (let offset = 0; offset < pcm.length; offset += chunkBytes) {
     const chunk = pcm.subarray(offset, Math.min(pcm.length, offset + chunkBytes));

packages/server/scripts/voice-roundtrip-debug.ts

@@ -49,7 +49,7 @@ async function main(): Promise<void> {
 
   try {
     await client.connect();
-    await client.fetchAgents({ subscribe: { subscriptionId: "voice-debug" } });
+    await client.fetchAgents({ subscribe: {} });
 
     const voiceCwd = mkdtempSync(path.join(tmpdir(), "voice-roundtrip-debug-"));
     const voiceAgent = await client.createAgent({
@@ -88,7 +88,7 @@ async function main(): Promise<void> {
       }
     });
 
-    const offStream = client.on("agent_stream", (msg) => {
+    const offStream = client.subscribeAgentTimeline(voiceAgentId, (msg) => {
       if (msg.type !== "agent_stream") return;
       if (msg.payload.event.type !== "timeline") return;
       const item = msg.payload.event.item;
@@ -100,6 +100,8 @@ async function main(): Promise<void> {
       });
     });
 
+    await offStream.ready;
+
     let audioChunkCount = 0;
     const firstAudio = new Promise<void>((resolve) => {
       const offAudio = client.on("audio_output", (msg) => {