From 4351445467e2af8a103c366e353eff4d0e4b3abc Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Fri, 30 Oct 2026 00:05:34 +0000 Subject: [PATCH] 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] --- packages/server/package.json | 2 +- .../src/server/agent/agent-manager.test.ts | 54 +++++++ .../server/src/server/agent/agent-manager.ts | 13 +- .../server/owned-subscriptions.e2e.test.ts | 134 ++++++++++++++++-- packages/server/src/server/session.ts | 32 +++-- 5 files changed, 209 insertions(+), 26 deletions(-) diff --git a/packages/server/package.json b/packages/server/package.json index 90e2ccace8..13e5c9b49c 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -89,7 +89,7 @@ "speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts", "test": "npm run test:unit && npm run test:integration", "test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"", - "test:integration": "vitest run --maxWorkers=1 src/server/git-filename-diffs.e2e.test.ts src/server/daemon-e2e/checkout-diff-content.e2e.test.ts src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/terminal-worker-lifecycle.e2e.test.ts src/server/workspace-create-worktree-source.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/daemon-e2e/oversized-message.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts src/server/plugins/lifecycle.e2e.test.ts src/server/connected-hosts/hosts.e2e.test.ts && vitest run --maxWorkers=1 src/server/agent/providers/omp/history.e2e.test.ts src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t \"merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts|password-protected daemon refuses|replays selected OMP history\"", + "test:integration": "vitest run --maxWorkers=1 src/server/git-filename-diffs.e2e.test.ts src/server/daemon-e2e/checkout-diff-content.e2e.test.ts src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/terminal-worker-lifecycle.e2e.test.ts src/server/workspace-create-worktree-source.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/daemon-e2e/oversized-message.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts src/server/plugins/lifecycle.e2e.test.ts src/server/connected-hosts/hosts.e2e.test.ts src/server/owned-subscriptions.e2e.test.ts && vitest run --maxWorkers=1 src/server/agent/providers/omp/history.e2e.test.ts src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t \"merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts|password-protected daemon refuses|replays selected OMP history\"", "test:hub-cli-contract": "vitest run --maxWorkers=1 src/server/hub/hub-cli-contract.test.ts", "test:integration:all": "npm run test:e2e", "test:integration:real": "vitest run real.e2e.test.ts", diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index bb521abc71..902fd79ebc 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/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-")); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index ae9e424d78..a761e723c1 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/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): void { + this.dispatch({ + type: "agent_removed", + agentId: agent.id, + workspaceId: agent.workspaceId, + internal: agent.internal === true, + }); + } + async deleteCommittedTimeline(agentId: string): Promise { 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 = diff --git a/packages/server/src/server/owned-subscriptions.e2e.test.ts b/packages/server/src/server/owned-subscriptions.e2e.test.ts index 68437daf70..2a3128ea9f 100644 --- a/packages/server/src/server/owned-subscriptions.e2e.test.ts +++ b/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): Promise { 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", }); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 3eac37cb7b..cca0b4f3fa 100644 --- a/packages/server/src/server/session.ts +++ b/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 { 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 {