From 21e7ccb036dce72c313114121e3ad82fdd917b15 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 23 Oct 2026 13:21:29 +0700 Subject: [PATCH] Restore agent finish notifications after daemon restart (#7683) * fix(server): restore agent finish notifications after daemon restart Finish observers lived only in memory, so a restarted daemon silently lost the wakeup parents relied on. Persist subscription intent through notification dispatch acknowledgment and reconcile local and remote children after bootstrap. * fix(server): reject finish subscriptions for unsent remote prompts Connection and snapshot preflight failures happen before remote prompt transmission. Reject those candidates while retaining ambiguous sends, and preserve the host identity in callbacks. Exercise local permission cycles and storage errors through MCP. * fix(server): notify parents when remote finish observation is exhausted Three failed observations or a removed registration consume the watch after the parent accepts dispatch. Report unknown child status, retain the captured outcome through delivery failures, and keep connecting attempts outside the failure budget. * fix(server): describe exhausted child observation accurately Remote RPC failures do not prove the host is offline. Report the failed observation and unknown child status, and give Windows integration polling more margin. --- docs/agent-lifecycle.md | 25 +- docs/connected-hosts.md | 23 +- .../src/server/agent/agent-manager.test.ts | 14 + .../server/src/server/agent/agent-manager.ts | 6 +- .../src/server/agent/agent-prompt.test.ts | 554 +----------------- .../server/src/server/agent/agent-prompt.ts | 426 +------------- .../src/server/agent/create-agent/create.ts | 46 +- .../finish-subscriptions.test.ts | 414 +++++++++++++ .../agent/finish-subscriptions/index.ts | 363 ++++++++++++ .../internal/json-store.test.ts | 51 ++ .../internal/json-store.ts | 58 ++ .../internal/notification.ts | 70 +++ .../internal/observation-buffer.ts | 83 +++ .../internal/observation-stream.test.ts | 217 +++++++ .../internal/observation-stream.ts | 121 ++++ .../internal/observation.ts | 103 ++++ .../src/server/agent/host-operations/index.ts | 285 ++++++--- .../src/server/agent/mcp-server.test.ts | 212 ++++++- .../src/server/agent/tools/paseo-tools.ts | 226 ++++--- packages/server/src/server/bootstrap.ts | 36 ++ .../server/connected-hosts/hosts.e2e.test.ts | 317 ++++++++++ .../daemon-restart-resume.e2e.test.ts | 156 +++++ packages/server/src/server/session.ts | 2 +- .../src/server/session.workspaces.test.ts | 1 + .../test-utils/finish-subscription-daemon.ts | 40 ++ 25 files changed, 2617 insertions(+), 1232 deletions(-) create mode 100644 packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts create mode 100644 packages/server/src/server/agent/finish-subscriptions/index.ts create mode 100644 packages/server/src/server/agent/finish-subscriptions/internal/json-store.test.ts create mode 100644 packages/server/src/server/agent/finish-subscriptions/internal/json-store.ts create mode 100644 packages/server/src/server/agent/finish-subscriptions/internal/notification.ts create mode 100644 packages/server/src/server/agent/finish-subscriptions/internal/observation-buffer.ts create mode 100644 packages/server/src/server/agent/finish-subscriptions/internal/observation-stream.test.ts create mode 100644 packages/server/src/server/agent/finish-subscriptions/internal/observation-stream.ts create mode 100644 packages/server/src/server/agent/finish-subscriptions/internal/observation.ts create mode 100644 packages/server/src/server/test-utils/finish-subscription-daemon.ts diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index ffcec92ddd..535a6dd712 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -79,7 +79,30 @@ Users can also detach an existing subagent from the subagents track. Detach is d Agent-originated messages and automatic notifications share a recipient-daemon delivery module. A queued receipt acknowledges admission before provider dispatch, allowing a child to report and finish in the same two-second quiet window. Human prompts dispatch directly. Conflicting explicit interruption or session-mode choices retain separate ordered dispatches. -The pending window is in memory. Clean shutdown flushes it before closing agents; an abrupt crash can lose it. A durable send receipt remains pending until actual dispatch acknowledgment, so a lost window never becomes a completed receipt. Completion waits cover queued admission within their original timeout and cancellation scope. +The pending message window is in memory. Clean shutdown flushes it before closing agents; an abrupt +crash can lose queued prompts. A durable send receipt remains pending until actual dispatch +acknowledgment, so a lost window never becomes a completed receipt. Completion waits cover queued +admission within their original timeout and cancellation scope. + +Finish notification intent has a separate daemon-owned lifetime. The finish-subscriptions module +persists caller, child, host, and ownership policy through an injectable store; its JSON adapter uses +`$PASEO_HOME/finish-subscriptions.json`. Registration must commit before creation or queued-prompt +acknowledgment promises a notification. A pending replacement retains its predecessor until dispatch +accepts; rejected dispatch removes only the candidate. Recovery coalesces registrations for the same +caller, child, and host into one watch. + +Bootstrap restores these watches after repairing stale running records to `closed` and before opening +ingress. Reading a closed child's result does not load its provider. A finish notification reports that +the child's run has settled; the parent decides whether its task is complete. Recovery never resumes +child work. Archived or deleted parents suppress delivery; parent-owned creation watches also stop +reporting after detachment. + +Intent remains stored until the parent's notification dispatch is acknowledged. Dispatch failure +retries the captured notification; it does not change the child's reported outcome. Remote host loss +follows the [connected-host observation policy](connected-hosts.md#delegating-agents). Shutdown stops observation before flushing message delivery, retaining +outstanding intent. A crash after successful dispatch but before removal commits can repeat the +notification. Clean shutdown can also flush an already queued callback while retaining its intent, +so that callback can repeat after restart. This is an at-least-once wakeup, without turn attribution or exactly-once delivery. Permission requests are notification checkpoints, not the end of that subscription. The caller is notified again after a permission response when the child finishes, errors, or requests another permission. The permission notification includes the normalized request plus the child and request IDs, so the caller can inspect it and respond without fetching agent status. diff --git a/docs/connected-hosts.md b/docs/connected-hosts.md index 1844456007..20c86c1295 100644 --- a/docs/connected-hosts.md +++ b/docs/connected-hosts.md @@ -44,10 +44,25 @@ connection candidates, including direct connections and routes discovered throug Adding a direct connection to a discovered host adds another candidate. If no candidate connects, the destination workspace keeps its existing disconnected or reconnecting behavior. -Finish notifications use the existing idle wait, including a pending initial run or an already idle -child. They name the server and agent. A daemon restart or route loss ends an outstanding notification -wait; inspect or wait for the child again rather than resending its prompt. Concurrent prompts gain -no additional turn attribution or delivery guarantee. +Finish notifications name the destination server and child agent. The caller's daemon owns durable +subscription intent and restores observation through the same routed wait and permission observation +used during ordinary operation. Restarting the caller's daemon does not imply that a remote child +stopped. Three failed remote observation attempts end the subscription with a notification that the +child could not be observed and its status is unknown. This does not prove the host is offline: remote +RPC failures also prevent observation. Attempts use capped backoff; a connection still +in progress does not spend that budget. A confirmed removed registration reports host unavailability +immediately through the same path. Neither notice claims the child finished. Intent remains stored +until the parent accepts dispatch, then is consumed; reconnecting does not restore that watch. The +parent decides whether to inspect the child or delegate again. + +A definitely unsent prompt returns a tool error and creates no notification. An ambiguous prompt +response retains intent because the destination may have admitted it; observation then follows the +same retry policy. A confirmed missing child reports unavailability. + +Remote creation and local subscription persistence are separate commits. A failed local write reports +failure even if the destination already created the child; inspect it before retrying creation. +See [finish notification lifetime](agent-lifecycle.md#relationships) for registration, acknowledgment, +and duplicate-delivery semantics. Concurrent prompts gain no turn attribution. ## Connections and recovery diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index ef5b361962..c01d477004 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -12069,3 +12069,17 @@ test.each(["live", "history", "accepted"] as const)( } }, ); + +test("reads the last durable assistant response without loading a stored-only child", async () => { + const store = new RecordingTimelineStore(); + await store.appendCommitted("stored-child", { + type: "assistant_message", + text: "Partial work saved before restart", + }); + const manager = new AgentManager({ clients: {}, durableTimelineStore: store, logger }); + expect(manager.getAgent("stored-child")).toBeNull(); + expect(await manager.getLastAssistantMessage("stored-child")).toBe( + "Partial work saved before restart", + ); + expect(manager.getAgent("stored-child")).toBeNull(); +}); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index de7fe1d986..b39fe782b8 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -3322,11 +3322,9 @@ export class AgentManager { } async getLastAssistantMessage(agentId: string): Promise { - const agent = this.agents.get(agentId); - if (!agent) { - return null; + if (!this.timelineStore.has(agentId)) { + return (await this.durableTimelineStore?.getLastAssistantMessage(agentId)) ?? null; } - return await this.getLastAssistantMessageFromStores(agentId); } diff --git a/packages/server/src/server/agent/agent-prompt.test.ts b/packages/server/src/server/agent/agent-prompt.test.ts index 6ec66d10e0..00e70756de 100644 --- a/packages/server/src/server/agent/agent-prompt.test.ts +++ b/packages/server/src/server/agent/agent-prompt.test.ts @@ -1,15 +1,12 @@ -import { createTestAgentMessages } from "../test-utils/agent-messages.js"; import { formatSystemNotificationPrompt, isSystemInjectedEnvelope, parseAgentMessage, - parseAgentMessages, formatAgentMessage, prepareAgentMessage, projectAgentMessages, } from "./agent-messages/index.js"; -import { expect, it, test, vi } from "vitest"; -import pino, { type Logger } from "pino"; +import { expect, test, vi } from "vitest"; import { randomUUID } from "node:crypto"; import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; @@ -17,9 +14,7 @@ import { join } from "node:path"; import { createTestLogger } from "../../test-utils/test-logger.js"; import { AgentManager } from "./agent-manager.js"; -import { AgentStorage } from "./agent-storage.js"; -import { setupFinishNotification, waitForAgentRunStartWithTimeout } from "./agent-prompt.js"; -import type { AgentManagerEvent, ManagedAgent } from "./agent-manager.js"; +import { waitForAgentRunStartWithTimeout } from "./agent-prompt.js"; import type { AgentClient, AgentRunResult, @@ -27,555 +22,12 @@ import type { AgentStreamEvent, } from "./agent-sdk-types.js"; -interface CapturedLogger { - logger: Logger; - records: Array>; - nextRecord: Promise; -} - -function createCapturedLogger(): CapturedLogger { - const records: Array> = []; - let resolveNextRecord!: () => void; - const nextRecord = new Promise((resolve) => { - resolveNextRecord = resolve; - }); - const logger = pino( - { level: "error" }, - { - write(line: string) { - records.push(JSON.parse(line) as Record); - resolveNextRecord(); - }, - }, - ); - return { logger, records, nextRecord }; -} - -interface FinishNotificationScenarioOptions { - childLastAssistantMessage?: string | null; - childParentAgentId?: string | null; - requireParentOwnership?: boolean; - parentPromptError?: Error; - logger?: Logger; -} - -interface FinishNotificationScenario { - startWatchingChild(): void; - requestChildPermission(requestId?: string): void; - resolveChildPermission(requestId?: string): void; - resolveChildPermissionFromState(requestId?: string): void; - resolveChildPermissionWhileIdle(requestId?: string): void; - finishChild(): void; - finishChildAndReadParentPrompt(): Promise; - closeChildAndReadParentPrompt(): Promise; - parentPrompts(): string[]; - steerAttemptCount(): number; - wasParentPrompted(): boolean; -} - -function createFinishNotificationScenario( - options?: FinishNotificationScenarioOptions, -): FinishNotificationScenario { - let subscriber: ((event: AgentManagerEvent) => void) | null = null; - let resolveParentPrompt: ((prompt: string) => void) | null = null; - let parentPrompted = false; - let steerAttemptCount = 0; - const parentPrompts: string[] = []; - - const childAgent: ManagedAgent = Object.create(null); - Reflect.set(childAgent, "id", "child-agent"); - Reflect.set(childAgent, "lifecycle", "idle"); - Reflect.set(childAgent, "config", { title: "Child Agent" }); - Reflect.set(childAgent, "pendingPermissions", new Map()); - - const callerAgent: ManagedAgent = Object.create(null); - Reflect.set(callerAgent, "id", "caller-agent"); - Reflect.set(callerAgent, "lifecycle", "idle"); - Reflect.set(callerAgent, "config", { title: "Caller Agent" }); - - const agentManager = new AgentManager({ clients: {}, logger: createTestLogger() }); - Reflect.set(agentManager, "getAgent", (agentId: string) => { - if (agentId === "child-agent") { - return childAgent; - } - if (agentId === "caller-agent") { - return callerAgent; - } - return null; - }); - Reflect.set(agentManager, "subscribe", (callback: (event: AgentManagerEvent) => void) => { - subscriber = callback; - return () => { - subscriber = null; - }; - }); - Reflect.set(agentManager, "getLastAssistantMessage", async () => { - return options?.childLastAssistantMessage ?? null; - }); - Reflect.set(agentManager, "tryRunOutOfBand", () => false); - Reflect.set(agentManager, "hasInFlightRun", () => Boolean(options?.parentPromptError)); - Reflect.set(agentManager, "steerOrReplaceActiveTurn", async () => { - steerAttemptCount += 1; - return { status: "inactive" }; - }); - Reflect.set(agentManager, "streamAgent", (_agentId: string, prompt: string) => { - parentPrompted = true; - parentPrompts.push(prompt); - resolveParentPrompt?.(prompt); - return (async function* noop() {})(); - }); - Reflect.set(agentManager, "replaceAgentRun", async (_agentId: string, prompt: string) => { - resolveParentPrompt?.(prompt); - throw options?.parentPromptError; - }); - - const agentStorage: AgentStorage = Object.create(AgentStorage.prototype); - Reflect.set(agentStorage, "get", async (agentId: string) => { - if (agentId === "child-agent") { - const parentAgentId = - options?.childParentAgentId === undefined ? "caller-agent" : options.childParentAgentId; - return { - title: "Child Agent", - labels: parentAgentId ? { "paseo.parent-agent-id": parentAgentId } : {}, - }; - } - return null; - }); - - return { - startWatchingChild() { - setupFinishNotification({ - agentMessages: createTestAgentMessages({ - agentManager, - agentStorage, - logger: options?.logger ?? createTestLogger(), - }), - agentManager, - agentStorage, - childAgentId: "child-agent", - callerAgentId: "caller-agent", - requireParentOwnership: options?.requireParentOwnership, - logger: options?.logger ?? createTestLogger(), - }); - }, - requestChildPermission(requestId = "permission-1") { - childAgent.lifecycle = "running"; - childAgent.pendingPermissions.set(requestId, { - id: requestId, - provider: "claude", - kind: "tool", - name: "Run command", - description: "Write the QA sentinel", - input: { - file_path: "/tmp/permission-qa.txt", - content: "PASEO_PERMISSION_NOTIFY_QA_OK\n", - }, - }); - subscriber?.({ - type: "agent_state", - agent: childAgent, - }); - subscriber?.({ - type: "agent_stream", - agentId: "child-agent", - event: { - type: "permission_requested", - provider: "codex", - request: childAgent.pendingPermissions.get(requestId)!, - }, - }); - }, - resolveChildPermission(requestId = "permission-1") { - childAgent.pendingPermissions.delete(requestId); - subscriber?.({ - type: "agent_stream", - agentId: "child-agent", - event: { - type: "permission_resolved", - provider: "codex", - requestId, - resolution: { behavior: "allow" }, - }, - }); - }, - resolveChildPermissionFromState(requestId = "permission-1") { - childAgent.pendingPermissions.delete(requestId); - subscriber?.({ type: "agent_state", agent: childAgent }); - }, - resolveChildPermissionWhileIdle(requestId = "permission-1") { - childAgent.pendingPermissions.delete(requestId); - childAgent.lifecycle = "idle"; - subscriber?.({ type: "agent_state", agent: childAgent }); - subscriber?.({ - type: "agent_stream", - agentId: "child-agent", - event: { - type: "permission_resolved", - provider: "codex", - requestId, - resolution: { behavior: "allow" }, - }, - }); - }, - finishChild() { - childAgent.lifecycle = "running"; - subscriber?.({ - type: "agent_state", - agent: childAgent, - }); - - childAgent.lifecycle = "idle"; - subscriber?.({ - 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) => { - resolveParentPrompt = resolve; - }); - this.finishChild(); - - return parentPrompt; - }, - async closeChildAndReadParentPrompt() { - const parentPrompt = new Promise((resolve) => { - resolveParentPrompt = resolve; - }); - - childAgent.lifecycle = "running"; - subscriber?.({ - type: "agent_state", - agent: childAgent, - }); - - childAgent.lifecycle = "closed"; - subscriber?.({ - type: "agent_state", - agent: childAgent, - }); - - return parentPrompt; - }, - parentPrompts() { - return parentPrompts; - }, - steerAttemptCount() { - return steerAttemptCount; - }, - wasParentPrompted() { - return parentPrompted; - }, - }; -} - test("isSystemInjectedEnvelope matches the envelope formatSystemNotificationPrompt produces", () => { expect(isSystemInjectedEnvelope(formatSystemNotificationPrompt("child finished"))).toBe(true); expect(isSystemInjectedEnvelope("hello world")).toBe(false); }); -test("finish notifications tell the parent the child's last assistant message", async () => { - const scenario = createFinishNotificationScenario({ - childLastAssistantMessage: "Implemented the cleanup and all checks pass.", - }); - - scenario.startWatchingChild(); - const parentPrompt = await scenario.finishChildAndReadParentPrompt(); - - expect(parseAgentMessage(parentPrompt)).toEqual({ - id: expect.any(String), - source: { - kind: "agent-notification", - agentId: "child-agent", - title: "Child Agent", - event: "finished", - }, - text: "Agent child-agent (Child Agent) finished.\n\n\nImplemented the cleanup and all checks pass.\n", - }); - expect(scenario.steerAttemptCount()).toBe(1); -}); - -test("finish notifications truncate oversized child responses", async () => { - const included = "x".repeat(4000); - const omitted = "TAIL-MARKER".repeat(50); - const scenario = createFinishNotificationScenario({ - childLastAssistantMessage: included + omitted, - }); - - scenario.startWatchingChild(); - const parentPrompt = await scenario.finishChildAndReadParentPrompt(); - - expect(parentPrompt).toContain(included); - expect(parentPrompt).toContain( - `[truncated ${omitted.length} chars; use get_agent_activity for the full response]`, - ); - expect(parentPrompt).not.toContain("TAIL-MARKER"); -}); - -test("closing a watched child notifies the caller", async () => { - const scenario = createFinishNotificationScenario(); - - scenario.startWatchingChild(); - const parentPrompt = await scenario.closeChildAndReadParentPrompt(); - - expect(parseAgentMessage(parentPrompt)).toEqual({ - id: expect.any(String), - source: { - kind: "agent-notification", - agentId: "child-agent", - title: "Child Agent", - event: "closed", - }, - text: "Agent child-agent (Child Agent) was closed.", - }); -}); - -test("finish notifications survive permission responses", async () => { - const scenario = createFinishNotificationScenario(); - - scenario.startWatchingChild(); - scenario.requestChildPermission(); - - await vi.waitFor(() => { - expect(scenario.parentPrompts()).toHaveLength(1); - }); - expect(scenario.parentPrompts()[0]).toContain("needs permission."); - const permissionPayload = parseAgentMessage(scenario.parentPrompts()[0])!.text.match( - /\n([\s\S]+?)\n<\/permission-request>/, - )?.[1]; - expect(permissionPayload).toBeDefined(); - expect(JSON.parse(permissionPayload!)).toEqual({ - agentId: "child-agent", - requestId: "permission-1", - request: { - id: "permission-1", - provider: "claude", - kind: "tool", - name: "Run command", - description: "Write the QA sentinel", - input: { - file_path: "/tmp/permission-qa.txt", - content: "PASEO_PERMISSION_NOTIFY_QA_OK\n", - }, - }, - }); - - scenario.resolveChildPermission(); - scenario.finishChild(); - - await vi.waitFor(() => { - expect(scenario.parentPrompts()).toHaveLength(2); - }); - expect(scenario.parentPrompts()[1]).toContain("finished."); -}); - -test("an idle permission resolution waits for the resumed run to finish", async () => { - const scenario = createFinishNotificationScenario(); - - scenario.startWatchingChild(); - scenario.requestChildPermission(); - await vi.waitFor(() => expect(scenario.parentPrompts()).toHaveLength(1)); - - scenario.resolveChildPermissionWhileIdle(); - scenario.requestChildPermission("permission-2"); - await vi.waitFor(() => expect(scenario.parentPrompts()).toHaveLength(2)); - expect(scenario.parentPrompts().every((prompt) => prompt.includes("needs permission."))).toBe( - true, - ); - - scenario.resolveChildPermission("permission-2"); - scenario.finishChild(); - await vi.waitFor(() => expect(scenario.parentPrompts()).toHaveLength(3)); - expect(scenario.parentPrompts()[2]).toContain("finished."); -}); - -test("finish notifications report every concurrently pending permission", async () => { - const scenario = createFinishNotificationScenario(); - - scenario.startWatchingChild(); - scenario.requestChildPermission("permission-1"); - scenario.requestChildPermission("permission-2"); - - await vi.waitFor(() => expect(scenario.parentPrompts()).toHaveLength(1)); - expect( - scenario - .parentPrompts() - .flatMap((prompt) => parseAgentMessages(prompt)!) - .map((message) => { - const payload = message.text.match( - /\n([\s\S]+?)\n<\/permission-request>/, - )?.[1]; - return JSON.parse(payload!).requestId; - }), - ).toEqual(["permission-1", "permission-2"]); - - scenario.resolveChildPermission("permission-1"); - scenario.resolveChildPermission("permission-2"); - scenario.finishChild(); - - await vi.waitFor(() => expect(scenario.parentPrompts()).toHaveLength(2)); - expect(scenario.parentPrompts()[1]).toContain("finished."); -}); - -test("finish notifications survive repeated permission cycles", async () => { - const scenario = createFinishNotificationScenario(); - - scenario.startWatchingChild(); - scenario.requestChildPermission(); - await vi.waitFor(() => expect(scenario.parentPrompts()).toHaveLength(1)); - scenario.resolveChildPermissionFromState(); - - scenario.requestChildPermission(); - await vi.waitFor(() => expect(scenario.parentPrompts()).toHaveLength(2)); - scenario.resolveChildPermission(); - scenario.finishChild(); - - await vi.waitFor(() => expect(scenario.parentPrompts()).toHaveLength(3)); - expect( - scenario.parentPrompts().map((prompt) => prompt.match(/(needs permission|finished)\./)?.[1]), - ).toEqual(["needs permission", "needs permission", "finished"]); -}); - -test("detaching a child ends its parent-owned finish notification", async () => { - const scenario = createFinishNotificationScenario({ - childParentAgentId: null, - requireParentOwnership: true, - }); - scenario.startWatchingChild(); - scenario.finishChild(); - await new Promise((resolve) => setTimeout(resolve, 0)); - expect(scenario.wasParentPrompted()).toBe(false); -}); - -test("follow-up finish notifications do not require a parent relationship", async () => { - const scenario = createFinishNotificationScenario({ childParentAgentId: "another-agent" }); - - scenario.startWatchingChild(); - const parentPrompt = await scenario.finishChildAndReadParentPrompt(); - - expect(parentPrompt).toContain("Agent child-agent (Child Agent) finished."); -}); - -test("finish notifications log a rejected parent prompt without an unhandled rejection", async () => { - const captured = createCapturedLogger(); - const scenario = createFinishNotificationScenario({ - parentPromptError: new Error("parent provider rejected replacement"), - logger: captured.logger, - }); - - scenario.startWatchingChild(); - await scenario.finishChildAndReadParentPrompt(); - await captured.nextRecord; - - expect(captured.records).toEqual([ - expect.objectContaining({ - msg: "Agent message batch dispatch failed", - agentId: "caller-agent", - messageIds: [expect.any(String)], - err: expect.objectContaining({ message: "parent provider rejected replacement" }), - }), - ]); -}); - -it("does not notify archived callers", async () => { - let subscriber: ((event: AgentManagerEvent) => void) | null = null; - - const childAgent: ManagedAgent = Object.create(null); - Reflect.set(childAgent, "id", "child-agent"); - Reflect.set(childAgent, "lifecycle", "idle"); - Reflect.set(childAgent, "config", { title: "Child Agent" }); - Reflect.set(childAgent, "pendingPermissions", new Map()); - - const callerAgent: ManagedAgent = Object.create(null); - Reflect.set(callerAgent, "id", "caller-agent"); - Reflect.set(callerAgent, "lifecycle", "idle"); - Reflect.set(callerAgent, "config", { title: "Caller Agent" }); - - const streamAgentSpy = vi.fn(() => (async function* noop() {})()); - const replaceAgentRunSpy = vi.fn(() => (async function* noop() {})()); - - const agentManager = new AgentManager({ clients: {}, logger: createTestLogger() }); - Reflect.set( - agentManager, - "getAgent", - vi.fn((agentId: string) => { - if (agentId === "child-agent") { - return childAgent; - } - if (agentId === "caller-agent") { - return callerAgent; - } - return null; - }), - ); - Reflect.set( - agentManager, - "subscribe", - vi.fn((callback: (event: AgentManagerEvent) => void) => { - subscriber = callback; - return () => { - subscriber = null; - }; - }), - ); - Reflect.set(agentManager, "hasInFlightRun", vi.fn().mockReturnValue(false)); - Reflect.set(agentManager, "streamAgent", streamAgentSpy); - Reflect.set(agentManager, "replaceAgentRun", replaceAgentRunSpy); - - const agentStorageGetSpy = vi.fn(async (agentId: string) => - agentId === "caller-agent" ? { archivedAt: "2024-01-01" } : null, - ); - const agentStorage: AgentStorage = Object.create(AgentStorage.prototype); - Reflect.set(agentStorage, "get", agentStorageGetSpy); - - setupFinishNotification({ - agentMessages: createTestAgentMessages({ - agentManager, - agentStorage, - logger: createTestLogger(), - }), - agentManager, - agentStorage, - childAgentId: "child-agent", - callerAgentId: "caller-agent", - logger: createTestLogger(), - }); - - expect(subscriber).not.toBeNull(); - - childAgent.lifecycle = "running"; - subscriber?.({ - type: "agent_state", - agent: childAgent, - }); - - childAgent.lifecycle = "idle"; - subscriber?.({ - 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"); - }); - - expect(streamAgentSpy).not.toHaveBeenCalled(); - expect(replaceAgentRunSpy).not.toHaveBeenCalled(); -}); - -// Deliberately independent literals rather than the production constants these tests -// guard: deriving the boundaries from AGENT_RUN_START_TIMEOUT_MS would keep the tests -// green if that constant were shortened back under a provider's startup budget. +// Independent of the production constant so shortening the budget breaks this test. const EXPECTED_RUN_START_BUDGET_MS = 60_000; // The slowest provider startup budget the run-start wait has to sit outside of today // (OpenCode's OPENCODE_SERVER_STARTUP_TIMEOUT_MS). diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index 63c2309956..a8fd2ebd4b 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -1,24 +1,11 @@ -import type { WaitForFinishResult } from "@getpaseo/client/internal/daemon-client"; -import { - prepareAgentMessage, - type AgentPromptSource, - type AgentMessageDelivery, -} from "./agent-messages/index.js"; +import { prepareAgentMessage, type AgentPromptSource } from "./agent-messages/index.js"; import type { Logger } from "pino"; -import type { - AgentPermissionRequest, - AgentPromptInput, - AgentRunOptions, -} from "./agent-sdk-types.js"; +import type { AgentPromptInput, AgentRunOptions } from "./agent-sdk-types.js"; import type { AgentManager, ManagedAgent } from "./agent-manager.js"; import type { AgentStorage } from "./agent-storage.js"; import { ensureAgentLoaded } from "./agent-loading.js"; import { isStaleProviderSessionError } from "./stale-provider-session-error.js"; -import { - getParentAgentIdFromLabels, - getParentAddressFromLabels, -} from "@getpaseo/protocol/agent-labels"; import type { ActiveTurnBehavior } from "@getpaseo/protocol/messages"; export type AgentUnarchiveController = Pick; @@ -412,412 +399,3 @@ export async function startCreatedAgentInitialPrompt( } return refreshedSnapshot; } - -export interface SetupFinishNotificationParams { - agentMessages: AgentMessageDelivery; - agentManager: AgentManager; - agentStorage: AgentStorage; - childAgentId: string; - callerAgentId: string; - childServerId?: string; - requireParentOwnership?: boolean; - /** Retain pre-dispatch outcomes only when dispatch keeps the existing turn. */ - dispatchResult?: Promise; - /** A replacement must observe its new start, not the displaced running snapshot. */ - waitForTurnStart?: boolean; - logger: Logger; -} - -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; - -interface FinishNotificationBodyInput { - childAgentId: string; - serverId?: string; - title: string; - reason: FinishNotificationReason; - lastAssistantMessage: string | null; - permissionRequest?: AgentPermissionRequest; -} - -function formatFinishNotificationBody(params: FinishNotificationBodyInput): string { - const statusLine = `Agent ${params.childAgentId} (${params.title}) ${params.reason}.`; - const sections = [statusLine]; - if (params.serverId) sections.push(`Server: ${params.serverId}`); - if (params.reason === "needs permission" && params.permissionRequest) { - sections.push( - "Respond with `respond_to_permission` using the `agentId` and `requestId` below.", - `\n${JSON.stringify( - { - agentId: params.childAgentId, - ...(params.serverId ? { serverId: params.serverId } : {}), - requestId: params.permissionRequest.id, - request: params.permissionRequest, - }, - null, - 2, - )}\n`, - ); - } - let lastAssistantMessage = params.lastAssistantMessage?.trim(); - if (lastAssistantMessage) { - if (lastAssistantMessage.length > FINISH_NOTIFICATION_MESSAGE_LIMIT) { - const omitted = lastAssistantMessage.length - FINISH_NOTIFICATION_MESSAGE_LIMIT; - lastAssistantMessage = `${lastAssistantMessage.slice(0, FINISH_NOTIFICATION_MESSAGE_LIMIT)}\n[truncated ${omitted} chars; use get_agent_activity for the full response]`; - } - sections.push(`\n${lastAssistantMessage}\n`); - } - return sections.join("\n\n"); -} - -interface NotifySafelyOptions { - terminal?: boolean; - permissionRequest?: AgentPermissionRequest; -} - -// Pending follow-ups coexist until acknowledgement. A successful dispatch replaces -// every earlier observation for the caller/child pair; a failure removes only itself. -interface ArmedFinishNotification { - stop(): void; - supersedeWhen(accepted: Promise): void; -} - -const armedFinishNotifications = new WeakMap< - AgentManager, - Map> ->(); - -/** Creation waits and accepted follow-ups share notification ownership. */ -export function setupFinishWait(params: { - agentMessages: AgentMessageDelivery; - agentManager: AgentManager; - agentStorage: AgentStorage; - originServerId?: string; - childServerId?: string; - childAgentId: string; - callerAgentId: string; - wait(signal: AbortSignal): Promise; - afterPermission(requestId: string, signal: AbortSignal): Promise; - logger: Logger; -}): () => void { - const controller = new AbortController(); - const byManager = - armedFinishNotifications.get(params.agentManager) ?? - new Map>(); - armedFinishNotifications.set(params.agentManager, byManager); - const key = JSON.stringify([ - params.childServerId ?? null, - params.childAgentId, - params.callerAgentId, - ]); - const registrations = byManager.get(key) ?? new Set(); - byManager.set(key, registrations); - for (const previous of registrations) previous.stop(); - byManager.set(key, registrations); - let superseded = Promise.resolve(false); - const stop = () => { - controller.abort(); - registrations.delete(registration); - if (registrations.size === 0 && byManager.get(key) === registrations) byManager.delete(key); - }; - const registration: ArmedFinishNotification = { - stop, - supersedeWhen(accepted) { - superseded = superseded.then(async (prior) => prior || (await accepted)); - }, - }; - registrations.add(registration); - async function watch(): Promise { - while (!controller.signal.aborted) { - const result = await params.wait(controller.signal); - if (controller.signal.aborted || (await superseded) || result.status === "timeout") return; - const parent = getParentAddressFromLabels( - result.final?.labels, - params.childServerId ?? "self", - ); - if ( - parent?.agentId !== params.callerAgentId || - parent.serverId !== (params.originServerId ?? "self") - ) - return; - if ((await params.agentStorage.get(params.callerAgentId))?.archivedAt) return; - const reason = { idle: "finished", error: "errored", permission: "needs permission" }[ - result.status - ] as FinishNotificationReason; - const permissionRequest = result.final?.pendingPermissions[0]; - params.agentMessages.enqueue({ - recipientId: params.callerAgentId, - prompt: formatFinishNotificationBody({ - childAgentId: params.childAgentId, - serverId: params.childServerId, - title: result.final?.title ?? params.childAgentId, - reason, - lastAssistantMessage: result.lastMessage, - permissionRequest, - }), - source: { - kind: "agent-notification", - agentId: params.childAgentId, - serverId: params.childServerId, - title: result.final?.title ?? params.childAgentId, - event: finishNotificationEvents[reason], - }, - activeTurnBehavior: "steer", - unarchive: false, - }); - if (result.status !== "permission" || !permissionRequest) return; - await params.afterPermission(permissionRequest.id, controller.signal); - } - } - void watch() - .catch((error) => { - if (!controller.signal.aborted) - params.logger.warn( - { err: error, agentId: params.childAgentId }, - "Child finish wait stopped", - ); - }) - .finally(stop); - return stop; -} - -export function setupFinishNotification(params: SetupFinishNotificationParams): () => void { - const { - agentManager, - agentStorage, - childAgentId, - callerAgentId, - requireParentOwnership = false, - waitForTurnStart = false, - logger, - } = params; - let hasSeenRunning = false; - let hasSeenTurn = false; - let stopped = false; - const notifiedPermissionRequestIds = new Set(); - let unsubscribe: (() => void) | null = null; - let notificationQueue = Promise.resolve(); - - const armedByManager = armedFinishNotifications.get(agentManager) ?? new Map(); - armedFinishNotifications.set(agentManager, armedByManager); - const armedKey = JSON.stringify([params.childServerId ?? null, childAgentId, callerAgentId]); - const registrations = armedByManager.get(armedKey) ?? new Set(); - armedByManager.set(armedKey, registrations); - const previous = [...registrations]; - let superseded = Promise.resolve(false); - let dispatchPending = Boolean(params.dispatchResult); - const registration: ArmedFinishNotification = { - stop, - supersedeWhen(accepted) { - superseded = superseded.then( - async (alreadySuperseded) => alreadySuperseded || (await accepted), - ); - }, - }; - registrations.add(registration); - if (params.dispatchResult) { - const dispatchAccepted = params.dispatchResult.then((result) => result !== false); - for (const observer of previous) observer.supersedeWhen(dispatchAccepted); - void params.dispatchResult.then((accepted) => { - dispatchPending = false; - if (accepted) { - for (const observer of previous) observer.stop(); - if (stopped || accepted === "turn_started") stop(); - } else { - stop(); - } - return undefined; - }); - } else { - for (const observer of previous) { - observer.supersedeWhen(Promise.resolve(true)); - observer.stop(); - } - } - - function stop(): void { - stopped = true; - unsubscribe?.(); - // Pending observations remain visible to overlapping sends until acknowledgement. - if (!dispatchPending) { - registrations.delete(registration); - if (registrations.size === 0 && armedByManager.get(armedKey) === registrations) { - armedByManager.delete(armedKey); - } - } - } - - async function notify( - reason: FinishNotificationReason, - permissionRequest?: AgentPermissionRequest, - ): Promise { - if (params.dispatchResult) { - const result = await params.dispatchResult; - if (result === false || result === "turn_started") return; - } - if (await superseded) return; - const callerRecord = await agentStorage.get(callerAgentId); - if (callerRecord?.archivedAt) { - return; - } - - const record = await agentStorage.get(childAgentId); - if (requireParentOwnership && getParentAgentIdFromLabels(record?.labels) !== callerAgentId) { - return; - } - const title = record?.title ?? childAgentId; - const lastAssistantMessage = await agentManager.getLastAssistantMessage(childAgentId); - const body = formatFinishNotificationBody({ - childAgentId, - title, - reason, - lastAssistantMessage, - permissionRequest, - }); - - params.agentMessages.enqueue({ - recipientId: callerAgentId, - prompt: body, - source: { - kind: "agent-notification", - agentId: childAgentId, - title, - event: finishNotificationEvents[reason], - }, - activeTurnBehavior: "steer", - unarchive: false, - }); - } - - function notifySafely(reason: FinishNotificationReason, options: NotifySafelyOptions = {}): void { - if (stopped) return; - if (options.terminal ?? true) stop(); - notificationQueue = notificationQueue - .then(() => notify(reason, options.permissionRequest)) - .catch((error) => { - logger.error( - { err: error, childAgentId, callerAgentId, reason }, - "Failed to notify caller agent", - ); - }); - } - - function observeState(agent: ManagedAgent): void { - for (const requestId of notifiedPermissionRequestIds) { - if (!agent.pendingPermissions.has(requestId)) { - notifiedPermissionRequestIds.delete(requestId); - } - } - if (agent.lifecycle === "running") { - if (waitForTurnStart && !hasSeenTurn) return; - hasSeenTurn = true; - if (agent.pendingPermissions.size === 0) { - hasSeenRunning = true; - } - return; - } - // Prior terminal state is not this task's outcome. A failure before turn start - // is reported by the dispatch acknowledgement instead of a finish callback. - if (agent.lifecycle === "error" && hasSeenTurn) { - notifySafely("errored"); - 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; - } - return; - } - - unsubscribe = agentManager.subscribe( - (event) => { - if (stopped) { - return; - } - - if (event.type === "agent_state") { - observeState(event.agent); - return; - } - - if (event.type === "timeline_replacement") { - return; - } - - if (event.event.type === "turn_started") { - hasSeenTurn = true; - hasSeenRunning = true; - 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 - // masquerade as the final completion. - hasSeenRunning = false; - if (!notifiedPermissionRequestIds.has(event.event.request.id)) { - notifiedPermissionRequestIds.add(event.event.request.id); - notifySafely("needs permission", { - terminal: false, - permissionRequest: event.event.request, - }); - } - return; - } - - if (event.event.type === "permission_resolved") { - notifiedPermissionRequestIds.delete(event.event.requestId); - const childAgent = agentManager.getAgent(childAgentId); - if (childAgent?.pendingPermissions.size === 0) { - hasSeenRunning = childAgent.lifecycle === "running"; - } - } - }, - { agentId: childAgentId, replayState: false }, - ); - - // Check if the child is already running (catches the case where - // the lifecycle flipped before our subscribe call was processed). - // Do NOT treat an immediate "idle" as "finished" — the agent may - // not have started yet (streamAgent sets a pending run before - // transitioning to "running"). - const childSnapshot = agentManager.getAgent(childAgentId); - if (!childSnapshot || childSnapshot.lifecycle === "closed") { - stop(); - return stop; - } - if (childSnapshot.lifecycle === "running" && !waitForTurnStart) { - hasSeenTurn = true; - hasSeenRunning = true; - } - return stop; -} diff --git a/packages/server/src/server/agent/create-agent/create.ts b/packages/server/src/server/agent/create-agent/create.ts index 79f674e389..77cdde673b 100644 --- a/packages/server/src/server/agent/create-agent/create.ts +++ b/packages/server/src/server/agent/create-agent/create.ts @@ -1,4 +1,5 @@ -import type { AgentPromptSource, AgentMessageDelivery } from "../agent-messages/index.js"; +import type { FinishSubscriptions } from "../finish-subscriptions/index.js"; +import type { AgentPromptSource } from "../agent-messages/index.js"; import type { Logger } from "pino"; import type { TerminalManager } from "../../../terminal/terminal-manager.js"; @@ -15,7 +16,7 @@ import type { AgentPromptInput, AgentRunOptions, AgentSessionConfig } from "../a import type { AgentStorage } from "../agent-storage.js"; import type { AgentOwner } from "../agent-owner.js"; import type { ProviderSnapshotManager } from "../provider-snapshot-manager.js"; -import { setupFinishNotification, startCreatedAgentInitialPrompt } from "../agent-prompt.js"; +import { startCreatedAgentInitialPrompt } from "../agent-prompt.js"; import { resolveCreateAgentTitles } from "../create-agent-title.js"; import { buildAgentPrompt } from "../prompt-attachments.js"; import { normalizeClientMessageId, resolveClientMessageId } from "../../client-message-id.js"; @@ -23,7 +24,7 @@ import { resolveRequiredProviderModel, type ResolvedProviderModel } from "../mcp import { resolveCreateAgentIntent } from "./intent.js"; export interface CreateAgentCommandDependencies { - agentMessages?: AgentMessageDelivery; + finishSubscriptions?: FinishSubscriptions; agentManager: AgentManager; agentStorage: AgentStorage; logger: Logger; @@ -175,28 +176,29 @@ export async function createAgentCommand( if (input.kind === "mcp") { input.onCreated?.({ agentId: snapshot.id, createdWorktree: resolved.createdWorktree ?? null }); } - const stopNotification = + const notification = input.kind === "mcp" && input.notifyOnFinish && input.callerAgentId && resolved.prompt !== undefined - ? setupFinishNotification({ - agentMessages: requireAgentMessages(dependencies), - agentManager: dependencies.agentManager, - agentStorage: dependencies.agentStorage, - childAgentId: snapshot.id, - callerAgentId: input.callerAgentId, - // Detached agents have no parent label, but still notify their creator. - requireParentOwnership: !(input.detached ?? false), - logger: dependencies.logger, - }) + ? await requireFinishSubscriptions(dependencies).register( + { + childAgentId: snapshot.id, + callerAgentId: input.callerAgentId, + requireParentOwnership: !(input.detached ?? false), + }, + true, + ) : undefined; if (resolved.prompt !== undefined) { - const sendResult = await sendInitialPrompt(dependencies, resolved, snapshot).catch((error) => { - stopNotification?.(); - throw error; - }); + const sendResult = await sendInitialPrompt(dependencies, resolved, snapshot).catch( + async (error) => { + await notification?.reject(); + throw error; + }, + ); + notification?.accept(); initialPromptStarted = sendResult.started; liveSnapshot = sendResult.liveSnapshot; initialPromptError = sendResult.error ?? null; @@ -561,7 +563,9 @@ async function resolveMcpCwd(params: { }; } -function requireAgentMessages(dependencies: CreateAgentCommandDependencies): AgentMessageDelivery { - if (!dependencies.agentMessages) throw new Error("Agent message delivery is not configured"); - return dependencies.agentMessages; +function requireFinishSubscriptions( + dependencies: CreateAgentCommandDependencies, +): FinishSubscriptions { + if (!dependencies.finishSubscriptions) throw new Error("Finish subscriptions are not configured"); + return dependencies.finishSubscriptions; } diff --git a/packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts b/packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts new file mode 100644 index 0000000000..872961111b --- /dev/null +++ b/packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts @@ -0,0 +1,414 @@ +import { HostObservationError } from "../host-operations/index.js"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, test } from "vitest"; +import type { WaitForFinishResult } from "@getpaseo/client/internal/daemon-client"; +import type { AgentMessageDelivery } from "../agent-messages/index.js"; +import { buildStoredAgentPayload } from "../agent-projections.js"; +import { createTestLogger } from "../../../test-utils/test-logger.js"; +import { + createFinishSubscriptions, + createJsonFinishSubscriptionStore, + type FinishObservation, + type FinishSubscriptions, +} from "./index.js"; + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + return { promise, resolve, reject }; +} +class Observation implements FinishObservation { + waits: Array<{ signal: AbortSignal; result: ReturnType> }> = + []; + permission = deferred(); + opening: Promise = Promise.resolve(); + openedSignals: AbortSignal[] = []; + async open(_record: unknown, signal: AbortSignal) { + this.openedSignals.push(signal); + await this.opening; + return { next: () => this.wait(signal), close: async () => {} }; + } + async wait(signal: AbortSignal) { + signal.throwIfAborted(); + const result = deferred(); + signal.addEventListener("abort", () => result.reject(new Error("Stopped")), { once: true }); + this.waits.push({ signal, result }); + return result.promise; + } + afterPermission() { + return this.permission.promise; + } +} +class Delivery implements AgentMessageDelivery { + prompts: string[] = []; + acknowledgments: Array>> = []; + enqueue(message: Parameters[0]) { + this.prompts.push(String(message.prompt)); + const acknowledgment = deferred<"steered">(); + this.acknowledgments.push(acknowledgment); + return { + status: "queued" as const, + messageId: "notification", + dispatched: acknowledgment.promise, + }; + } + async close() {} + async waitForDispatch() {} +} +const homes: string[] = []; +const services: FinishSubscriptions[] = []; +afterEach(async () => { + for (const service of services.splice(0)) await service.close(); + await Promise.all(homes.splice(0).map((home) => rm(home, { recursive: true, force: true }))); +}); +async function scenario(directory?: string, removalFailures = 0) { + const home = directory ?? (await mkdtemp(join(tmpdir(), "finish-recovery-"))); + if (!directory) homes.push(home); + const durableStore = createJsonFinishSubscriptionStore(home); + const store = { + ...durableStore, + async remove(id: string) { + if (removalFailures > 0) { + removalFailures--; + throw new Error("Subscription removal unavailable"); + } + await durableStore.remove(id); + }, + }; + const observation = new Observation(); + const delivery = new Delivery(); + let active = true; + const service = createFinishSubscriptions({ + store, + messages: () => delivery, + open: observation.open.bind(observation), + callerIsActive: async () => active, + serverId: "origin", + logger: createTestLogger(), + retryDelayMs: 1, + }); + services.push(service); + await service.start(); + return { + home, + store, + observation, + delivery, + service, + archive: () => { + active = false; + }, + }; +} +const intent = { callerAgentId: "parent", childAgentId: "child" }; +function outcome(status: "idle" | "running" | "closed" | "error" = "closed"): WaitForFinishResult { + return { + status: status === "error" ? "error" : "idle", + final: buildStoredAgentPayload( + { + id: "child", + provider: "codex", + cwd: "/tmp", + title: "Child", + createdAt: "2026-01-01T00:00:00Z", + updatedAt: "2026-01-01T00:00:00Z", + labels: { "paseo.parent-agent-id": "parent" }, + lastStatus: status, + }, + ["codex"], + ), + lastMessage: "Partial work saved", + error: null, + }; +} +async function until(check: () => boolean) { + await expect.poll(check).toBe(true); +} + +test("retains intent until parent dispatch acknowledgment, then restart sends nothing", async () => { + const first = await scenario(); + await first.service.register(intent); + await until(() => first.observation.waits.length === 1); + first.observation.waits[0]!.result.resolve(outcome()); + await until(() => first.delivery.prompts.length === 1); + expect(first.delivery.prompts[0]).toContain("was closed"); + expect(first.delivery.prompts[0]).toContain("Partial work saved"); + expect(await first.store.list()).toHaveLength(1); + first.delivery.acknowledgments[0]!.resolve("steered"); + await expect.poll(async () => (await first.store.list()).length).toBe(0); + await first.service.close(); + const restarted = await scenario(first.home); + expect(restarted.observation.waits).toHaveLength(0); +}); + +test("restart coalesces predecessor and queued candidate into one settled notification", async () => { + const first = await scenario(); + await first.service.register(intent); + await first.service.register(intent, true); + expect(await first.store.list()).toHaveLength(2); + await first.service.close(); + const restarted = await scenario(first.home); + await until(() => restarted.observation.waits.length === 1); + restarted.observation.waits[0]!.result.resolve(outcome()); + await until(() => restarted.delivery.prompts.length === 1); + restarted.delivery.acknowledgments[0]!.resolve("steered"); + await expect.poll(async () => (await restarted.store.list()).length).toBe(0); + expect(restarted.delivery.prompts).toHaveLength(1); +}); + +test("rejected queued follow-up preserves the predecessor and its buffered finish", async () => { + const state = await scenario(); + await state.service.register(intent); + await until(() => state.observation.waits.length === 1); + const pending = await state.service.register(intent, true); + state.observation.waits[0]!.result.resolve(outcome("idle")); + await Promise.resolve(); + expect(state.delivery.prompts).toHaveLength(0); + await pending.reject(); + await until(() => state.delivery.prompts.length === 1); + expect(await state.store.list()).toHaveLength(1); +}); + +test("accepted replacement ignores old idle and stale acknowledgment cannot delete new intent", async () => { + const state = await scenario(); + await state.service.register(intent); + await until(() => state.observation.waits.length === 1); + state.observation.waits[0]!.result.resolve(outcome("idle")); + await until(() => state.delivery.prompts.length === 1); + const next = await state.service.register(intent, true); + next.accept(); + await until(() => state.observation.waits.length === 2); + state.delivery.acknowledgments[0]!.resolve("steered"); + await expect.poll(async () => (await state.store.list()).length).toBe(1); + state.observation.waits[1]!.result.resolve(outcome("idle")); + await until(() => state.delivery.prompts.length === 2); +}); + +test("shutdown before dispatch acknowledgment leaves intent for recovery", async () => { + const state = await scenario(); + await state.service.register(intent); + await until(() => state.observation.waits.length === 1); + state.observation.waits[0]!.result.resolve(outcome()); + await until(() => state.delivery.prompts.length === 1); + await state.service.close(); + state.delivery.acknowledgments[0]!.resolve("steered"); + const restarted = await scenario(state.home); + await until(() => restarted.observation.waits.length === 1); + restarted.observation.waits[0]!.result.resolve(outcome()); + await until(() => restarted.delivery.prompts.length === 1); +}); + +test("remote recovery before three failed observations still reports child completion", async () => { + const first = await scenario(); + await first.service.register({ ...intent, serverId: "remote" }); + await first.service.close(); + const state = await scenario(first.home); + await until(() => state.observation.waits.length === 1); + state.observation.waits[0]!.result.reject(new HostObservationError("unreachable")); + await until(() => state.observation.waits.length === 2); + expect(await state.store.list()).toHaveLength(1); + expect(state.delivery.prompts).toHaveLength(0); + state.observation.waits[1]!.result.resolve(outcome("idle")); + await until(() => state.delivery.prompts.length === 1); + expect(state.delivery.prompts[0]).toContain("Server: remote"); +}); + +test("archived parent suppresses delivery and removes obsolete intent", async () => { + const state = await scenario(); + await state.service.register(intent); + await until(() => state.observation.waits.length === 1); + state.archive(); + state.observation.waits[0]!.result.resolve(outcome()); + await expect.poll(async () => (await state.store.list()).length).toBe(0); + expect(state.delivery.prompts).toHaveLength(0); +}); + +test("permission checkpoint retains intent and completion reports after permission clears", async () => { + const state = await scenario(); + await state.service.register(intent); + await until(() => state.observation.waits.length === 1); + const result = outcome("running"); + result.status = "permission"; + result.final!.pendingPermissions = [ + { + id: "permission", + name: "Write", + input: { path: "file" }, + kind: "tool", + } as (typeof result.final.pendingPermissions)[number], + ]; + state.observation.waits[0]!.result.resolve(result); + await until(() => state.delivery.prompts.length === 1); + expect(state.delivery.prompts[0]).toContain("respond_to_permission"); + expect(await state.store.list()).toHaveLength(1); + state.delivery.acknowledgments[0]!.resolve("steered"); + state.observation.permission.resolve(); + await until(() => state.observation.waits.length === 2); + state.observation.waits[1]!.result.resolve(outcome("idle")); + await until(() => state.delivery.prompts.length === 2); +}); + +test("coalesced unconditional follow-up survives detachment of a parent-owned creation", async () => { + const first = await scenario(); + await first.service.register({ ...intent, requireParentOwnership: true }); + await first.service.register(intent, true); + await first.service.close(); + const restarted = await scenario(first.home); + await until(() => restarted.observation.waits.length === 1); + const detached = outcome(); + detached.final!.labels = {}; + restarted.observation.waits[0]!.result.resolve(detached); + await until(() => restarted.delivery.prompts.length === 1); +}); + +test("detached child suppresses a parent-owned creation subscription", async () => { + const state = await scenario(); + await state.service.register({ ...intent, requireParentOwnership: true }); + await until(() => state.observation.waits.length === 1); + const detached = outcome(); + detached.final!.labels = {}; + state.observation.waits[0]!.result.resolve(detached); + await expect.poll(async () => (await state.store.list()).length).toBe(0); + expect(state.delivery.prompts).toHaveLength(0); +}); + +test("dispatch failure retains intent and retries the ordinary finish notification", async () => { + const state = await scenario(); + await state.service.register(intent); + await until(() => state.observation.waits.length === 1); + state.observation.waits[0]!.result.resolve(outcome()); + await until(() => state.delivery.prompts.length === 1); + state.delivery.acknowledgments[0]!.reject(new Error("Provider unavailable")); + await until(() => state.delivery.prompts.length === 2); + expect(await state.store.list()).toHaveLength(1); + expect(state.observation.waits).toHaveLength(1); + state.delivery.acknowledgments[1]!.resolve("steered"); + await expect.poll(async () => (await state.store.list()).length).toBe(0); +}); + +test("error and missing-child notifications never claim successful completion", async () => { + const state = await scenario(); + await state.service.register(intent); + await until(() => state.observation.waits.length === 1); + state.observation.waits[0]!.result.resolve(outcome("error")); + await until(() => state.delivery.prompts.length === 1); + expect(state.delivery.prompts[0]).toContain("errored"); + await state.service.register({ ...intent, childAgentId: "missing-child" }); + await until(() => state.observation.waits.length === 2); + state.observation.waits[1]!.result.resolve({ + status: "error", + final: null, + error: "Agent not found: missing-child", + lastMessage: null, + }); + await until(() => state.delivery.prompts.length === 2); + expect(state.delivery.prompts[1]).toContain("is unavailable"); + expect(state.delivery.prompts[1]).not.toContain("finished"); +}); + +test("finish notification limits a long last response and points to activity for the remainder", async () => { + const state = await scenario(); + await state.service.register(intent); + await until(() => state.observation.waits.length === 1); + const result = outcome("idle"); + result.lastMessage = "x".repeat(4100); + state.observation.waits[0]!.result.resolve(result); + await until(() => state.delivery.prompts.length === 1); + expect(state.delivery.prompts[0]).toContain("x".repeat(4000)); + expect(state.delivery.prompts[0]).not.toContain("x".repeat(4001)); + expect(state.delivery.prompts[0]).toContain("get_agent_activity"); +}); + +test("three failed remote observations consume intent only after unavailable-notification acknowledgment", async () => { + const state = await scenario(); + await state.service.register({ ...intent, serverId: "remote" }); + for (let attempt = 0; attempt < 3; attempt++) { + await until(() => state.observation.waits.length === attempt + 1); + expect(state.delivery.prompts).toEqual([]); + state.observation.waits[attempt]!.result.reject(new HostObservationError("unreachable")); + } + await until(() => state.delivery.prompts.length === 1); + expect(state.delivery.prompts[0]).toContain( + "Could not observe child agent child on host remote after three failed attempts.", + ); + expect(state.delivery.prompts[0]).toContain("Status of child agent child (child) is unknown."); + expect(state.delivery.prompts[0]).not.toContain("finished."); + expect(await state.store.list()).toHaveLength(1); + expect(state.observation.waits).toHaveLength(3); + state.delivery.acknowledgments[0]!.resolve("steered"); + await expect.poll(async () => (await state.store.list()).length).toBe(0); + await state.service.close(); + const restarted = await scenario(state.home); + expect(restarted.observation.waits).toEqual([]); + expect(restarted.delivery.prompts).toEqual([]); +}); + +test("connecting does not spend the remote failure budget and recovery before three failures still completes", async () => { + const state = await scenario(); + await state.service.register({ ...intent, serverId: "remote" }); + for (let attempt = 0; attempt < 4; attempt++) { + await until(() => state.observation.waits.length === attempt + 1); + expect(state.delivery.prompts).toEqual([]); + state.observation.waits[attempt]!.result.reject(new HostObservationError("connecting")); + } + for (let attempt = 4; attempt < 6; attempt++) { + await until(() => state.observation.waits.length === attempt + 1); + expect(state.delivery.prompts).toEqual([]); + state.observation.waits[attempt]!.result.reject(new HostObservationError("unreachable")); + } + await until(() => state.observation.waits.length === 7); + expect(state.delivery.prompts).toEqual([]); + state.observation.waits[6]!.result.resolve(outcome("idle")); + await until(() => state.delivery.prompts.length === 1); + expect(state.delivery.prompts[0]).toContain("finished."); +}); + +test("removed-host notice retries parent dispatch without observing again or deleting unacknowledged intent", async () => { + const state = await scenario(); + await state.service.register({ ...intent, serverId: "remote" }); + await until(() => state.observation.waits.length === 1); + state.observation.waits[0]!.result.reject(new HostObservationError("removed")); + await until(() => state.delivery.prompts.length === 1); + expect(state.delivery.prompts[0]).toContain("Host remote was removed or is unavailable."); + expect(state.delivery.prompts[0]).toContain("Status of child agent child (child) is unknown."); + state.delivery.acknowledgments[0]!.reject(new Error("Parent provider unavailable")); + await until(() => state.delivery.prompts.length === 2); + expect(state.delivery.prompts[1]).toBe(state.delivery.prompts[0]); + expect(state.observation.waits).toHaveLength(1); + expect(await state.store.list()).toHaveLength(1); + state.delivery.acknowledgments[1]!.resolve("steered"); + await expect.poll(async () => (await state.store.list()).length).toBe(0); +}); + +test("failed removal after dispatch acknowledgment retries only storage, preserving the accepted outcome", async () => { + const state = await scenario(undefined, 1); + await state.service.register(intent); + await until(() => state.observation.waits.length === 1); + state.observation.waits[0]!.result.resolve(outcome("idle")); + await until(() => state.delivery.prompts.length === 1); + state.delivery.acknowledgments[0]!.resolve("steered"); + await expect.poll(async () => (await state.store.list()).length).toBe(0); + expect(state.delivery.prompts).toHaveLength(1); + expect(state.delivery.prompts[0]).toContain("finished."); + expect(state.observation.waits).toHaveLength(1); +}); + +test("shutdown during observation setup owns rejection of the already-canceled next result", async () => { + const state = await scenario(); + const opening = deferred(); + state.observation.opening = opening.promise; + await state.service.register(intent); + await until(() => state.observation.openedSignals.length === 1); + const closing = state.service.close(); + await until(() => state.observation.openedSignals[0]!.aborted); + opening.resolve(); + await closing; + expect(await state.store.list()).toHaveLength(1); + expect(state.delivery.prompts).toEqual([]); +}); diff --git a/packages/server/src/server/agent/finish-subscriptions/index.ts b/packages/server/src/server/agent/finish-subscriptions/index.ts new file mode 100644 index 0000000000..58c3c8eff1 --- /dev/null +++ b/packages/server/src/server/agent/finish-subscriptions/index.ts @@ -0,0 +1,363 @@ +import { setTimeout as pause } from "node:timers/promises"; +import { HostObservationError } from "../host-operations/index.js"; +import { randomUUID } from "node:crypto"; +import type { Logger } from "pino"; +import type { WaitForFinishResult } from "@getpaseo/client/internal/daemon-client"; +import type { FinishObservationResult } from "./internal/observation-buffer.js"; +import type { AgentMessageDelivery } from "../agent-messages/index.js"; +import { getParentAddressFromLabels } from "@getpaseo/protocol/agent-labels"; +import { + formatFinishNotificationBody, + finishNotificationEvents, + type FinishNotificationReason, +} from "./internal/notification.js"; +export { createJsonFinishSubscriptionStore } from "./internal/json-store.js"; + +export interface FinishSubscription { + id: string; + callerAgentId: string; + childAgentId: string; + serverId?: string; + /** Creation subscriptions stop reporting after the child is detached. */ + requireParentOwnership?: boolean; +} +export interface FinishSubscriptionStore { + list(): Promise; + put(subscription: FinishSubscription): Promise; + remove(id: string): Promise; +} +export interface FinishObservationSession { + next(): Promise; + close(): Promise; +} +export interface FinishObservation { + open(subscription: FinishSubscription, signal: AbortSignal): Promise; +} +interface Dependencies extends FinishObservation { + store: FinishSubscriptionStore; + messages(): AgentMessageDelivery; + callerIsActive(agentId: string): Promise; + serverId: string; + logger: Logger; + retryDelayMs?: number; +} +export interface FinishRegistration { + /** The prompt was accepted. Replaces preceding registrations for this pair. */ + accept(): void; + /** Rejected dispatch preserves preceding registrations. */ + reject(): Promise; +} +export interface FinishSubscriptions { + start(): Promise; + register( + input: Omit, + pendingDispatch?: boolean, + ): Promise; + close(): Promise; +} +interface Watch { + records: FinishSubscription[]; + controller: AbortController; + decision: Promise; + decide(accepted: boolean): void; + pending: boolean; + task?: Promise; +} +function key(record: FinishSubscription): string { + return JSON.stringify([record.callerAgentId, record.childAgentId, record.serverId]); +} + +/** Owns durable intent, live observation, supersession, and dispatch acknowledgment. */ +export function createFinishSubscriptions(deps: Dependencies): FinishSubscriptions { + const tasks = new Set>(); + const watches = new Map>(); + let closed = false; + let mutations = Promise.resolve(); + function serialize(action: () => Promise): Promise { + const result = mutations.then(action); + mutations = result.then( + () => undefined, + () => undefined, + ); + return result; + } + function attach(records: FinishSubscription[], pending: boolean): Watch { + let decide!: Watch["decide"]; + const decision = new Promise((resolve) => { + decide = resolve; + }); + const watch: Watch = { records, controller: new AbortController(), pending, decision, decide }; + const group = watches.get(key(records[0]!)) ?? new Set(); + group.add(watch); + watches.set(key(records[0]!), group); + if (!pending) decide(true); + watch.task = run(watch, group).catch((error) => + deps.logger.error({ err: error }, "Finish subscription failed"), + ); + tasks.add(watch.task); + void watch.task.then(() => tasks.delete(watch.task!)); + return watch; + } + function stop(watch: Watch, group: Set): void { + watch.controller.abort(); + watch.decide(false); + group.delete(watch); + if (!group.size && watches.get(key(watch.records[0]!)) === group) + watches.delete(key(watch.records[0]!)); + } + async function remove(watch: Watch): Promise { + // Only IDs captured by this observer are eligible; newer registrations survive. + for (const record of watch.records) await deps.store.remove(record.id); + } + async function delay(signal: AbortSignal, attempt: number): Promise { + try { + await pause( + Math.min((deps.retryDelayMs ?? 1000) * 2 ** Math.min(attempt, 3), 5000), + undefined, + { signal }, + ); + } catch (error) { + if (!signal.aborted) throw error; + } + } + async function abortable(operation: Promise, signal: AbortSignal): Promise { + let abort!: () => void; + const cancellation = new Promise((_resolve, reject) => { + abort = () => reject(signal.reason); + if (signal.aborted) abort(); + else signal.addEventListener("abort", abort, { once: true }); + }); + try { + return await Promise.race([operation, cancellation]); + } finally { + signal.removeEventListener("abort", abort); + } + } + async function decisions(group: Set): Promise { + // New candidates may arrive while an older dispatch decision is pending. + while (true) { + if (closed) return; + const pending = [...group].filter((watch) => watch.pending); + if (!pending.length) return; + await Promise.all(pending.map((watch) => watch.decision)); + } + } + async function suppressed( + record: FinishSubscription, + result?: WaitForFinishResult, + ): Promise { + if (!(await deps.callerIsActive(record.callerAgentId))) return true; + if (!record.requireParentOwnership || !result?.final) return false; + const parent = getParentAddressFromLabels( + result.final.labels, + record.serverId ?? deps.serverId, + ); + return parent?.agentId !== record.callerAgentId || parent.serverId !== deps.serverId; + } + function reasonFor(result: FinishObservationResult): FinishNotificationReason { + if (result.reason === "host unreachable" || result.reason === "host removed") + return result.reason; + if (!result.final) return "is unavailable"; + if (result.final.status === "closed") return "was closed"; + if (result.status === "error") return "errored"; + if (result.status === "permission") return "needs permission"; + return result.reason ?? "finished"; + } + async function deliver( + record: FinishSubscription, + result: FinishObservationResult, + permissions: Set, + signal: AbortSignal, + ): Promise { + const requests = result.final?.pendingPermissions ?? []; + const reason = reasonFor(result); + for (const id of permissions) + if (!requests.some((request) => request.id === id)) permissions.delete(id); + const acknowledgments: Promise[] = []; + for (const request of reason === "needs permission" ? requests : [undefined]) { + if (signal.aborted) return false; + if (request && permissions.has(request.id)) continue; + const title = result.final?.title ?? record.childAgentId; + const receipt = deps.messages().enqueue({ + recipientId: record.callerAgentId, + prompt: formatFinishNotificationBody({ + childAgentId: record.childAgentId, + serverId: record.serverId ?? deps.serverId, + title, + reason, + lastAssistantMessage: result.lastMessage, + permissionRequest: request, + }), + source: { + kind: "agent-notification", + agentId: record.childAgentId, + serverId: record.serverId ?? deps.serverId, + title, + event: finishNotificationEvents[reason], + }, + activeTurnBehavior: "steer", + unarchive: false, + }); + acknowledgments.push( + receipt.dispatched.then(() => { + if (request) permissions.add(request.id); + return undefined; + }), + ); + } + await abortable(Promise.all(acknowledgments), signal); + if (signal.aborted) return false; + if (reason !== "needs permission") return true; + return false; + } + async function run(watch: Watch, group: Set): Promise { + const signal = watch.controller.signal; + if (!(await watch.decision) || signal.aborted) return; + const record = { + ...watch.records[0]!, + requireParentOwnership: watch.records.every( + (intent) => intent.requireParentOwnership === true, + ), + }; + const permissions = new Set(); + let observation: FinishObservationSession | undefined; + let pending: FinishObservationResult | undefined; + let acknowledged = false; + let remoteFailures = 0; + let retries = 0; + let warned = false; + async function observeNext(): Promise { + try { + observation ??= await deps.open(record, signal); + const result = await abortable(observation.next(), signal); + remoteFailures = 0; + return result; + } catch (error) { + await observation?.close(); + observation = undefined; + signal.throwIfAborted(); + if (!(error instanceof HostObservationError) || !record.serverId) throw error; + if (error.reason === "connecting") return undefined; + if (error.reason === "unreachable" && ++remoteFailures < 3) return undefined; + // Host visibility ends this subscription; it says nothing about child completion. + return { + status: "error", + final: null, + lastMessage: null, + error: null, + reason: error.reason === "removed" ? "host removed" : "host unreachable", + }; + } + } + try { + while (!signal.aborted) { + try { + if (await suppressed(record)) { + await remove(watch); + stop(watch, group); + return; + } + if (!pending) pending = await observeNext(); + if (!pending) { + await delay(signal, retries++); + continue; + } + if (signal.aborted) return; + await decisions(group); + if (signal.aborted) return; + if (await suppressed(record, pending)) { + await remove(watch); + stop(watch, group); + return; + } + if (!acknowledged) acknowledged = await deliver(record, pending, permissions, signal); + if (acknowledged) { + // A removal failure retries only the commit, never the already accepted callback. + await remove(watch); + stop(watch, group); + return; + } + pending = undefined; + retries = 0; + warned = false; + } catch (error) { + if (signal.aborted) return; + // Delivery and storage failures retain the captured outcome and durable intent. + // They do not spend the remote-observation budget or change the child's outcome. + if (!warned) { + deps.logger.warn( + { err: error, subscription: record.id }, + "Retrying finish subscription", + ); + warned = true; + } + await delay(signal, retries++); + } + } + } finally { + await observation?.close(); + } + } + + return { + start: () => + serialize(async () => { + const groups = new Map(); + for (const record of await deps.store.list()) { + const records = groups.get(key(record)) ?? []; + records.push(record); + groups.set(key(record), records); + } + for (const records of groups.values()) attach(records, false); + }), + register: (input, pendingDispatch = false) => + serialize(async () => { + if (closed) throw new Error("Finish subscriptions are closed"); + const record: FinishSubscription = { + ...input, + serverId: input.serverId === deps.serverId ? undefined : input.serverId, + id: randomUUID(), + }; + await deps.store.put(record); + const group = watches.get(key(record)) ?? new Set(); + const predecessors = [...group]; + const watch = attach([record], true); + const actualGroup = watches.get(key(record))!; + const registration: FinishRegistration = { + accept() { + if (!watch.pending || closed) return; + watch.pending = false; + for (const prior of predecessors) stop(prior, actualGroup); + watch.decide(true); + void serialize(async () => { + for (const prior of predecessors) await remove(prior); + }).catch((error) => + deps.logger.error({ err: error }, "Failed to remove superseded finish subscriptions"), + ); + }, + reject: () => + serialize(async () => { + stop(watch, actualGroup); + await remove(watch); + }), + }; + if (!pendingDispatch) registration.accept(); + return registration; + }), + async close() { + closed = true; + await mutations; + for (const group of watches.values()) + for (const watch of group) { + watch.controller.abort(); + watch.decide(false); + } + await mutations; + await Promise.all(tasks); + await deps.store.list(); + // Dispatch may still be pending. The delivery owner drains it after observers stop. + }, + }; +} + +export { createFinishObservation } from "./internal/observation.js"; diff --git a/packages/server/src/server/agent/finish-subscriptions/internal/json-store.test.ts b/packages/server/src/server/agent/finish-subscriptions/internal/json-store.test.ts new file mode 100644 index 0000000000..5cf696393c --- /dev/null +++ b/packages/server/src/server/agent/finish-subscriptions/internal/json-store.test.ts @@ -0,0 +1,51 @@ +import { mkdtemp, rm, mkdir, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, test } from "vitest"; +import { createJsonFinishSubscriptionStore } from "../index.js"; + +const homes: string[] = []; +afterEach(async () => { + await Promise.all(homes.splice(0).map((home) => rm(home, { recursive: true, force: true }))); +}); +async function temporaryHome() { + const result = await mkdtemp(join(tmpdir(), "finish-subscriptions-")); + homes.push(result); + return result; +} +const subscription = { + id: "subscription", + callerAgentId: "parent", + childAgentId: "child", + serverId: "remote", +}; + +test("acknowledged concurrent writes survive reopening and removal", async () => { + const directory = await temporaryHome(); + const store = createJsonFinishSubscriptionStore(directory); + await Promise.all([store.put(subscription), store.put({ ...subscription, id: "second" })]); + const restarted = createJsonFinishSubscriptionStore(directory); + expect(await restarted.list()).toEqual([subscription, { ...subscription, id: "second" }]); + await restarted.remove(subscription.id); + expect(await createJsonFinishSubscriptionStore(directory).list()).toEqual([ + { ...subscription, id: "second" }, + ]); +}); + +test("write failure rejects registration and does not publish an unwritten record", async () => { + const directory = await temporaryHome(); + const store = createJsonFinishSubscriptionStore(directory); + expect(await store.list()).toEqual([]); + await mkdir(join(directory, "finish-subscriptions.json")); + await expect(store.put(subscription)).rejects.toThrow(); + expect(await store.list()).toEqual([]); +}); + +test("invalid persisted records fail visibly", async () => { + const directory = await temporaryHome(); + await writeFile( + join(directory, "finish-subscriptions.json"), + JSON.stringify([{ callerAgentId: 42 }]), + ); + await expect(createJsonFinishSubscriptionStore(directory).list()).rejects.toThrow(); +}); diff --git a/packages/server/src/server/agent/finish-subscriptions/internal/json-store.ts b/packages/server/src/server/agent/finish-subscriptions/internal/json-store.ts new file mode 100644 index 0000000000..02b9635521 --- /dev/null +++ b/packages/server/src/server/agent/finish-subscriptions/internal/json-store.ts @@ -0,0 +1,58 @@ +import { readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { z } from "zod"; +import { writeJsonFileAtomic } from "../../../atomic-file.js"; +import type { FinishSubscription, FinishSubscriptionStore } from "../index.js"; + +const recordsSchema = z.array( + z.object({ + id: z.string().min(1), + callerAgentId: z.string().min(1), + childAgentId: z.string().min(1), + serverId: z.string().min(1).optional(), + requireParentOwnership: z.boolean().optional(), + }), +); + +export function createJsonFinishSubscriptionStore(paseoHome: string): FinishSubscriptionStore { + const file = join(paseoHome, "finish-subscriptions.json"); + let records: FinishSubscription[] | undefined; + let pending = Promise.resolve(); + async function load(): Promise { + if (records) return records; + try { + records = recordsSchema.parse(JSON.parse(await readFile(file, "utf8"))); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + records = []; + } + return records; + } + function serialized(action: () => Promise): Promise { + const result = pending.then(action); + pending = result.then( + () => undefined, + () => undefined, + ); + return result; + } + async function mutate( + update: (current: FinishSubscription[]) => FinishSubscription[], + ): Promise { + const next = update(await load()); + await writeJsonFileAtomic(file, next); + records = next; + } + return { + list: () => serialized(async () => (await load()).map((record) => Object.assign({}, record))), + put: (subscription) => + serialized(() => + mutate((current) => [ + ...current.filter((record) => record.id !== subscription.id), + { ...subscription }, + ]), + ), + remove: (id) => + serialized(() => mutate((current) => current.filter((record) => record.id !== id))), + }; +} diff --git a/packages/server/src/server/agent/finish-subscriptions/internal/notification.ts b/packages/server/src/server/agent/finish-subscriptions/internal/notification.ts new file mode 100644 index 0000000000..d1cdc9bad7 --- /dev/null +++ b/packages/server/src/server/agent/finish-subscriptions/internal/notification.ts @@ -0,0 +1,70 @@ +import type { AgentPermissionRequest } from "../../agent-sdk-types.js"; +export type FinishNotificationReason = + | "host unreachable" + | "host removed" + | "finished" + | "errored" + | "needs permission" + | "is unavailable" + | "was closed" + | "was canceled"; + +export const finishNotificationEvents = { + "host unreachable": "errored", + "host removed": "errored", + finished: "finished", + errored: "errored", + "needs permission": "permission-required", + "was closed": "closed", + "is unavailable": "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; + +interface FinishNotificationBodyInput { + childAgentId: string; + serverId?: string; + title: string; + reason: FinishNotificationReason; + lastAssistantMessage: string | null; + permissionRequest?: AgentPermissionRequest; +} + +export function formatFinishNotificationBody(params: FinishNotificationBodyInput): string { + if (params.reason === "host unreachable" || params.reason === "host removed") { + const host = + params.reason === "host removed" + ? `Host ${params.serverId} was removed or is unavailable.` + : `Could not observe child agent ${params.childAgentId} on host ${params.serverId} after three failed attempts.`; + return `${host}\n\nStatus of child agent ${params.childAgentId} (${params.title}) is unknown.\n\nServer: ${params.serverId}\n\nThis finish subscription has ended. Reassess the delegated work; reconnecting will not resume notifications.`; + } + const statusLine = `Agent ${params.childAgentId} (${params.title}) ${params.reason}.`; + const sections = [statusLine]; + if (params.serverId) sections.push(`Server: ${params.serverId}`); + if (params.reason === "needs permission" && params.permissionRequest) { + sections.push( + "Respond with `respond_to_permission` using the `agentId` and `requestId` below.", + `\n${JSON.stringify( + { + agentId: params.childAgentId, + ...(params.serverId ? { serverId: params.serverId } : {}), + requestId: params.permissionRequest.id, + request: params.permissionRequest, + }, + null, + 2, + )}\n`, + ); + } + let lastAssistantMessage = params.lastAssistantMessage?.trim(); + if (lastAssistantMessage) { + if (lastAssistantMessage.length > FINISH_NOTIFICATION_MESSAGE_LIMIT) { + const omitted = lastAssistantMessage.length - FINISH_NOTIFICATION_MESSAGE_LIMIT; + lastAssistantMessage = `${lastAssistantMessage.slice(0, FINISH_NOTIFICATION_MESSAGE_LIMIT)}\n[truncated ${omitted} chars; use get_agent_activity for the full response]`; + } + sections.push(`\n${lastAssistantMessage}\n`); + } + return sections.join("\n\n"); +} diff --git a/packages/server/src/server/agent/finish-subscriptions/internal/observation-buffer.ts b/packages/server/src/server/agent/finish-subscriptions/internal/observation-buffer.ts new file mode 100644 index 0000000000..95e80dd3a9 --- /dev/null +++ b/packages/server/src/server/agent/finish-subscriptions/internal/observation-buffer.ts @@ -0,0 +1,83 @@ +import type { WaitForFinishResult } from "@getpaseo/client/internal/daemon-client"; + +export type FinishObservationResult = WaitForFinishResult & { + reason?: "was canceled" | "host unreachable" | "host removed"; +}; + +/** Coalesces duplicate checkpoints and retains one terminal result while delivery is pending. */ +export function createObservationBuffer(signal: AbortSignal) { + const delivered = new Set(); + const permissions = new Map(); + const resolvedBeforeInitial = new Set(); + let initialPending = true; + let terminal: FinishObservationResult | undefined; + let failure: unknown; + let wake: (() => void) | undefined; + const abort = () => { + failure = signal.reason; + wake?.(); + }; + signal.addEventListener("abort", abort, { once: true }); + function push(result: FinishObservationResult): void { + if (result.status === "timeout") return; + if (result.status === "permission") { + for (const request of result.final?.pendingPermissions ?? []) + if (!delivered.has(request.id)) permissions.set(request.id, result); + } else terminal = result; + wake?.(); + } + return { + pushInitial(result: FinishObservationResult) { + initialPending = false; + if (result.status === "permission" && result.final) { + const pendingPermissions = result.final.pendingPermissions.filter( + (request) => !resolvedBeforeInitial.has(request.id), + ); + if (pendingPermissions.length) + push({ ...result, final: { ...result.final, pendingPermissions } }); + } else if ( + !terminal && + result.final?.status !== "running" && + result.final?.status !== "initializing" + ) + push(result); + resolvedBeforeInitial.clear(); + }, + push, + resolved(id: string) { + if (initialPending) resolvedBeforeInitial.add(id); + delivered.delete(id); + permissions.delete(id); + }, + fail(error: unknown) { + failure = error; + wake?.(); + }, + async next(): Promise { + while (true) { + signal.throwIfAborted(); + const first = permissions.entries().next().value; + if (first) { + for (const request of first[1].final?.pendingPermissions ?? []) { + delivered.add(request.id); + permissions.delete(request.id); + } + return first[1]; + } + if (terminal) { + const result = terminal; + terminal = undefined; + return result; + } + if (failure) throw failure; + await new Promise((resolve) => { + wake = resolve; + }); + wake = undefined; + } + }, + dispose() { + signal.removeEventListener("abort", abort); + }, + }; +} diff --git a/packages/server/src/server/agent/finish-subscriptions/internal/observation-stream.test.ts b/packages/server/src/server/agent/finish-subscriptions/internal/observation-stream.test.ts new file mode 100644 index 0000000000..1d15a890df --- /dev/null +++ b/packages/server/src/server/agent/finish-subscriptions/internal/observation-stream.test.ts @@ -0,0 +1,217 @@ +import { expect, test } from "vitest"; +import type { WaitForFinishResult } from "@getpaseo/client/internal/daemon-client"; +import type { AgentStreamEvent } from "../../agent-sdk-types.js"; +import { + HostObservationError, + createHostObservationResolver, + type ToolHosts, + type HostObservation, +} from "../../host-operations/index.js"; +import { buildStoredAgentPayload } from "../../agent-projections.js"; +import { openFinishObservation } from "./observation-stream.js"; + +class Host implements HostObservation { + handler?: Parameters[1]; + released = false; + failure?: (error: unknown) => void; + baseline = Promise.withResolvers(); + async wait() { + return this.baseline.promise; + } + async observe( + _id: string, + handler: Parameters[1], + _signal: AbortSignal, + onError: (error: unknown) => void, + ) { + this.handler = handler; + this.failure = onError; + return async () => { + this.released = true; + }; + } + emit(status: "idle" | "running", event?: AgentStreamEvent) { + this.handler!( + buildStoredAgentPayload( + { + id: "child", + cwd: "/tmp", + provider: "codex", + labels: {}, + lastStatus: status, + createdAt: "2026-01-01", + updatedAt: "2026-01-01", + }, + ["codex"], + ), + event, + "Last response", + ); + } +} +function permission(id: string): AgentStreamEvent { + return { + type: "permission_requested", + provider: "codex", + request: { id, provider: "codex", name: "Write", kind: "tool", input: { path: id } }, + }; +} +async function scenario() { + const host = new Host(); + const controller = new AbortController(); + const observation = await openFinishObservation( + { use: async (_server, action) => action(host) }, + { id: "watch", callerAgentId: "parent", childAgentId: "child" }, + controller.signal, + ); + return { host, observation, controller }; +} + +test("permission resolution followed by transient idle waits for resumed work to settle", async () => { + const { host, observation } = await scenario(); + try { + host.emit("running", permission("first")); + expect((await observation.next()).status).toBe("permission"); + host.emit("idle", { type: "permission_resolved", provider: "codex", requestId: "first" }); + let settled = false; + const next = observation.next().then((result) => { + settled = true; + return result; + }); + await Promise.resolve(); + expect(settled).toBe(false); + host.emit("running", { type: "turn_started", provider: "codex" }); + await Promise.resolve(); + expect(settled).toBe(false); + host.emit("idle", { type: "turn_completed", provider: "codex" }); + expect(await next).toMatchObject({ status: "idle", lastMessage: "Last response" }); + } finally { + await observation.close(); + } + expect(host.released).toBe(true); +}); + +test("second checkpoint and terminal outcome remain buffered while the first callback is pending", async () => { + const { host, observation } = await scenario(); + try { + host.emit("running", permission("first")); + expect((await observation.next()).status).toBe("permission"); + // The recipient has not acknowledged the first callback; no next() is in flight. + host.emit("running", permission("second")); + host.emit("idle", { type: "turn_completed", provider: "codex" }); + expect((await observation.next()).final?.pendingPermissions[0]?.id).toBe("second"); + expect((await observation.next()).status).toBe("idle"); + } finally { + await observation.close(); + } +}); + +test("abort releases continuous observation even with a pending initial wait", async () => { + const { host, observation, controller } = await scenario(); + const next = observation.next(); + controller.abort(); + await expect(next).rejects.toThrow(); + await observation.close(); + expect(host.released).toBe(true); +}); + +test("late initial permission snapshot cannot resurrect an already resolved request", async () => { + const { host, observation } = await scenario(); + try { + const requestEvent = permission("resolved"); + host.emit("running", requestEvent); + await observation.next(); + host.emit("idle", { type: "permission_resolved", provider: "codex", requestId: "resolved" }); + const final = buildStoredAgentPayload( + { + id: "child", + cwd: "/tmp", + provider: "codex", + labels: {}, + lastStatus: "running", + createdAt: "2026-01-01", + updatedAt: "2026-01-01", + }, + ["codex"], + ); + if (requestEvent.type !== "permission_requested") throw new Error("Expected permission"); + final.pendingPermissions = [requestEvent.request]; + host.baseline.resolve({ status: "permission", final, lastMessage: null, error: null }); + await host.baseline.promise; + await Promise.resolve(); + host.emit("idle", { type: "turn_completed", provider: "codex" }); + expect((await observation.next()).status).toBe("idle"); + } finally { + await observation.close(); + } +}); + +test("lookup failure uses its error channel and never fabricates a missing-child result", async () => { + const { host, observation } = await scenario(); + try { + host.failure!(new Error("Route disconnected")); + await expect(observation.next()).rejects.toThrow("Route disconnected"); + } finally { + await observation.close(); + } +}); + +test("a terminal event observed after newer work starts waits for that running work", async () => { + const { host, observation } = await scenario(); + try { + host.emit("running", { type: "turn_completed", provider: "codex" }); + let settled = false; + const next = observation.next().then((result) => { + settled = true; + return result; + }); + await Promise.resolve(); + expect(settled).toBe(false); + host.emit("idle", { type: "turn_completed", provider: "codex" }); + expect((await next).status).toBe("idle"); + } finally { + await observation.close(); + } +}); + +test.each([ + { status: "connecting", reason: "connecting" }, + { status: "offline", reason: "unreachable" }, + { status: "disabled", reason: "unreachable" }, + { status: "removed", reason: "removed" }, +] as const)( + "remote observation classifies $status from the authoritative directory, ignoring arbitrary error codes", + async ({ status, reason }) => { + const row = { + registrationId: "registration", + serverId: "remote", + label: null, + source: { id: "owner", key: "remote" }, + metadata: {}, + enabled: status !== "disabled", + status: status === "removed" ? ("offline" as const) : status, + targetDescription: "Remote", + lastOnlineAt: null, + nextRetryAt: null, + error: null, + }; + const hosts: ToolHosts = { + directory: async () => ({ + self: { serverId: "origin", label: "Origin" }, + hosts: status === "removed" ? [] : [row], + }), + async use() { + throw Object.assign(new Error("Untrusted remote error"), { code: "host_unknown" }); + }, + }; + const local = new Host(); + const resolver = createHostObservationResolver({ + serverId: "origin", + local, + hosts: () => hosts, + }); + const error = await resolver.use("remote", async () => {}).catch((failure: unknown) => failure); + expect(error).toBeInstanceOf(HostObservationError); + expect(error).toMatchObject({ reason }); + }, +); diff --git a/packages/server/src/server/agent/finish-subscriptions/internal/observation-stream.ts b/packages/server/src/server/agent/finish-subscriptions/internal/observation-stream.ts new file mode 100644 index 0000000000..34c059810f --- /dev/null +++ b/packages/server/src/server/agent/finish-subscriptions/internal/observation-stream.ts @@ -0,0 +1,121 @@ +import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; +import type { AgentStreamEvent } from "../../agent-sdk-types.js"; +import type { HostObservation } from "../../host-operations/index.js"; +import type { FinishSubscription, FinishObservationSession } from "../index.js"; +import { createObservationBuffer } from "./observation-buffer.js"; + +interface ObservationScope { + use( + serverId: string | undefined, + action: (host: HostObservation) => Promise, + signal?: AbortSignal, + ): Promise; +} +function isTerminal(snapshot: AgentSnapshotPayload, event?: AgentStreamEvent): boolean { + if (snapshot.status === "closed" || snapshot.status === "error") return true; + return ( + snapshot.status === "idle" && + (event?.type === "turn_completed" || event?.type === "turn_canceled") + ); +} +function observeState( + buffer: ReturnType, + snapshot: AgentSnapshotPayload | null, + event?: AgentStreamEvent, + lastMessage: string | null = null, +): void { + if (!snapshot) { + buffer.push({ status: "idle", final: null, lastMessage: null, error: null }); + return; + } + if (event?.type === "permission_resolved") { + buffer.resolved(event.requestId); + return; + } + if (event?.type === "turn_started") return; + if (event?.type === "permission_requested") { + buffer.push({ + status: "permission", + final: { ...snapshot, pendingPermissions: [event.request] }, + lastMessage, + error: null, + }); + } else if (isTerminal(snapshot, event)) { + buffer.push({ + status: snapshot.status === "error" ? "error" : "idle", + final: snapshot, + lastMessage, + error: snapshot.lastError ?? null, + ...(event?.type === "turn_canceled" ? { reason: "was canceled" as const } : {}), + }); + } +} + +/** One continuous source feeds a bounded buffer while the caller awaits dispatch. */ +export async function openFinishObservation( + resolver: ObservationScope, + record: FinishSubscription, + signal: AbortSignal, +): Promise { + signal.throwIfAborted(); + const controller = new AbortController(); + const lifetime = AbortSignal.any([signal, controller.signal]); + const buffer = createObservationBuffer(lifetime); + let opened!: () => void; + let failed!: (error: unknown) => void; + const ready = new Promise((resolve, reject) => { + opened = resolve; + failed = reject; + }); + const route = resolver + .use( + record.serverId, + async (host) => { + const release = await host.observe( + record.childAgentId, + (snapshot, event, lastMessage) => observeState(buffer, snapshot, event, lastMessage), + lifetime, + (error) => buffer.fail(error), + ); + opened(); + // The ordinary wait supplies the initial settled snapshot and covers queued admission. + // Continuous observation buffers checkpoints and terminal events throughout delivery. + void host + .wait(record.childAgentId, lifetime) + .then((result) => { + buffer.pushInitial(result); + return undefined; + }) + .catch((error) => buffer.fail(error)); + try { + if (!lifetime.aborted) + await new Promise((resolve) => + lifetime.addEventListener("abort", () => resolve(), { once: true }), + ); + } finally { + await release(); + } + }, + lifetime, + ) + .catch((error) => { + failed(error); + buffer.fail(error); + }); + try { + await ready; + } catch (error) { + controller.abort(); + buffer.dispose(); + await route; + throw error; + } + return { + next: () => buffer.next(), + async close() { + controller.abort(); + buffer.dispose(); + await route; + }, + }; +} diff --git a/packages/server/src/server/agent/finish-subscriptions/internal/observation.ts b/packages/server/src/server/agent/finish-subscriptions/internal/observation.ts new file mode 100644 index 0000000000..887a4ba075 --- /dev/null +++ b/packages/server/src/server/agent/finish-subscriptions/internal/observation.ts @@ -0,0 +1,103 @@ +import type { Logger } from "pino"; +import type { AgentManager } from "../../agent-manager.js"; +import type { AgentStorage } from "../../agent-storage.js"; +import type { AgentMessageDelivery } from "../../agent-messages/index.js"; +import { buildStoredAgentPayload, toAgentPayload } from "../../agent-projections.js"; +import { serializeSnapshotWithMetadata } from "../../mcp-shared.js"; +import { createHostObservationResolver, type ToolHosts } from "../../host-operations/index.js"; +import { openFinishObservation } from "./observation-stream.js"; +import type { AgentStreamEvent } from "../../agent-sdk-types.js"; +import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; +import type { FinishObservation } from "../index.js"; + +export function createFinishObservation(input: { + agentManager: AgentManager; + agentStorage: AgentStorage; + messages(): AgentMessageDelivery; + hosts(): ToolHosts; + serverId: string; + logger: Logger; +}): FinishObservation { + const { agentManager, agentStorage } = input; + const resolver = createHostObservationResolver({ + serverId: input.serverId, + hosts: input.hosts, + local: { + async wait(agentId, signal) { + await input.messages().waitForDispatch(agentId, signal); + signal.throwIfAborted(); + const live = agentManager.getAgent(agentId); + if (!live) { + const stored = await agentStorage.get(agentId); + return { + status: stored?.lastStatus === "error" ? "error" : "idle", + final: stored ? buildStoredAgentPayload(stored, [stored.provider]) : null, + lastMessage: await agentManager.getLastAssistantMessage(agentId), + error: stored?.lastError ?? null, + }; + } + const result = await agentManager.waitForAgentEvent(agentId, { + signal, + waitForActive: true, + }); + const current = agentManager.getAgent(agentId); + const status = result.permission ? "permission" : result.status; + return { + status: status === "permission" || status === "error" ? status : "idle", + final: current + ? await serializeSnapshotWithMetadata(agentStorage, current, input.logger) + : null, + lastMessage: result.lastMessage, + error: current?.lastError ?? null, + }; + }, + async observe(agentId, handler, signal, onError) { + const stored = await agentStorage.get(agentId); + let updates = Promise.resolve(); + const publish = (snapshot: AgentSnapshotPayload | null, event?: AgentStreamEvent) => { + updates = updates + .then(async () => + handler(snapshot, event, await agentManager.getLastAssistantMessage(agentId)), + ) + .catch(onError); + }; + signal.throwIfAborted(); + const unsubscribe = agentManager.subscribe( + (event) => { + if (event.type === "agent_state" && event.agent.lifecycle !== "idle") + publish(toAgentPayload(event.agent, { title: stored?.title })); + else if ( + event.type === "agent_stream" && + [ + "permission_requested", + "permission_resolved", + "turn_started", + "turn_completed", + "turn_canceled", + "turn_failed", + ].includes(event.event.type) + ) { + const current = agentManager.getAgent(agentId); + if (current) publish(toAgentPayload(current, { title: stored?.title }), event.event); + } + }, + { agentId, replayState: false }, + ); + const current = agentManager.getAgent(agentId); + if (current) + publish(toAgentPayload(current, { title: stored?.title }), { + type: "turn_started", + provider: current.provider, + }); + else publish(stored ? buildStoredAgentPayload(stored, [stored.provider]) : null); + const stop = () => unsubscribe(); + signal.addEventListener("abort", stop, { once: true }); + return async () => { + signal.removeEventListener("abort", stop); + unsubscribe(); + }; + }, + }, + }); + return { open: (record, signal) => openFinishObservation(resolver, record, signal) }; +} diff --git a/packages/server/src/server/agent/host-operations/index.ts b/packages/server/src/server/agent/host-operations/index.ts index 294e869555..2242c3b872 100644 --- a/packages/server/src/server/agent/host-operations/index.ts +++ b/packages/server/src/server/agent/host-operations/index.ts @@ -1,9 +1,11 @@ +import type { AgentStreamEvent } from "../agent-sdk-types.js"; import { parentAddressLabels } from "@getpaseo/protocol/agent-labels"; import type { ActiveTurnBehavior, AgentSnapshotPayload } from "@getpaseo/protocol/messages"; import { randomUUID } from "node:crypto"; import { z } from "zod"; import { DaemonClient, + DaemonConnectionError, type FetchWorkspacesEntry, type WaitForFinishResult, } from "@getpaseo/client/internal/daemon-client"; @@ -17,6 +19,18 @@ import { summarizeAgentActivity } from "../activity-curator.js"; import { expandUserPath } from "../../path-utils.js"; import type { PaseoToolExecutionContext, PaseoToolResult } from "../tools/types.js"; +export class HostOperationRejectedError extends Error { + readonly code: unknown; + constructor(error: unknown) { + super(error instanceof Error ? error.message : String(error), { cause: error }); + this.name = "HostOperationRejectedError"; + this.code = error instanceof Error && "code" in error ? error.code : undefined; + } +} +function isAmbiguousPromptFailure(error: unknown): boolean { + if (error instanceof DaemonConnectionError) return true; + return error instanceof Error && error.message.startsWith("Timeout waiting for message"); +} export type HostOperation = ( input: Record, context: PaseoToolExecutionContext, @@ -29,24 +43,46 @@ export interface HostOperations { getStatus: HostOperation; getActivity: HostOperation; listAgents: HostOperation; +} +/** Connection authority at the host boundary, independent of child lifecycle. */ +export class HostObservationError extends Error { + constructor( + readonly reason: "connecting" | "unreachable" | "removed", + cause?: unknown, + ) { + super(`Host observation ${reason}`, { cause }); + this.name = "HostObservationError"; + } +} +export interface HostObservation { wait(agentId: string, signal: AbortSignal): Promise; observe( agentId: string, - handler: (snapshot: AgentSnapshotPayload | null) => void, + handler: ( + snapshot: AgentSnapshotPayload | null, + event?: AgentStreamEvent, + lastMessage?: string | null, + ) => void, signal: AbortSignal, + onError: (error: unknown) => void, ): Promise<() => Promise>; } export interface ToolHosts { directory(): Promise; - use(serverId: string, action: (client: DaemonClient) => Promise): Promise; + use( + serverId: string, + action: (client: DaemonClient) => Promise, + signal?: AbortSignal, + ): Promise; } /** Daemon MCP has the daemon's authority, bounded by the destination's admission of its link. */ export function createToolHosts(hosts: ConnectedHosts): ToolHosts { return { directory: () => hosts.list(), - async use(serverId, action) { + async use(serverId, action, signal) { const lifetime = new AbortController(); + let routeFailure: unknown; const transportFactory: DaemonTransportFactory = () => { const opens = new Set<() => void>(); const closes = new Set<(event?: unknown) => void>(); @@ -61,7 +97,7 @@ export function createToolHosts(hosts: ConnectedHosts): ToolHosts { sourceId: "mcp", originSessionId: "mcp", permissions: OWNER_PERMISSIONS, - signal: lifetime.signal, + signal: signal ? AbortSignal.any([lifetime.signal, signal]) : lifetime.signal, frame: (data) => { for (const listener of messages) listener(data, typeof data !== "string"); }, @@ -78,6 +114,7 @@ export function createToolHosts(hosts: ConnectedHosts): ToolHosts { return undefined; }) .catch((error) => { + routeFailure = error; for (const listener of errors) listener(error); for (const listener of closes) listener(error); }); @@ -114,10 +151,17 @@ export function createToolHosts(hosts: ConnectedHosts): ToolHosts { transportFactory, reconnect: { enabled: false }, }); + const stop = () => void client.close(); + signal?.addEventListener("abort", stop, { once: true }); try { - await client.connect(); + signal?.throwIfAborted(); + await client.connect().catch((error) => { + // The action has not run; no application request could have been transmitted. + throw new HostOperationRejectedError(routeFailure ?? error); + }); return await action(client); } finally { + signal?.removeEventListener("abort", stop); lifetime.abort(); await client.close(); } @@ -212,18 +256,29 @@ export function createHostResolver(input: { if (!input.callerAgentId && args.sessionMode !== undefined) { await client.setAgentMode(agentId, args.sessionMode as string); } - const receipt = await client.sendMessage(agentId, args.prompt as string, { - sessionMode: input.callerAgentId - ? (args.sessionMode as string | undefined) - : undefined, - sourceAgentId: input.callerAgentId, - sourceAgentServerId: input.callerAgentId ? input.serverId : undefined, - activeTurnBehavior: - (args.activeTurnBehavior as ActiveTurnBehavior | undefined) ?? - (input.callerAgentId ? "steer" : "interrupt"), - }); - const agent = (await client.fetchAgent(agentId))?.agent; - if (!agent) throw new Error(`Agent ${agentId} not found`); + const agent = ( + await client.fetchAgent(agentId).catch((error) => { + // Snapshot preflight precedes sendMessage, so failure is definitely unsent. + throw new HostOperationRejectedError(error); + }) + )?.agent; + if (!agent) + throw new HostOperationRejectedError(new Error(`Agent ${agentId} not found`)); + const receipt = await client + .sendMessage(agentId, args.prompt as string, { + sessionMode: input.callerAgentId + ? (args.sessionMode as string | undefined) + : undefined, + sourceAgentId: input.callerAgentId, + sourceAgentServerId: input.callerAgentId ? input.serverId : undefined, + activeTurnBehavior: + (args.activeTurnBehavior as ActiveTurnBehavior | undefined) ?? + (input.callerAgentId ? "steer" : "interrupt"), + }) + .catch((error) => { + if (isAmbiguousPromptFailure(error)) throw error; + throw new HostOperationRejectedError(error); + }); return result({ success: true, ...receipt, @@ -310,45 +365,6 @@ export function createHostResolver(input: { throw new Error(response.error ?? "Workspace creation failed"); return result(workspaceSummary(response.workspace)); }, - async observe(agentId, handler, signal) { - const observation = client.observeTimeline([agentId], { signal }); - const refresh = async () => handler((await client.fetchAgent(agentId))?.agent ?? null); - const unsubscribe = observation.subscribe({ - snapshot: () => {}, - update: (message) => { - if (message.type !== "agent_stream" || message.payload.agentId !== agentId) return; - const event = message.payload.event; - if ( - event.type === "permission_resolved" || - event.type === "turn_started" || - event.type === "turn_completed" || - event.type === "turn_canceled" || - event.type === "turn_failed" - ) - void refresh().catch(() => handler(null)); - }, - error: () => handler(null), - }); - try { - await observation.ready; - await refresh(); - } catch (error) { - unsubscribe(); - await observation.release(); - throw error; - } - return async () => { - unsubscribe(); - await observation.release(); - }; - }, - wait: (agentId, signal) => { - const close = () => void client.close(); - signal.addEventListener("abort", close, { once: true }); - return client - .waitForFinish(agentId, 0) - .finally(() => signal.removeEventListener("abort", close)); - }, }), ); }, @@ -366,36 +382,129 @@ function workspaceSummary(workspace: FetchWorkspacesEntry) { }; } -/** Wait pauses at permissions; observe only until that request clears before re-arming it. */ -export async function waitForPermissionChange( - host: HostOperations, - agentId: string, - requestId: string, - signal: AbortSignal, -): Promise { - signal.throwIfAborted(); - let changed!: () => void; - let failed!: (error: unknown) => void; - const ready = new Promise((resolve, reject) => { - changed = resolve; - failed = reject; - }); - const abort = () => failed(signal.reason ?? new Error("Finish wait stopped")); - signal.addEventListener("abort", abort, { once: true }); - let release: (() => Promise) | undefined; - try { - release = await host.observe( - agentId, - (snapshot) => { - if (!snapshot) failed(new Error(`Agent ${agentId} is unavailable`)); - else if (!snapshot.pendingPermissions.some((request) => request.id === requestId)) - changed(); - }, - signal, - ); - await ready; - } finally { - signal.removeEventListener("abort", abort); - await release?.(); - } +function remoteObservation(client: DaemonClient): HostObservation { + return { + async observe(agentId, handler, signal, onError) { + const observation = client.observeTimeline([agentId], { signal }); + const refresh = async (event?: AgentStreamEvent) => { + let snapshot = (await client.fetchAgent(agentId))?.agent ?? null; + let lastMessage: string | null = null; + if ( + snapshot && + (snapshot.status === "idle" || + snapshot.status === "closed" || + snapshot.status === "error" || + snapshot.pendingPermissions.length > 0) + ) { + // 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; + } + } + handler(snapshot, event, lastMessage); + }; + let updates = Promise.resolve(); + const unsubscribe = observation.subscribe({ + snapshot: () => {}, + update: (message) => { + if (message.type !== "agent_stream" || message.payload.agentId !== agentId) return; + const event = message.payload.event; + if ( + event.type === "permission_requested" || + event.type === "permission_resolved" || + event.type === "turn_started" || + event.type === "turn_completed" || + event.type === "turn_canceled" || + event.type === "turn_failed" + ) + updates = updates.then(() => refresh(event)).catch(onError); + }, + error: onError, + }); + try { + await observation.ready; + await refresh(); + } catch (error) { + unsubscribe(); + await observation.release(); + throw error; + } + return async () => { + unsubscribe(); + await observation.release(); + }; + }, + wait: (agentId, signal) => { + const close = () => void client.close(); + signal.addEventListener("abort", close, { once: true }); + return client + .waitForFinish(agentId, 0) + .then((outcome) => { + if (!outcome.final && outcome.error && outcome.error !== `Agent not found: ${agentId}`) + throw new Error(outcome.error); + return outcome; + }) + .finally(() => signal.removeEventListener("abort", close)); + }, + }; +} + +/** Shares routed status observation with daemon services independently of MCP execution. */ +export function createHostObservationResolver(input: { + serverId: string; + local: HostObservation; + hosts(): ToolHosts; +}) { + return { + async use( + serverId: string | undefined, + action: (host: HostObservation) => Promise, + signal?: AbortSignal, + ): Promise { + if (!serverId || serverId === input.serverId) return action(input.local); + const hosts = input.hosts(); + async function classify(error: unknown): Promise { + // This is the origin's authoritative registration directory, not a destination RPC error. + const row = (await hosts.directory()).hosts.find((host) => host.serverId === serverId); + if (!row) return new HostObservationError("removed", error); + if (row.status === "connecting") return new HostObservationError("connecting", error); + return new HostObservationError("unreachable", error); + } + async function fail(error: unknown): Promise { + throw await classify(error); + } + try { + return await hosts.use( + serverId, + (client) => { + const remote = remoteObservation(client); + return action({ + wait: (agentId, lifetime) => remote.wait(agentId, lifetime).catch(fail), + observe(agentId, handler, lifetime, onError) { + async function report(error: unknown): Promise { + try { + onError(await classify(error)); + } catch (failure) { + onError(failure); + } + } + return remote.observe(agentId, handler, lifetime, report).catch(fail); + }, + }); + }, + signal, + ); + } catch (error) { + if (error instanceof HostObservationError) throw error; + return fail(error); + } + }, + }; } diff --git a/packages/server/src/server/agent/mcp-server.test.ts b/packages/server/src/server/agent/mcp-server.test.ts index 2d066efdc9..b8a0d820ec 100644 --- a/packages/server/src/server/agent/mcp-server.test.ts +++ b/packages/server/src/server/agent/mcp-server.test.ts @@ -1,12 +1,20 @@ +import { + createFinishSubscriptions, + createFinishObservation, + createJsonFinishSubscriptionStore, + type FinishSubscriptionStore, + type FinishSubscriptions, + type FinishSubscription, +} from "./finish-subscriptions/index.js"; import type { AgentMessageDelivery } from "./agent-messages/index.js"; import { createTestAgentMessages } from "../test-utils/agent-messages.js"; -import { setupFinishNotification, type PromptDispatchDisposition } from "./agent-prompt.js"; -import { parseAgentMessage } from "./agent-messages/index.js"; +import { type PromptDispatchDisposition } from "./agent-prompt.js"; +import { parseAgentMessage, parseAgentMessages } from "./agent-messages/index.js"; import { execFileSync } from "node:child_process"; import { randomUUID } from "node:crypto"; import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { realpathSync } from "node:fs"; import { access, mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises"; import { join, resolve as resolvePath } from "node:path"; @@ -23,6 +31,7 @@ import { FakeOmp } from "./providers/omp/test-utils/fake-omp.js"; import type { AgentClient, AgentMode, + AgentPermissionRequest, AgentPromptInput, AgentProvider, AgentRunResult, @@ -125,6 +134,12 @@ interface RegisteredMcpToolWithHandler extends RegisteredMcpTool { }>; } +const subscriptionServices: FinishSubscriptions[] = []; +afterEach(async () => { + for (const service of subscriptionServices.splice(0)) await service.close(); +}); +const serverSubscriptionRecords = new WeakMap>(); +const serverSubscriptions = new WeakMap(); const serverDeliveries = new WeakMap(); const toolDeliveries = new WeakMap(); @@ -840,13 +855,43 @@ function createPaseoWorktreeForMcpTest(options: { // Typed in-memory workspace port for catalogue tests without workspace setup. async function createAgentMcpServer( options: Parameters[0], + subscriptionStore?: FinishSubscriptionStore, ) { const messages = createTestAgentMessages(options); + const records = new Map(); + const subscriptions = createFinishSubscriptions({ + store: subscriptionStore ?? { + list: async () => [...records.values()], + put: async (record) => { + records.set(record.id, record); + }, + remove: async (id) => { + records.delete(id); + }, + }, + messages: () => messages, + callerIsActive: async (id) => !(await options.agentStorage.get(id))?.archivedAt, + serverId: options.serverId ?? "self", + logger: options.logger, + ...createFinishObservation({ + ...options, + messages: () => messages, + serverId: options.serverId ?? "self", + hosts: () => { + if (!options.hosts) throw new Error("No hosts"); + return options.hosts(); + }, + }), + }); + subscriptionServices.push(subscriptions); const server = await createAgentMcpServerWithDependencies({ agentMessages: () => messages, + finishSubscriptions: () => subscriptions, workspaceRegistry: { get: async () => null, list: async () => [], upsert: async () => {} }, ...options, }); + serverSubscriptionRecords.set(server, records); + serverSubscriptions.set(server, subscriptions); serverDeliveries.set(server, messages); return server; } @@ -3399,6 +3444,7 @@ class HeldTurnAgentSession implements AgentSession { readonly prompts: string[] = []; readonly steeredPrompts: AgentPromptInput[] = []; private readonly subscribers = new Set<(event: AgentStreamEvent) => void>(); + private readonly permissions = new Map(); private activeTurnId: string | null = null; private turnStartGate: Promise | null = null; private nextStartError: Error | null = null; @@ -3527,11 +3573,25 @@ class HeldTurnAgentSession implements AgentSession { async setMode(): Promise {} - getPendingPermissions() { - return []; + requestPermission(id: string): void { + const request: AgentPermissionRequest = { + id, + name: "Write", + kind: "tool", + input: { path: id }, + }; + this.permissions.set(id, request); + this.pushEvent({ type: "permission_requested", provider: this.provider, request }); } - async respondToPermission(): Promise {} + getPendingPermissions() { + return [...this.permissions.values()]; + } + + async respondToPermission(requestId: string): Promise { + this.permissions.delete(requestId); + this.pushEvent({ type: "permission_resolved", provider: this.provider, requestId }); + } describePersistence() { return { provider: this.provider, sessionId: this.id }; @@ -3842,14 +3902,13 @@ describe("send_agent_prompt MCP tool", () => { secondResolve = resolve; }); for (const dispatchResult of [first, second]) { - setupFinishNotification({ - agentMessages: serverDeliveries.get(server)!, - agentManager: manager, - agentStorage: storage, - childAgentId: child.id, - callerAgentId: parent.id, - dispatchResult, - logger, + const registration = await serverSubscriptions + .get(server)! + .register({ childAgentId: child.id, callerAgentId: parent.id }, true); + void dispatchResult.then((result) => { + if (result) registration.accept(); + else void registration.reject(); + return undefined; }); } firstResolve(firstAccepted ? "steered" : false); @@ -3925,8 +3984,7 @@ describe("send_agent_prompt MCP tool", () => { await serverDeliveries .get(server)! .waitForDispatch("child-agent", new AbortController().signal); - expect(spies.agentManager.subscribe).toHaveBeenCalled(); - expect(spies.agentManager.waitForAgentEvent).not.toHaveBeenCalled(); + expect(serverSubscriptionRecords.get(server)!.size).toBe(1); expect(response.structuredContent.guidance).toContain("You will get notified"); }); @@ -6735,3 +6793,125 @@ it("rejects hidden agent-worktree placement and directs callers to workspace cre }), ).rejects.toThrow("create_workspace"); }); + +describe("local durable finish subscription MCP wiring", () => { + const logger = createTestLogger(); + const existingCwd = process.cwd(); + it.each(["concurrent", "repeated"] as const)( + "reports %s permission requests and final completion", + async (sequence) => { + const workdir = await mkdtemp(join(tmpdir(), "mcp-local-permissions-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + const parentClient = new HeldTurnAgentClient("claude", false); + const childClient = new HeldTurnAgentClient("codex", true); + const manager = new AgentManager({ + clients: { claude: parentClient, codex: childClient }, + registry: storage, + logger, + }); + let mcp: Client | undefined; + try { + const parent = await manager.createAgent( + { provider: "claude", cwd: existingCwd }, + undefined, + { workspaceId: "wks_parent" }, + ); + const child = await manager.createAgent( + { provider: "codex", cwd: existingCwd }, + undefined, + { workspaceId: "wks_parent" }, + ); + const server = await createAgentMcpServer({ + agentManager: manager, + agentStorage: storage, + callerAgentId: parent.id, + providerSnapshotManager: createOpenCodeManager().manager, + logger, + }); + mcp = await connectInMemoryMcpClient(server); + await sendMcpAndAwaitDispatch(mcp, server, { + agentId: child.id, + prompt: "Permission work", + }); + const prompts = () => + parentClient.sessions[0]!.prompts.flatMap((prompt) => parseAgentMessages(prompt) ?? []) + .map((message) => message.text) + .join("\n"); + const permissionIds = () => + [...prompts().matchAll(/\n([\s\S]+?)\n<\/permission-request>/g)].map( + (match) => JSON.parse(match[1]!).requestId, + ); + const respond = async (requestId: string) => { + const result = await mcp!.callTool({ + name: "respond_to_permission", + arguments: { agentId: child.id, requestId, response: { behavior: "allow" } }, + }); + expect(result.isError).not.toBe(true); + }; + childClient.sessions[0]!.requestPermission("first"); + if (sequence === "concurrent") { + childClient.sessions[0]!.requestPermission("second"); + await vi.waitFor(() => expect(permissionIds()).toEqual(["first", "second"])); + await respond("first"); + await respond("second"); + } else { + await vi.waitFor(() => expect(permissionIds()).toEqual(["first"])); + await respond("first"); + childClient.sessions[0]!.requestPermission("second"); + await vi.waitFor(() => expect(permissionIds()).toEqual(["first", "second"])); + await respond("second"); + } + expect(serverSubscriptionRecords.get(server)!.size).toBe(1); + expect(prompts()).not.toContain("finished."); + childClient.sessions[0]!.finishTurn("Permission work complete"); + await vi.waitFor(() => expect(prompts()).toContain("finished.")); + expect(prompts()).toContain("Permission work complete"); + expect(permissionIds()).toEqual(["first", "second"]); + await vi.waitFor(() => expect(serverSubscriptionRecords.get(server)!.size).toBe(0)); + } finally { + await mcp?.close(); + await removeAgentStateDir(manager, storage, workdir); + } + }, + ); + + it("returns a tool error before sending a prompt when subscription storage cannot commit", async () => { + const workdir = await mkdtemp(join(tmpdir(), "mcp-registration-failure-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + const client = new HeldTurnAgentClient("codex", true); + const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger }); + const subscriptions = createJsonFinishSubscriptionStore(workdir); + let mcp: Client | undefined; + try { + const parent = await manager.createAgent({ provider: "codex", cwd: existingCwd }, undefined, { + workspaceId: "wks_parent", + }); + const child = await manager.createAgent({ provider: "codex", cwd: existingCwd }, undefined, { + workspaceId: "wks_parent", + }); + expect(await subscriptions.list()).toEqual([]); + await mkdir(join(workdir, "finish-subscriptions.json")); + const server = await createAgentMcpServer( + { + agentManager: manager, + agentStorage: storage, + callerAgentId: parent.id, + providerSnapshotManager: createOpenCodeManager().manager, + logger, + }, + subscriptions, + ); + mcp = await connectInMemoryMcpClient(server); + const result = await mcp.callTool({ + name: "send_agent_prompt", + arguments: { agentId: child.id, prompt: "Must not dispatch" }, + }); + expect(result.isError).toBe(true); + expect(client.sessions[1]!.prompts).toEqual([]); + expect(await subscriptions.list()).toEqual([]); + } finally { + await mcp?.close(); + await removeAgentStateDir(manager, storage, workdir); + } + }); +}); diff --git a/packages/server/src/server/agent/tools/paseo-tools.ts b/packages/server/src/server/agent/tools/paseo-tools.ts index 74c4361c9e..6b84f17161 100644 --- a/packages/server/src/server/agent/tools/paseo-tools.ts +++ b/packages/server/src/server/agent/tools/paseo-tools.ts @@ -1,7 +1,8 @@ +import type { FinishSubscriptions, FinishRegistration } from "../finish-subscriptions/index.js"; import type { AgentMessageDelivery } from "../agent-messages/index.js"; import { createHostResolver, - waitForPermissionChange, + HostOperationRejectedError, type HostOperation, type HostOperations, type ToolHosts, @@ -77,8 +78,6 @@ import { import { type PromptDispatchDisposition, sendPromptToAgent, - setupFinishNotification, - setupFinishWait, waitForAgentRunStartWithTimeout, } from "../agent-prompt.js"; import { respondToAgentPermission } from "../permission-response.js"; @@ -118,6 +117,7 @@ import { isPaseoToolEnabled } from "../paseo-tool-policy.js"; export interface PaseoToolHostDependencies { agentMessages?: () => AgentMessageDelivery; + finishSubscriptions?: () => FinishSubscriptions; serverId?: string; hosts?: () => ToolHosts; agentManager: AgentManager; @@ -175,10 +175,6 @@ export interface PaseoToolHostDependencies { logger: Logger; } -function settledStatus(status: string): "error" | "idle" { - return status === "error" ? "error" : "idle"; -} - interface ProviderSummary { id: AgentProvider; label: string; @@ -663,6 +659,49 @@ function createToolCatalog( if (!options.agentMessages) throw new Error("Agent message delivery is not configured"); return options.agentMessages(); } + function finishSubscriptions(): FinishSubscriptions { + if (!options.finishSubscriptions) throw new Error("Finish subscriptions are not configured"); + return options.finishSubscriptions(); + } + async function prepareRemoteNotification( + name: string, + destination: string | undefined, + input: Record, + ): Promise { + if ( + name !== "send_agent_prompt" || + destination === options.serverId || + !callerAgentId || + input.notifyOnFinish === false + ) + return undefined; + return finishSubscriptions().register( + { childAgentId: input.agentId as string, callerAgentId, serverId: destination }, + true, + ); + } + async function registerRemoteCreation( + name: string, + destination: string | undefined, + input: Record, + response: PaseoToolResult, + ): Promise { + if ( + name !== "create_agent" || + destination === options.serverId || + !callerAgentId || + input.notifyOnFinish === false || + response.isError + ) + return; + const childAgentId = (response.structuredContent as { agentId: string }).agentId; + await finishSubscriptions().register({ + childAgentId, + callerAgentId, + serverId: destination, + requireParentOwnership: true, + }); + } const localHostOperations = new Map(); const tools = new Map(); const hostTools = new Map([ @@ -691,36 +730,6 @@ function createToolCatalog( getStatus: (args, context) => localHostOperations.get("get_agent_status")!(args, context), getActivity: (args, context) => localHostOperations.get("get_agent_activity")!(args, context), listAgents: (args, context) => localHostOperations.get("list_agents")!(args, context), - async observe(agentId, handler, signal) { - const unsubscribe = agentManager.subscribe( - (event) => { - if (event.type === "agent_state") handler(toAgentPayload(event.agent)); - }, - { agentId, replayState: true }, - ); - const stop = () => unsubscribe(); - signal.addEventListener("abort", stop, { once: true }); - return async () => { - signal.removeEventListener("abort", stop); - unsubscribe(); - }; - }, - async wait(agentId, signal) { - const settled = await agentManager.waitForAgentEvent(agentId, { - signal, - waitForActive: true, - }); - const snapshot = agentManager.getAgent(agentId); - const final = snapshot - ? await serializeSnapshotWithMetadata(agentStorage, snapshot, childLogger) - : null; - return { - status: settled.permission ? "permission" : settledStatus(settled.status), - final, - lastMessage: settled.lastMessage, - error: final?.lastError ?? null, - }; - }, }, }); const registerTool = ( @@ -752,34 +761,14 @@ function createToolCatalog( }; handler = async ({ serverId, ...input }, context) => { const destination = serverId ?? options.serverId; + const remoteNotification = await prepareRemoteNotification(name, destination, input); try { const response = await hostResolver.use(serverId, (host) => host[operation](input, context), ); - let childAgentId: string | undefined; - if (name === "create_agent") { - childAgentId = (response.structuredContent as { agentId: string }).agentId; - } else if (name === "send_agent_prompt" && destination !== options.serverId) { - childAgentId = input.agentId as string; - } - if (childAgentId && callerAgentId && input.notifyOnFinish !== false) { - setupFinishWait({ - agentMessages: agentMessages(), - agentManager, - agentStorage, - originServerId: options.serverId, - childServerId: destination, - childAgentId, - callerAgentId, - logger: childLogger, - wait: (signal) => - hostResolver.use(destination, (host) => host.wait(childAgentId, signal)), - afterPermission: (requestId, signal) => - hostResolver.use(destination, (host) => - waitForPermissionChange(host, childAgentId, requestId, signal), - ), - }); - } + if (response.isError) await remoteNotification?.reject(); + else remoteNotification?.accept(); + await registerRemoteCreation(name, destination, input, response); return { ...response, structuredContent: { @@ -788,6 +777,11 @@ function createToolCatalog( }, }; } catch (error) { + // A lost response does not establish whether the remote daemon admitted the prompt. + // Reconcile the child rather than silently dropping notification intent. + if (error instanceof PaseoToolInputError || error instanceof HostOperationRejectedError) + await remoteNotification?.reject(); + else remoteNotification?.accept(); if (!options.serverId) throw error; const failure = { serverId: destination, @@ -1621,6 +1615,7 @@ function createToolCatalog( { agentManager, agentStorage, + finishSubscriptions: options.finishSubscriptions?.(), logger: childLogger, paseoHome: options.paseoHome, worktreesRoot: options.worktreesRoot, @@ -1649,7 +1644,7 @@ function createToolCatalog( }, mode: parsedArgs.settings?.modeId, promptFailure: "throw", - notifyOnFinish: false, + notifyOnFinish, detached: resolvedArgs.detached, callerAgentId, callerContext, @@ -1966,68 +1961,56 @@ function createToolCatalog( notifyOnFinish = Boolean(callerAgentId), activeTurnBehavior = callerAgentId ? "steer" : "interrupt", }) => { - let stopNotification: (() => void) | undefined; - let acceptDispatch!: (result: PromptDispatchDisposition | false) => void; - const dispatchResult = new Promise((resolve) => { - acceptDispatch = resolve; - }); - const notifying = Boolean(callerAgentId && notifyOnFinish); - function prepareDispatch(): void { - if (!callerAgentId || !notifyOnFinish) return; - stopNotification = setupFinishNotification({ - agentMessages: agentMessages(), - agentManager, - agentStorage, - childAgentId: agentId, - childServerId: options.serverId, - callerAgentId, - waitForTurnStart: false, - dispatchResult, - logger: childLogger, - }); + if (callerAgentId && !agentManager.getAgent(agentId) && !(await agentStorage.get(agentId))) { + throw new PaseoToolInputError(`Agent ${agentId} not found`); } - function confirmDispatch(disposition: PromptDispatchDisposition): void { - // A new turn must ignore outcomes from the displaced turn. Its iterator - // has not started yet, so rearm here; steering keeps the early observer. - if (disposition !== "turn_started") { - acceptDispatch(disposition); - return; - } - acceptDispatch(disposition); - if (!callerAgentId || !notifyOnFinish) return; - stopNotification?.(); - stopNotification = setupFinishNotification({ - agentMessages: agentMessages(), - agentManager, - agentStorage, - childAgentId: agentId, - childServerId: options.serverId, - callerAgentId, - waitForTurnStart: true, - logger: childLogger, - }); + const notifying = Boolean(callerAgentId && notifyOnFinish); + const notification = + notifying && callerAgentId + ? await (() => { + if (!options.finishSubscriptions) + throw new Error("Finish subscriptions are not configured"); + return options + .finishSubscriptions() + .register({ childAgentId: agentId, callerAgentId }, true); + })() + : undefined; + function confirmDispatch(_disposition: PromptDispatchDisposition): void { + notification?.accept(); + } + function rejectNotification(): void { + void notification + ?.reject() + .catch((error) => + childLogger.error( + { err: error, agentId }, + "Failed to remove rejected finish subscription", + ), + ); } if (callerAgentId) { - if (!agentManager.getAgent(agentId) && !(await agentStorage.get(agentId))) { - throw new PaseoToolInputError(`Agent ${agentId} not found`); - } - const receipt = agentMessages().enqueue({ - recipientId: agentId, - prompt, - source: { - kind: "agent-message", - agentId: callerAgentId, - title: resolveCallerAgent()?.config.title ?? undefined, - }, - sessionMode, - activeTurnBehavior, - beforeDispatch: prepareDispatch, - onDispatch: confirmDispatch, - }); + const receipt = (() => { + try { + return agentMessages().enqueue({ + recipientId: agentId, + prompt, + source: { + kind: "agent-message", + agentId: callerAgentId, + title: resolveCallerAgent()?.config.title ?? undefined, + }, + sessionMode, + activeTurnBehavior, + onDispatch: confirmDispatch, + }); + } catch (error) { + rejectNotification(); + throw error; + } + })(); void receipt.dispatched.catch(() => { - acceptDispatch(false); - stopNotification?.(); + rejectNotification(); }); return { content: [], @@ -2054,8 +2037,7 @@ function createToolCatalog( onDispatch: confirmDispatch, logger: childLogger, }).catch((error) => { - acceptDispatch(false); - stopNotification?.(); + rejectNotification(); if (error instanceof AgentNotFoundError) { throw new PaseoToolInputError(error.message); } @@ -2065,7 +2047,7 @@ function createToolCatalog( // Return once the provider has accepted the turn, so the status reports it running. if (disposition === "turn_started") { await waitForAgentRunStartWithTimeout(agentManager, agentId).catch((error) => { - stopNotification?.(); + rejectNotification(); throw error; }); } diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index d8d611be70..03decb5f67 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1,3 +1,8 @@ +import { + createFinishSubscriptions, + createFinishObservation, + createJsonFinishSubscriptionStore, +} from "./agent/finish-subscriptions/index.js"; import { createWorkspaceRuntimeStore } from "./workspace-runtime-store/index.js"; import type { PluginRegistries } from "@getpaseo/protocol/plugin-registry"; import { describeHookWorkspace } from "./plugins/lifecycle/index.js"; @@ -1412,9 +1417,38 @@ export async function createPaseoDaemon( ); logger.info({ elapsed: elapsed() }, "Preparing voice and MCP runtime"); + const finishSubscriptions = createFinishSubscriptions({ + store: createJsonFinishSubscriptionStore(config.paseoHome), + serverId, + logger, + messages: () => { + if (!wsServer) throw new Error("Daemon is not accepting agent messages"); + return wsServer.agentMessages; + }, + callerIsActive: async (agentId) => { + const record = await agentStorage.get(agentId); + return record ? !record.archivedAt : Boolean(agentManager.getAgent(agentId)); + }, + ...createFinishObservation({ + agentManager, + agentStorage, + serverId, + logger, + messages: () => { + if (!wsServer) throw new Error("Daemon is not accepting agent messages"); + return wsServer.agentMessages; + }, + hosts: () => { + if (!wsServer) throw new Error("Daemon is not accepting host operations"); + return wsServer.toolHosts(); + }, + }), + }); + const createAgentToolHostDependencies = ( runtime: PaseoToolRuntimeContext, ): PaseoToolHostDependencies => ({ + finishSubscriptions: () => finishSubscriptions, agentMessages: () => { if (!wsServer) throw new Error("Daemon is not accepting agent messages"); return wsServer.agentMessages; @@ -1790,6 +1824,7 @@ export async function createPaseoDaemon( pluginRuntime.bindPaseoSessionHost(wsServer); await pluginRuntime.start(); providerSnapshotManager.settlePluginProviders(); + await finishSubscriptions.start(); wsServer.beginAcceptingConnections(); const relaySecurityLogger = logger.child({ module: "relay-security" }); relayRuntime = createRelayRuntime({ @@ -1874,6 +1909,7 @@ export async function createPaseoDaemon( scriptHealthMonitor.stop(); // Freeze both ingress and registration before taking the agent closure snapshot. wsServer?.prepareForShutdown(); + await finishSubscriptions.close(); await wsServer?.agentMessages.close(); agentManager.prepareForShutdown(); await closeAllAgents(logger, agentManager); 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 e24e431cd0..8a2d4f9c39 100644 --- a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts +++ b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts @@ -1715,3 +1715,320 @@ test( expect(requests[1]).not.toHaveProperty("sessionMode"); }, ); + +test( + "durable remote finish intent survives origin restart and unavailable-host recovery", + { timeout: 60000 }, + async () => { + const root = await mkdtemp(join(tmpdir(), "durable-remote-origin-")); + cleanup.push(() => rm(root, { recursive: true, force: true })); + const parentPrompts: string[] = []; + const finish = Promise.withResolvers(); + cleanup.push(async () => { + finish.resolve(); + }); + const b = await daemon({ + agentClients: createTestAgentClients({ beforeTurnComplete: () => finish.promise }), + }); + const clients = () => + createTestAgentClients({ + onStartTurn: (prompt) => { + if (String(prompt).includes('kind="agent-notification"')) + parentPrompts.push(String(prompt)); + }, + }); + const a = await daemon({ + paseoHomeRoot: root, + cleanup: false, + agentClients: clients(), + mcpEnabled: true, + }); + const origin = await api(a); + const destination = await api(b); + const saved = await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` }); + const directory = await online(origin.paseo); + const remoteId = directory.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: "Remote parent", + }); + const mcp = new Client({ name: "durable-remote-parent", 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 response = await mcp.callTool({ + name: "create_agent", + arguments: { + serverId: remoteId, + workspaceId: remoteWorkspace.id, + provider: "claude/claude-test", + title: "Remote child", + initialPrompt: "Keep working while origin restarts", + settings: { modeId: "full-access" }, + }, + }); + expect(response.isError).not.toBe(true); + const childId = (response.structuredContent as { agentId: string }).agentId; + expect((await destination.client.fetchAgent(childId))?.agent.status).toBe("running"); + const records = async (home: string) => + JSON.parse(await readFile(join(home, "finish-subscriptions.json"), "utf8")); + expect(await records(a.paseoHome)).toHaveLength(1); + // An unavailable saved route must retain intent, including through a restart. + await origin.paseo.hosts.update(saved.registrationId, { enabled: false }); + await origin.paseo.dispose(); + await origin.client.close(); + await mcp.close(); + await a.close(); + const restarted = await daemon({ + paseoHomeRoot: root, + cleanup: false, + agentClients: clients(), + mcpEnabled: true, + }); + const recovered = await api(restarted); + expect(await records(restarted.paseoHome)).toHaveLength(1); + expect(parentPrompts).toEqual([]); + expect((await destination.client.fetchAgent(childId))?.agent.status).toBe("running"); + await recovered.paseo.hosts.update(saved.registrationId, { enabled: true }); + await online(recovered.paseo); + expect(parentPrompts).toEqual([]); + expect(await records(restarted.paseoHome)).toHaveLength(1); + finish.resolve(); + await expect.poll(() => parentPrompts, { timeout: 15000 }).toHaveLength(1); + expect(parentPrompts[0]).toContain(`Agent ${childId}`); + expect(parentPrompts[0]).toContain(`Server: ${remoteId}`); + expect(parentPrompts[0]).toContain('event="finished"'); + await expect.poll(() => records(restarted.paseoHome), { timeout: 5000 }).toEqual([]); + }, +); + +test( + "remote finish observation survives a permission checkpoint until resumed completion", + { timeout: 30000 }, + async () => { + const parentPrompts: string[] = []; + const finish = Promise.withResolvers(); + cleanup.push(async () => { + finish.resolve(); + }); + const a = await daemon({ + mcpEnabled: true, + agentClients: createTestAgentClients({ + onStartTurn: (prompt) => { + if (String(prompt).includes('kind="agent-notification"')) + parentPrompts.push(String(prompt)); + }, + }), + }); + const b = await daemon({ + agentClients: createTestAgentClients({ beforeTurnComplete: () => finish.promise }), + }); + const origin = await api(a); + const destination = await api(b); + await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` }); + const directory = await online(origin.paseo); + 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: "Permission parent", + }); + const mcp = new Client({ name: "permission-parent", version: "1" }); + cleanup.push(() => mcp.close()); + await mcp.connect( + new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${a.port}/mcp/agents?callerAgentId=${parent.id}`), + ), + ); + const response = await mcp.callTool({ + name: "create_agent", + arguments: { + serverId: directory.hosts[0]!.serverId, + workspaceId: remoteWorkspace.id, + provider: "claude/claude-test", + title: "Permission child", + initialPrompt: "Run rm -f permission.txt", + settings: { modeId: "default" }, + }, + }); + expect(response.isError).not.toBe(true); + const childId = (response.structuredContent as { agentId: string }).agentId; + await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(1); + expect(parentPrompts[0]).toContain('event="permission-required"'); + expect(parentPrompts[0]).toContain("respond_to_permission"); + const child = (await destination.client.fetchAgent(childId))!.agent; + expect(child.pendingPermissions).toHaveLength(1); + expect( + JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")), + ).toHaveLength(1); + await destination.client.respondToPermission(childId, child.pendingPermissions[0]!.id, { + behavior: "allow", + }); + expect(parentPrompts).toHaveLength(1); + expect((await destination.client.fetchAgent(childId))?.agent.status).toBe("running"); + finish.resolve(); + await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(2); + expect(parentPrompts[1]).toContain('event="finished"'); + await expect + .poll(async () => + JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")), + ) + .toEqual([]); + }, +); + +test( + "disabled remote send rejects unsent finish intent before reconnecting to an idle child", + { timeout: 30000 }, + async () => { + const parentPrompts: string[] = []; + const a = await daemon({ + mcpEnabled: true, + agentClients: createTestAgentClients({ + onStartTurn: (prompt) => { + if (String(prompt).includes('kind="agent-notification"')) + parentPrompts.push(String(prompt)); + }, + }), + }); + const b = await daemon({ agentClients: createTestAgentClients() }); + const origin = await api(a); + const destination = await api(b); + const saved = await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` }); + const directory = await online(origin.paseo); + const serverId = directory.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: "Parent", + }); + const child = await remoteWorkspace.agents.create({ + config: { provider: "claude/claude-test", modeId: "full-access" }, + title: "Idle child", + }); + const mcp = new Client({ name: "offline-parent", version: "1" }); + cleanup.push(() => mcp.close()); + await mcp.connect( + new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${a.port}/mcp/agents?callerAgentId=${parent.id}`), + ), + ); + await origin.paseo.hosts.update(saved.registrationId, { enabled: false }); + const response = await mcp.callTool({ + name: "send_agent_prompt", + arguments: { serverId, agentId: child.id, prompt: "Never sent" }, + }); + expect(response.isError).toBe(true); + const records = () => + readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8").then(JSON.parse); + expect(await records()).toEqual([]); + await origin.paseo.hosts.update(saved.registrationId, { enabled: true }); + await online(origin.paseo); + expect((await destination.client.fetchAgent(child.id))?.agent.status).toBe("idle"); + expect(await records()).toEqual([]); + expect(parentPrompts).toEqual([]); + }, +); + +test.each(["removed", "disabled"] as const)( + "%s remote destination reports unknown child status and consumes its finish subscription", + { timeout: 30000 }, + async (availability) => { + const parentPrompts: string[] = []; + const finish = Promise.withResolvers(); + cleanup.push(async () => finish.resolve()); + const a = await daemon({ + mcpEnabled: true, + agentClients: createTestAgentClients({ + onStartTurn: (prompt) => { + if (String(prompt).includes('kind="agent-notification"')) + parentPrompts.push(String(prompt)); + }, + }), + }); + const b = await daemon({ + agentClients: createTestAgentClients({ beforeTurnComplete: () => finish.promise }), + }); + const origin = await api(a); + const destination = await api(b); + const saved = await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` }); + const directory = await online(origin.paseo); + const serverId = directory.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: "Parent", + }); + const mcp = new Client({ name: "removed-host-parent", version: "1" }); + cleanup.push(() => mcp.close()); + await mcp.connect( + new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${a.port}/mcp/agents?callerAgentId=${parent.id}`), + ), + ); + const response = await mcp.callTool({ + name: "create_agent", + arguments: { + serverId, + workspaceId: remoteWorkspace.id, + provider: "claude/claude-test", + title: "Child", + initialPrompt: "Held work", + settings: { modeId: "full-access" }, + }, + }); + expect(response.isError).not.toBe(true); + const childId = (response.structuredContent as { agentId: string }).agentId; + if (availability === "removed") await origin.paseo.hosts.remove(saved.registrationId); + else await origin.paseo.hosts.update(saved.registrationId, { enabled: false }); + await expect.poll(() => parentPrompts, { timeout: 15000 }).toHaveLength(1); + expect(parentPrompts[0]).toContain(`Status of child agent ${childId}`); + expect(parentPrompts[0]).toContain(`Server: ${serverId}`); + expect(parentPrompts[0]).toContain( + availability === "removed" + ? `Host ${serverId} was removed or is unavailable.` + : `Could not observe child agent ${childId} on host ${serverId} after three failed attempts.`, + ); + expect(parentPrompts[0]).toContain("is unknown."); + expect(parentPrompts[0]).not.toContain("finished."); + await expect + .poll(async () => + JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")), + ) + .toEqual([]); + if (availability === "removed") + await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` }); + else await origin.paseo.hosts.update(saved.registrationId, { enabled: true }); + await online(origin.paseo); + finish.resolve(); + const completed = await destination.client.waitForFinish(childId, 5000); + expect(completed.final?.status).toBe("idle"); + expect(parentPrompts).toHaveLength(1); + expect( + JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")), + ).toEqual([]); + }, +); diff --git a/packages/server/src/server/daemon-e2e/daemon-restart-resume.e2e.test.ts b/packages/server/src/server/daemon-e2e/daemon-restart-resume.e2e.test.ts index 6486f8e04a..efc99c9b59 100644 --- a/packages/server/src/server/daemon-e2e/daemon-restart-resume.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/daemon-restart-resume.e2e.test.ts @@ -1,3 +1,5 @@ +import { AgentStorage } from "../agent/agent-storage.js"; +import { createTestLogger } from "../../test-utils/test-logger.js"; import { afterEach, beforeEach, describe, expect, test } from "vitest"; import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; @@ -64,3 +66,157 @@ describe("daemon restart resume", () => { } }, 30_000); }); + +// These run in separate OS processes so SIGKILL cannot invoke daemon shutdown hooks. +import { fork, type ChildProcess } from "node:child_process"; +import { once } from "node:events"; +import { readFile, mkdtemp, rm } from "node:fs/promises"; +import { fileURLToPath } from "node:url"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import { DaemonClient } from "../test-utils/daemon-client.js"; + +async function restartFixture(root: string) { + const child = fork( + fileURLToPath(new URL("../test-utils/finish-subscription-daemon.ts", import.meta.url)), + [root], + { + execArgv: ["--conditions=source", "--import", "tsx"], + stdio: ["ignore", "pipe", "pipe", "ipc"], + }, + ); + let output = ""; + child.stdout?.on("data", (data) => { + output += data; + }); + child.stderr?.on("data", (data) => { + output += data; + }); + const ready = await Promise.race([ + once(child, "message").then( + ([message]) => message as { port: number; paseoHome: string; cwd: string }, + ), + once(child, "exit").then(([code]) => { + throw new Error(`Fixture exited ${code}: ${output}`); + }), + ]); + return { child, ...ready }; +} +async function stopFixture(child: ChildProcess, abrupt: boolean) { + if (child.exitCode !== null || child.signalCode !== null) return; + const exited = once(child, "exit"); + if (abrupt) child.kill("SIGKILL"); + else child.send("stop"); + await exited; +} +async function readPrompts(root: string): Promise { + try { + return (await readFile(path.join(root, "parent-prompts.jsonl"), "utf8")) + .trim() + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line) as string); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return []; + throw error; + } +} + +test.each([false, true])( + "durable finish subscriptions recover a running child after restart (abrupt=%s)", + async (abrupt) => { + const root = await mkdtemp(path.join(tmpdir(), "durable-finish-process-")); + const processes: ChildProcess[] = []; + let client: DaemonClient | undefined; + let mcp: Client | undefined; + try { + const first = await restartFixture(root); + processes.push(first.child); + client = new DaemonClient({ url: `ws://127.0.0.1:${first.port}/ws`, appVersion: "0.13.0" }); + await client.connect(); + await client.fetchAgents({ subscribe: {} }); + const parent = await client.createAgent({ + provider: "claude", + cwd: first.cwd, + title: "Durable parent", + modeId: "full-access", + }); + mcp = new Client({ name: "durable-parent", version: "1" }); + await mcp.connect( + new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${first.port}/mcp/agents?callerAgentId=${parent.id}`), + ), + ); + const result = await mcp.callTool({ + name: "create_agent", + arguments: { + provider: "claude/claude-test", + title: "Held child", + initialPrompt: "HOLD_CHILD", + }, + }); + expect(result.isError).not.toBe(true); + const childId = (result.structuredContent as { agentId: string }).agentId; + expect((await client.fetchAgent(childId))?.agent.status).toBe("running"); + // Kill only after the running snapshot commits, so this exercises bootstrap repair + // rather than recovery of the earlier idle snapshot while its write is pending. + await expect + .poll( + async () => + ( + await new AgentStorage(path.join(first.paseoHome, "agents"), createTestLogger()).get( + childId, + ) + )?.lastStatus, + ) + .toBe("running"); + expect( + JSON.parse(await readFile(path.join(first.paseoHome, "finish-subscriptions.json"), "utf8")), + ).toHaveLength(1); + await mcp.close(); + mcp = undefined; + await client.close(); + client = undefined; + await stopFixture(first.child, abrupt); + const second = await restartFixture(root); + processes.push(second.child); + await expect.poll(() => readPrompts(root), { timeout: 15000 }).toHaveLength(1); + const prompts = await readPrompts(root); + expect(prompts[0]).toContain(`Agent ${childId}`); + expect(prompts[0]).toContain("was closed"); + await expect + .poll( + async () => + JSON.parse( + await readFile(path.join(second.paseoHome, "finish-subscriptions.json"), "utf8"), + ), + { timeout: 5000 }, + ) + .toEqual([]); + client = new DaemonClient({ url: `ws://127.0.0.1:${second.port}/ws`, appVersion: "0.13.0" }); + await client.connect(); + await client.fetchAgents({ subscribe: {} }); + expect((await client.fetchAgent(childId))?.agent.status).toBe("closed"); + await client.close(); + client = undefined; + await stopFixture(second.child, false); + const third = await restartFixture(root); + processes.push(third.child); + expect(await readPrompts(root)).toHaveLength(1); + expect( + JSON.parse(await readFile(path.join(third.paseoHome, "finish-subscriptions.json"), "utf8")), + ).toEqual([]); + } catch (error) { + console.error( + await readFile(path.join(root, "fixture.log"), "utf8").catch(() => "No fixture log"), + ); + throw error; + } finally { + await mcp?.close(); + await client?.close(); + for (const child of processes.toReversed()) await stopFixture(child, true); + await rm(root, { recursive: true, force: true }); + } + }, + 60000, +); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index e5fdbca32f..9a2b2c0820 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -5360,7 +5360,7 @@ export class Session { if (!record || record.internal) { return null; } - return { record, lastMessage: null }; + return { record, lastMessage: await this.agentManager.getLastAssistantMessage(agentId) }; } private async getAgentPayloadById(agentId: string): Promise { diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index dcc1d65b40..e4a0ca7668 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -598,6 +598,7 @@ function createSessionForWorkspaceTests( listAgents: () => [], listProviderSubagentActivity: () => [], getAgent: () => null, + getLastAssistantMessage: async () => null, archiveAgent: async () => ({ archivedAt: new Date().toISOString() }), archiveSnapshot: async () => ({}), unarchiveSnapshot: async () => true, diff --git a/packages/server/src/server/test-utils/finish-subscription-daemon.ts b/packages/server/src/server/test-utils/finish-subscription-daemon.ts new file mode 100644 index 0000000000..d35bc55ec8 --- /dev/null +++ b/packages/server/src/server/test-utils/finish-subscription-daemon.ts @@ -0,0 +1,40 @@ +import pino from "pino"; +// Isolated process fixture for restart/termination tests. Never targets a developer home. +import { appendFile, mkdir } from "node:fs/promises"; +import { join } from "node:path"; +import { createTestPaseoDaemon } from "./paseo-daemon.js"; +import { createTestAgentClients } from "./fake-agent-client.js"; + +const root = process.argv[2]; +if (!root || !process.send) throw new Error("Expected isolated test home and IPC parent"); +await mkdir(join(root, "static"), { recursive: true }); +const held = Promise.withResolvers(); +const daemon = await createTestPaseoDaemon({ + paseoHomeRoot: root, + staticDir: join(root, "static"), + logger: pino({ level: "warn" }, pino.destination(join(root, "fixture.log"))), + cleanup: false, + mcpEnabled: true, + agentClients: createTestAgentClients({ + onStartTurn: (prompt) => { + if (String(prompt).includes('kind="agent-notification"')) { + void appendFile(join(root, "parent-prompts.jsonl"), `${JSON.stringify(prompt)}\n`); + } + }, + beforeTurnComplete: (prompt) => + String(prompt).includes("HOLD_CHILD") ? held.promise : Promise.resolve(), + }), +}); +process.send({ port: daemon.port, paseoHome: daemon.paseoHome, cwd: daemon.staticDir }); +process.on("message", (message) => { + if (message === "finish") held.resolve(); + if (message === "stop") { + void daemon.close().then( + () => process.exit(0), + (error) => { + console.error(error); + process.exit(1); + }, + ); + } +});