diff --git a/packages/server/src/server/agent/host-operations/index.ts b/packages/server/src/server/agent/host-operations/index.ts index 2242c3b872..25f1893bdd 100644 --- a/packages/server/src/server/agent/host-operations/index.ts +++ b/packages/server/src/server/agent/host-operations/index.ts @@ -399,14 +399,16 @@ function remoteObservation(client: DaemonClient): HostObservation { // A new turn may start between the snapshot and the current-response read. // Bound that wait and refresh state instead of attributing the old event to it. const current = await client.waitForFinish(agentId, 100); - if (current.status === "timeout") - snapshot = (await client.fetchAgent(agentId))?.agent ?? null; - else { - if (!current.final && current.error && current.error !== `Agent not found: ${agentId}`) - throw new Error(current.error); - snapshot = current.final; - lastMessage = current.lastMessage; + if (current.status === "timeout") { + // Queued delivery can leave a closed/error snapshot visible before the new turn. + // A timed-out wait has not established an outcome for that queued work. + if (event?.type === "permission_resolved") handler(snapshot, event); + return; } + if (!current.final && current.error && current.error !== `Agent not found: ${agentId}`) + throw new Error(current.error); + snapshot = current.final; + lastMessage = current.lastMessage; } handler(snapshot, event, lastMessage); }; diff --git a/packages/server/src/server/agent/mcp-parity.e2e.test.ts b/packages/server/src/server/agent/mcp-parity.e2e.test.ts index efb14e6d31..cddecc970c 100644 --- a/packages/server/src/server/agent/mcp-parity.e2e.test.ts +++ b/packages/server/src/server/agent/mcp-parity.e2e.test.ts @@ -404,6 +404,7 @@ describe("Suite A: Core Fixes", () => { expect(parseAgentMessage(callerPrompts[0]).source).toEqual({ kind: "agent-notification", agentId: childId, + serverId: daemon.daemon.getServerId(), title: "Notification child", event: "finished", }); @@ -517,6 +518,7 @@ describe("Suite A: Core Fixes", () => { expect(notification.source).toEqual({ kind: "agent-notification", agentId: childId, + serverId: daemon.daemon.getServerId(), title: "Watched child", event: "errored", }); @@ -579,7 +581,11 @@ describe("Suite A: Core Fixes", () => { type: "tool_call", agentMessage: { event: outcome, - sender: { id: childId, title: "Follow-up child" }, + sender: { + id: childId, + serverId: daemonHandle.daemon.getServerId(), + title: "Follow-up child", + }, text: expect.stringContaining(`(Follow-up child) ${outcome}.`), }, }), diff --git a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts index ec47461386..66e30a0b65 100644 --- a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts +++ b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts @@ -2357,3 +2357,74 @@ test("a pre-write host budget rejection leaves sibling routes connected", async true, ]); }, 30000); + +test.each([false, true])( + "remote prompt to a closed agent (archived=%s) reports its new turn instead of stale closure", + { timeout: 30000 }, + async (archived) => { + const parentPrompts: string[] = []; + const started = Promise.withResolvers(); + const finish = Promise.withResolvers(); + const a = await daemon({ + agentClients: createTestAgentClients({ + onStartTurn: (prompt) => parentPrompts.push(String(prompt)), + }), + }); + const b = await daemon({ + agentClients: createTestAgentClients({ + onStartTurn: () => started.resolve(), + beforeTurnComplete: () => finish.promise, + }), + }); + cleanup.push(async () => finish.resolve()); + const origin = await api(a); + const destination = await api(b); + await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` }); + const remoteId = (await online(origin.paseo)).hosts[0]!.serverId!; + const workspace = await origin.paseo.workspaces.create({ + source: { kind: "directory", path: a.staticDir }, + }); + const remoteWorkspace = await destination.paseo.workspaces.create({ + source: { kind: "directory", path: b.staticDir }, + }); + const parent = await workspace.agents.create({ + config: { provider: "claude/claude-test", modeId: "full-access" }, + title: "Closed target caller", + }); + const child = await remoteWorkspace.agents.create({ + config: { provider: "claude/claude-test", modeId: "full-access" }, + title: "Closed target", + }); + if (archived) await child.archive(); + else await b.daemon.agentManager.closeAgent(child.id); + expect((await destination.client.fetchAgent(child.id))?.agent.status).toBe("closed"); + const mcp = new Client({ name: "closed-target-caller", version: "1" }); + await mcp.connect( + new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${a.port}/mcp/agents?callerAgentId=${parent.id}`), + ), + ); + cleanup.push(() => mcp.close()); + const result = await mcp.callTool({ + name: "send_agent_prompt", + arguments: { serverId: remoteId, agentId: child.id, prompt: "Finish the new task" }, + }); + expect(result.isError).not.toBe(true); + expect(result.structuredContent).toMatchObject({ deliveryStatus: "queued", status: "closed" }); + await started.promise; + expect(parentPrompts).toEqual([]); + finish.resolve(); + await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(1); + expect(parentPrompts[0]).toContain('event="finished"'); + expect(parentPrompts[0]).not.toContain('event="closed"'); + expect((await destination.client.fetchAgent(child.id))?.agent).toMatchObject({ + status: "idle", + archivedAt: null, + }); + await expect + .poll(async () => + JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")), + ) + .toEqual([]); + }, +);