diff --git a/packages/server/src/server/agent/agent-prompt.test.ts b/packages/server/src/server/agent/agent-prompt.test.ts index 5a92da7bf7..dcdfdb1cbf 100644 --- a/packages/server/src/server/agent/agent-prompt.test.ts +++ b/packages/server/src/server/agent/agent-prompt.test.ts @@ -222,6 +222,11 @@ function createFinishNotificationScenario( type: "agent_state", agent: childAgent, }); + subscriber?.({ + type: "agent_stream", + agentId: "child-agent", + event: { type: "turn_completed", provider: "codex" }, + }); }, async finishChildAndReadParentPrompt() { const parentPrompt = new Promise((resolve) => { @@ -540,6 +545,11 @@ it("does not notify archived callers", async () => { type: "agent_state", agent: childAgent, }); + subscriber?.({ + type: "agent_stream", + agentId: "child-agent", + event: { type: "turn_completed", provider: "codex" }, + }); await vi.waitFor(() => { expect(agentStorageGetSpy).toHaveBeenCalledWith("caller-agent"); diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index aad25a79b8..4044fd84bc 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -407,13 +407,20 @@ export interface SetupFinishNotificationParams { logger: Logger; } -type FinishNotificationReason = "finished" | "errored" | "needs permission" | "was closed"; +type FinishNotificationReason = + | "finished" + | "errored" + | "needs permission" + | "was closed" + | "was canceled"; const finishNotificationEvents = { finished: "finished", errored: "errored", "needs permission": "permission-required", "was closed": "closed", + // Use the existing failure event so older clients still parse agent-message metadata. + "was canceled": "errored", } as const; const FINISH_NOTIFICATION_MESSAGE_LIMIT = 4000; @@ -568,10 +575,8 @@ export function setupFinishNotification(params: SetupFinishNotificationParams): notifySafely("errored"); return; } - if (agent.lifecycle === "idle" && hasSeenRunning) { - notifySafely("finished"); - return; - } + // Idle is published before the terminal stream event for both completion + // and cancellation. Let that event identify the outcome instead. if (agent.lifecycle === "closed") { notifySafely("was closed"); return; @@ -600,6 +605,18 @@ export function setupFinishNotification(params: SetupFinishNotificationParams): return; } + if (event.event.type === "turn_completed" || event.event.type === "turn_canceled") { + // Replacing a turn keeps the child running through the displaced turn's + // cancellation. The task settles only when the child becomes idle. + if (agentManager.getAgent(childAgentId)?.lifecycle !== "idle") return; + if (event.event.type === "turn_completed" && hasSeenRunning) { + notifySafely("finished"); + } else if (event.event.type === "turn_canceled" && hasSeenTurn) { + notifySafely("was canceled"); + } + return; + } + if (event.event.type === "permission_requested") { // A permission pause is an intermediate checkpoint. Forget the run // observed before it so an idle state during follow-up startup cannot 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 6a5b360ca8..bfb6be7964 100644 --- a/packages/server/src/server/agent/mcp-parity.e2e.test.ts +++ b/packages/server/src/server/agent/mcp-parity.e2e.test.ts @@ -12,6 +12,8 @@ import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo import { createTestAgentClients } from "../test-utils/fake-agent-client.js"; import type { AgentClient, AgentProvider, AgentSessionConfig } from "./agent-sdk-types.js"; import { PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels"; +import { DaemonClient } from "../test-utils/daemon-client.js"; +import { parseAgentMessage } from "./agent-messages/index.js"; interface StructuredContent { [key: string]: unknown; @@ -311,6 +313,167 @@ afterAll(async () => { }); describe("Suite A: Core Fixes", () => { + test.each([ + ["Stop", "cancelAgent"], + ["Reload", "refreshAgent"], + ] as const)( + "%s tells the parent that a watched child was canceled", + async (_action, method) => { + let releaseChild!: () => void; + let childReachedGate!: () => void; + const heldTurn = new Promise((resolve) => { + releaseChild = resolve; + }); + const childAtGate = new Promise((resolve) => { + childReachedGate = resolve; + }); + const parentPrompts: string[] = []; + const daemon = await createTestPaseoDaemon({ + agentClients: createTestAgentClients({ + onStartTurn(prompt, config) { + if (config.title === "Watching parent" && typeof prompt === "string") + parentPrompts.push(prompt); + }, + async beforeTurnComplete(prompt) { + if ( + typeof prompt === "string" && + parseAgentMessage(prompt).text === "Hold this child turn" + ) { + childReachedGate(); + await heldTurn; + } + }, + }), + }); + const client = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + appVersion: "0.11.2", + }); + let mcp: McpClient | undefined; + let unsubscribe: (() => void) | undefined; + try { + await client.connect(); + await client.fetchAgents({ subscribe: {} }); + const parent = await client.createAgent({ + provider: "claude", + cwd: daemon.paseoHome, + title: "Watching parent", + modeId: "bypassPermissions", + }); + mcp = await createMcpClient( + buildExpectedAgentMcpUrl({ host: "127.0.0.1", port: daemon.port, agentId: parent.id }), + ); + const created = await callToolStructured(mcp, "create_agent", { + relationship: { kind: "subagent" }, + workspace: { kind: "current" }, + title: "Watched child", + provider: "codex/gpt-5.4", + initialPrompt: "Hold this child turn", + notifyOnFinish: true, + settings: { modeId: "full-access" }, + }); + const childId = str(created.agentId); + await childAtGate; + expect(daemon.daemon.agentManager.getAgent(childId)?.lifecycle).toBe("running"); + const terminalEvents: string[] = []; + unsubscribe = daemon.daemon.agentManager.subscribe( + (event) => { + if ( + event.type === "agent_stream" && + event.agentId === childId && + (event.event.type === "turn_completed" || event.event.type === "turn_canceled") + ) + terminalEvents.push(event.event.type); + }, + { agentId: childId, replayState: false }, + ); + + await client[method](childId); + await waitFor({ + timeoutMs: 10000, + label: "parent cancellation notification", + check: () => (parentPrompts.length === 1 ? true : null), + }); + const notification = parseAgentMessage(parentPrompts[0]); + expect(terminalEvents).toEqual(["turn_canceled"]); + expect(notification.source).toEqual({ + kind: "agent-notification", + agentId: childId, + title: "Watched child", + event: "errored", + }); + expect(notification.text).toContain(`Agent ${childId} (Watched child) was canceled.`); + expect(notification.text).not.toContain("finished."); + await waitFor({ + timeoutMs: 10000, + label: "parent notification turn drained", + check: () => + daemon.daemon.agentManager.getAgent(parent.id)?.lifecycle === "idle" ? true : null, + }); + expect(parentPrompts).toHaveLength(1); + } finally { + releaseChild(); + unsubscribe?.(); + await mcp?.close(); + await client.close(); + await daemon.close(); + } + }, + 30000, + ); + + test.each([ + { outcome: "finished", prompt: "say done and stop", status: "idle" }, + { outcome: "errored", prompt: "Emit a turn failure", status: "error" }, + ])( + "a watched child still reports $outcome after send_agent_prompt", + async ({ outcome, prompt, status }) => { + const childId = await createChildAgent({ + title: "Follow-up child", + settings: { modeId: "bypassPermissions" }, + }); + try { + await waitFor({ + timeoutMs: 10000, + label: "initial child turn completed", + check: () => + daemonHandle.daemon.agentManager.getAgent(childId)?.lifecycle === "idle" ? true : null, + }); + await callToolStructured(agentScopedClient, "send_agent_prompt", { + agentId: childId, + prompt, + notifyOnFinish: true, + }); + const notifications = await waitFor({ + timeoutMs: 10000, + label: `${outcome} notification`, + check: () => { + const items = daemonHandle.daemon.agentManager + .getTimeline(parentAgentId) + .filter( + (item) => item.type === "tool_call" && item.agentMessage?.sender.id === childId, + ); + return items.length > 0 ? items : null; + }, + }); + expect(notifications).toEqual([ + expect.objectContaining({ + type: "tool_call", + agentMessage: { + event: outcome, + sender: { id: childId, title: "Follow-up child" }, + text: expect.stringContaining(`(Follow-up child) ${outcome}.`), + }, + }), + ]); + expect(daemonHandle.daemon.agentManager.getAgent(childId)?.lifecycle).toBe(status); + } finally { + await archiveAgentIfPresent(childId); + } + }, + 30000, + ); + test("AGENT_WAIT_TIMEOUT_MS is 30000", () => { expect(AGENT_WAIT_TIMEOUT_MS).toBe(30_000); }); diff --git a/packages/server/src/server/agent/mcp-server.test.ts b/packages/server/src/server/agent/mcp-server.test.ts index 70b49e304b..f1311d31e2 100644 --- a/packages/server/src/server/agent/mcp-server.test.ts +++ b/packages/server/src/server/agent/mcp-server.test.ts @@ -4359,6 +4359,10 @@ describe("send_agent_prompt MCP tool", () => { await vi.waitFor(() => expect(finishNotifications()).not.toHaveLength(0)); await new Promise((resolve) => setTimeout(resolve, 100)); expect(finishNotifications()).toHaveLength(1); + expect(parentClient.sessions[0]!.prompts).toHaveLength(1); + expect(parseAgentMessage(parentClient.sessions[0]!.prompts[0])).toMatchObject({ + source: { kind: "agent-notification", agentId: childId, event: "finished" }, + }); } finally { await removeAgentStateDir(agentManager, storage, workdir); }