Keep other windows in sync when an agent is deleted (#8248)

* fix(server): broadcast agent removals across logical sessions

* test(server): make subscription fixtures portable on Windows

The newly enabled integration suite assumed /bin/cat and a one-second reply deadline, so Windows CI failed before completing its subscription coverage. Use the running Node executable for echo terminals and let diagnostics finish their bounded external tool probes.

Co-authored-by: loganmeyer <50949542+loganmeyer@users.noreply.github.com>

---------

Co-authored-by: loganmeyer <50949542+loganmeyer@users.noreply.github.com>
Co-authored-by: paseo-bot[bot] <paseo-bot[bot]@users.noreply.github.com>

packages/server/src/server/agent/agent-manager.test.ts

@@ -8059,6 +8059,60 @@ test("global subscribers receive internal agent events only when they opt in", a
   unsubscribeOptedIn();
 });
 
+test("agent removal notifications respect scoped subscriptions and internal visibility", () => {
+  const manager = new AgentManager({ clients: {}, logger });
+  const removedId = "00000000-0000-4000-8000-000000000901";
+  const internalId = "00000000-0000-4000-8000-000000000902";
+  const otherId = "00000000-0000-4000-8000-000000000903";
+  const publicEvents: AgentManagerEvent[] = [];
+  const allEvents: AgentManagerEvent[] = [];
+  const scopedEvents: AgentManagerEvent[] = [];
+  const internalEvents: AgentManagerEvent[] = [];
+  const unrelatedEvents: AgentManagerEvent[] = [];
+  const releases = [
+    manager.subscribe((event) => publicEvents.push(event), { replayState: false }),
+    manager.subscribe((event) => allEvents.push(event), {
+      replayState: false,
+      includeInternal: true,
+    }),
+    manager.subscribe((event) => scopedEvents.push(event), {
+      replayState: false,
+      agentId: removedId,
+    }),
+    manager.subscribe((event) => internalEvents.push(event), {
+      replayState: false,
+      agentId: internalId,
+    }),
+    manager.subscribe((event) => unrelatedEvents.push(event), {
+      replayState: false,
+      agentId: otherId,
+    }),
+  ];
+  try {
+    manager.notifyAgentRemoved({ id: removedId, workspaceId: "workspace" });
+    manager.notifyAgentRemoved({ id: internalId, workspaceId: "workspace", internal: true });
+    const publicRemoval = {
+      type: "agent_removed",
+      agentId: removedId,
+      workspaceId: "workspace",
+      internal: false,
+    };
+    const internalRemoval = {
+      type: "agent_removed",
+      agentId: internalId,
+      workspaceId: "workspace",
+      internal: true,
+    };
+    expect(publicEvents).toEqual([publicRemoval]);
+    expect(allEvents).toEqual([publicRemoval, internalRemoval]);
+    expect(scopedEvents).toEqual([publicRemoval]);
+    expect(internalEvents).toEqual([internalRemoval]);
+    expect(unrelatedEvents).toEqual([]);
+  } finally {
+    for (const release of releases) release();
+  }
+});
+
 test("getAgent returns internal agents by ID", async () => {
   const internalAgentId = "00000000-0000-4000-8000-000000000107";
   const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-"));

packages/server/src/server/agent/agent-manager.ts

@@ -236,6 +236,7 @@ export type {
 
 export type AgentManagerEvent =
   | { type: "agent_state"; agent: ManagedAgent }
+  | { type: "agent_removed"; agentId: string; workspaceId?: string; internal: boolean }
   | { type: "provider_subagent"; event: ProviderSubagentStoreEvent }
   | { type: "timeline_replacement"; agentId: string; epoch: string }
   | {
@@ -3519,6 +3520,15 @@ export class AgentManager {
     await this.deleteCommittedTimeline(agentId);
   }
 
+  notifyAgentRemoved(agent: Pick<StoredAgentRecord, "id" | "workspaceId" | "internal">): void {
+    this.dispatch({
+      type: "agent_removed",
+      agentId: agent.id,
+      workspaceId: agent.workspaceId,
+      internal: agent.internal === true,
+    });
+  }
+
   async deleteCommittedTimeline(agentId: string): Promise<void> {
     await this.durableTimelineStore?.deleteAgent(agentId);
   }
@@ -5376,7 +5386,7 @@ export class AgentManager {
     for (const subscriber of this.subscribers) {
       if (
         subscriber.agentId &&
-        event.type === "agent_stream" &&
+        (event.type === "agent_stream" || event.type === "agent_removed") &&
         subscriber.agentId !== event.agentId
       ) {
         continue;
@@ -5412,6 +5422,7 @@ export class AgentManager {
 
   private eventBelongsToInternalAgent(event: AgentManagerEvent): boolean {
     if (event.type === "agent_state") return event.agent.internal === true;
+    if (event.type === "agent_removed") return event.internal;
     if (event.type === "agent_stream") return this.agents.get(event.agentId)?.internal === true;
     if (event.type !== "provider_subagent") return false;
     const parentAgentId =

packages/server/src/server/owned-subscriptions.e2e.test.ts

@@ -1,7 +1,8 @@
 import { MockLoadTestAgentClient } from "./agent/providers/mock-load-test-agent.js";
 import { createPaseoApi } from "@getpaseo/client";
+import type { AgentManager } from "./agent/agent-manager.js";
 import { execFileSync } from "node:child_process";
-import { mkdir, writeFile } from "node:fs/promises";
+import { mkdir, readFile, writeFile } from "node:fs/promises";
 import path from "node:path";
 import { expect, test } from "vitest";
 import type { z } from "zod";
@@ -71,15 +72,18 @@ class SubscriptionPeer {
   async request(message: z.input<typeof SessionInboundMessageSchema>): Promise<void> {
     this.socket.send(JSON.stringify({ type: "session", message }));
     await expect
-      .poll(() =>
-        this.frames.some(
-          (frame) =>
-            frame.type === "session" &&
-            "payload" in frame.message &&
-            "requestId" in frame.message.payload &&
-            "requestId" in message &&
-            frame.message.payload.requestId === message.requestId,
-        ),
+      .poll(
+        () =>
+          this.frames.some(
+            (frame) =>
+              frame.type === "session" &&
+              "payload" in frame.message &&
+              "requestId" in frame.message.payload &&
+              "requestId" in message &&
+              frame.message.payload.requestId === message.requestId,
+          ),
+        // Diagnostics invoke external tools with their own multi-second deadlines.
+        { timeout: 10_000 },
       )
       .toBe(true);
   }
@@ -119,6 +123,16 @@ class SubscriptionPeer {
   }
 }
 
+function removedAgentIds(peer: SubscriptionPeer): string[] {
+  return peer.frames.flatMap((frame) =>
+    frame.type === "session" &&
+    frame.message.type === "agent_update" &&
+    frame.message.payload.kind === "remove"
+      ? [frame.message.payload.agentId]
+      : [],
+  );
+}
+
 test("a pure list reply belongs only to its source socket in a shared logical session", async () => {
   const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
   const peers: SubscriptionPeer[] = [];
@@ -225,6 +239,96 @@ test("combined filtered agent lists create independent server-owned observations
   }
 });
 
+test.each([
+  {
+    state: "live",
+    prepare: async (_manager: AgentManager, _agentId: string) => {},
+  },
+  {
+    state: "stored-only",
+    prepare: (manager: AgentManager, agentId: string) => manager.closeAgent(agentId),
+  },
+])(
+  "$state agent deletion reaches directory observers across logical sessions without broadcasting its reply",
+  async ({ prepare }) => {
+    const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
+    const admin = new DaemonClient({
+      url: `ws://127.0.0.1:${daemon.port}/ws`,
+      appVersion: "0.8.0",
+    });
+    const peers: SubscriptionPeer[] = [];
+    try {
+      await admin.connect();
+      const workspace = (
+        await admin.createWorkspace({ source: { kind: "directory", path: daemon.staticDir } })
+      ).workspace;
+      if (!workspace) throw new Error("Workspace creation failed");
+      const removed = await admin.createAgent({
+        provider: "codex",
+        cwd: daemon.staticDir,
+        workspaceId: workspace.id,
+      });
+      const survivor = await admin.createAgent({
+        provider: "codex",
+        cwd: daemon.staticDir,
+        workspaceId: workspace.id,
+      });
+      const sentinel = path.join(daemon.staticDir, "keep-project-file.txt");
+      await writeFile(sentinel, "keep this project file\n");
+      await prepare(daemon.daemon.agentManager, removed.id);
+      const requester = await SubscriptionPeer.connect(daemon.port, "deletion-origin");
+      const sibling = await SubscriptionPeer.connect(daemon.port, "deletion-origin");
+      const observer = await SubscriptionPeer.connect(daemon.port, "deletion-observer");
+      const legacy = await SubscriptionPeer.connect(daemon.port, "deletion-legacy", {
+        owned_subscriptions: false,
+      });
+      peers.push(requester, sibling, observer, legacy);
+      for (const [index, peer] of peers.entries()) {
+        await peer.request({
+          type: "fetch_agents_request",
+          requestId: `directory-${index}`,
+          subscribe: {},
+        });
+        const entries = peer.frames.flatMap((frame) =>
+          frame.type === "session" && frame.message.type === "fetch_agents_response"
+            ? frame.message.payload.entries.map((entry) => entry.agent.id)
+            : [],
+        );
+        expect(entries.sort()).toEqual([removed.id, survivor.id].sort());
+        peer.frames.length = 0;
+      }
+      await requester.request({
+        type: "delete_agent_request",
+        requestId: "delete-observed-agent",
+        agentId: removed.id,
+      });
+      await expect
+        .poll(() => peers.map(removedAgentIds))
+        .toEqual([[removed.id], [removed.id], [removed.id], [removed.id]]);
+      const replies = peers.map((peer) =>
+        peer.frames.flatMap((frame) =>
+          frame.type === "session" && frame.message.type === "agent_deleted"
+            ? [frame.message.payload]
+            : [],
+        ),
+      );
+      expect(replies).toEqual([
+        [{ agentId: removed.id, requestId: "delete-observed-agent" }],
+        [],
+        [],
+        [],
+      ]);
+      expect(await daemon.daemon.agentStorage.get(removed.id)).toBeNull();
+      expect((await daemon.daemon.agentStorage.get(survivor.id))?.id).toBe(survivor.id);
+      expect(await readFile(sentinel, "utf8")).toBe("keep this project file\n");
+    } finally {
+      for (const peer of peers) peer.close();
+      await admin.close();
+      await daemon.close();
+    }
+  },
+);
+
 test("workspace filters and identical queries coexist until their own release", async () => {
   const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
   const admin = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.8.0" });
@@ -679,7 +783,8 @@ test("same-workspace terminal directory observers do not share a teardown slot",
     peer.frames.length = 0;
     const result = await admin.createTerminal(daemon.staticDir, "Observed terminal", undefined, {
       workspaceId: workspace.id,
-      command: "/bin/cat",
+      command: process.execPath,
+      args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"],
     });
     expect(result.error).toBe(null);
     await expect
@@ -728,7 +833,8 @@ test("terminal output has separate server IDs and binary slots, including in a s
     ).workspace!;
     const result = await admin.createTerminal(daemon.staticDir, "Output ownership", undefined, {
       workspaceId: workspace.id,
-      command: "/bin/cat",
+      command: process.execPath,
+      args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"],
     });
     expect(result.error).toBe(null);
     const idle = await SubscriptionPeer.connect(daemon.port, "terminal-output-session");
@@ -1423,7 +1529,7 @@ test("request outcomes for import and attention reach only their requesting sock
   });
   const requester = await SubscriptionPeer.connect(daemon.port, "reply-shared");
   const sibling = await SubscriptionPeer.connect(daemon.port, "reply-shared");
-  const admin = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  const admin = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.8.0" });
   try {
     await admin.connect();
     await requester.request({
@@ -1462,7 +1568,7 @@ test("request outcomes for import and attention reach only their requesting sock
       }),
     );
     const agent = await admin.createAgent({
-      provider: "codex",
+      provider: "mock",
       cwd: daemon.staticDir,
       title: "Attention",
     });

packages/server/src/server/session.ts

@@ -1992,6 +1992,24 @@ export class Session {
           return;
         }
 
+        if (event.type === "agent_removed") {
+          if (event.internal) return;
+          void this.agentUpdates
+            .removeAgent(event.agentId)
+            .then(() => {
+              if (event.workspaceId)
+                return this.emitWorkspaceUpdateForWorkspaceId(event.workspaceId);
+              return undefined;
+            })
+            .catch((error) => {
+              this.sessionLogger.error(
+                { err: error, agentId: event.agentId },
+                "Failed to emit agent removal",
+              );
+            });
+          return;
+        }
+
         if (event.type === "agent_state") {
           this.sessionLogger.trace(
             {
@@ -3337,10 +3355,8 @@ export class Session {
   private async handleDeleteAgentRequest(agentId: string, requestId: string): Promise<void> {
     this.sessionLogger.info({ agentId }, `Deleting agent ${agentId} from registry`);
 
-    const knownWorkspaceId =
-      this.agentManager.getAgent(agentId)?.workspaceId ??
-      (await this.agentStorage.get(agentId))?.workspaceId ??
-      null;
+    const knownAgent =
+      this.agentManager.getAgent(agentId) ?? (await this.agentStorage.get(agentId));
 
     // File-backed storage still needs an early delete fence before closeAgent().
     beginAgentDeleteIfSupported(this.agentStorage, agentId);
@@ -3361,6 +3377,8 @@ export class Session {
     try {
       await this.agentStorage.remove(agentId);
       await this.agentManager.deleteAgentState(agentId);
+      // Directory removal is daemon-wide; agent_deleted below stays a source-scoped RPC reply.
+      if (knownAgent) this.agentManager.notifyAgentRemoved(knownAgent);
     } catch (error) {
       this.sessionLogger.error({ err: error, agentId }, `Failed to fully delete agent ${agentId}`);
     }
@@ -3372,12 +3390,6 @@ export class Session {
         requestId,
       },
     });
-
-    await this.agentUpdates.removeAgent(agentId);
-
-    if (knownWorkspaceId) {
-      await this.emitWorkspaceUpdateForWorkspaceId(knownWorkspaceId);
-    }
   }
 
   private async handleArchiveAgentRequest(agentId: string, requestId: string): Promise<void> {