From d0568f9f1b61791e4523887f677fc97435b15e7c Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 21 Oct 2026 18:28:03 +0700 Subject: [PATCH] Batch agent reports and completion notifications (#7561) * fix(server): batch agent reports and completion notifications Admit agent-originated messages immediately so a child can report and finish before one recipient-owned delivery. Keep human prompts direct, retain individual provenance through replay, and complete send receipts only after provider dispatch. * test(server): wire queued delivery into bridge integration fixture * test(server): resolve batching CLI fixture independently of cwd Workspace CI runs from packages/server, while the original regression was verified from the repository root. Resolve the CLI entrypoint relative to this test file in both contexts. * test(server): run batching CLI fixture with source exports Server integration CI builds only the client dependency stack. Match the existing CLI subprocess fixtures so the send regression exercises workspace sources without requiring compiled server output. * test(server): await persisted batch receipts within integration deadline The provider-start callback precedes asynchronous receipt commits. Use the same explicit integration deadline as dispatch instead of the matcher default of one second on Windows CI. --- docs/agent-lifecycle.md | 7 +- packages/app/src/composer/actions.ts | 2 +- packages/app/src/runtime/host-runtime.test.ts | 1 + packages/cli/src/commands/agent/send.ts | 11 +- packages/client/src/daemon-client.ts | 35 +- packages/protocol/src/agent-message.ts | 1 + packages/protocol/src/messages.ts | 4 + .../src/server/agent/activity-curator.test.ts | 6 +- .../src/server/agent/agent-manager.test.ts | 78 +++ .../server/src/server/agent/agent-manager.ts | 99 ++-- .../agent/agent-messages/delivery.test.ts | 285 +++++++++ .../server/agent/agent-messages/delivery.ts | 189 ++++++ .../server/agent/agent-messages/envelope.ts | 221 +++++++ .../src/server/agent/agent-messages/index.ts | 202 +------ .../src/server/agent/agent-prompt.test.ts | 76 ++- .../server/src/server/agent/agent-prompt.ts | 49 +- .../src/server/agent/create-agent/create.ts | 9 +- .../src/server/agent/host-operations/index.ts | 16 +- .../src/server/agent/mcp-server.test.ts | 98 ++-- .../agent/providers/opencode/bridge.test.ts | 14 +- .../src/server/agent/tools/paseo-tools.ts | 55 +- packages/server/src/server/bootstrap.ts | 5 + .../server/connected-hosts/hosts.e2e.test.ts | 551 +++++++++++++----- packages/server/src/server/session.test.ts | 2 + packages/server/src/server/session.ts | 150 +++-- .../session.workspace-git-watch.test.ts | 2 + ...on.workspace-resolution-invariants.test.ts | 2 + .../src/server/session.workspaces.test.ts | 10 + .../snapshot-mutation-ownership.test.ts | 2 + .../src/server/test-utils/agent-messages.ts | 43 ++ .../src/server/test-utils/session-stubs.ts | 13 + .../server/src/server/websocket-server.ts | 37 ++ .../server/src/server/wire-compat.test.ts | 2 + skills/paseo/SKILL.md | 8 +- 34 files changed, 1725 insertions(+), 560 deletions(-) create mode 100644 packages/server/src/server/agent/agent-messages/delivery.test.ts create mode 100644 packages/server/src/server/agent/agent-messages/delivery.ts create mode 100644 packages/server/src/server/agent/agent-messages/envelope.ts create mode 100644 packages/server/src/server/test-utils/agent-messages.ts diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index b49e8309db..ffcec92ddd 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -75,7 +75,12 @@ Setup belongs to the workspace, including combined workspace-plus-agent creation Users can also detach an existing subagent from the subagents track. Detach is deliberately a manual lifecycle gesture, not an agent-facing MCP tool. It removes the parent and open-tab lifecycle labels: it does not stop, archive, move, or restart the agent. The agent keeps its current `cwd` and `workspaceId`, leaves the former parent's track, and behaves like a root agent for tab close, workspace activity, and future parent archive. -`notifyOnFinish` defaults to `true` for agent-scoped creation and prompt follow-ups because most delegated work needs to report back to the creating agent. Set it to `false` only for truly fire-and-forget agents or prompts. +`notifyOnFinish` defaults to `true` for agent-scoped creation and prompt follow-ups because most delegated work needs to report back to the creating agent. Set it to `false` for explicit reporting or no completion callback. + +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. + 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. A watched child that closes before its finish event also notifies the caller so delegated work cannot disappear silently during archive or workspace teardown. diff --git a/packages/app/src/composer/actions.ts b/packages/app/src/composer/actions.ts index 4ae1764a02..7e197b4f13 100644 --- a/packages/app/src/composer/actions.ts +++ b/packages/app/src/composer/actions.ts @@ -54,7 +54,7 @@ export interface ComposerSendClient { images: Array<{ data: string; mimeType: string }>; attachments: ReturnType["attachments"]; }, - ) => Promise; + ) => Promise; uploadFile: (input: { fileName: string; mimeType: string; bytes: Uint8Array }) => Promise<{ requestId: string; file: { diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 1425dc3560..8f4d2acab5 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -136,6 +136,7 @@ class FakeDaemonClient { if (response) await response; const failure = this.sendAgentMessageFailures.shift(); if (failure) throw failure; + return {}; } async waitForSentMessages(count: number): Promise { diff --git a/packages/cli/src/commands/agent/send.ts b/packages/cli/src/commands/agent/send.ts index 1f288d4ca2..acdea38507 100644 --- a/packages/cli/src/commands/agent/send.ts +++ b/packages/cli/src/commands/agent/send.ts @@ -14,7 +14,7 @@ import { extname, resolve } from "node:path"; /** Result type for agent send command */ export interface AgentSendResult { agentId: string; - status: "sent" | "completed" | "timeout" | "permission" | "error"; + status: "queued" | "sent" | "completed" | "timeout" | "permission" | "error"; message: string; } @@ -201,7 +201,7 @@ export async function runSendCommand( const sourceAgentId = process.env.PASEO_AGENT_ID?.trim() || undefined; // Send the message - await client.sendAgentMessage(agentIdArg, promptInput, { + const receipt = await client.sendAgentMessage(agentIdArg, promptInput, { images, sourceAgentId, activeTurnBehavior: options.behavior ?? (sourceAgentId ? "steer" : "interrupt"), @@ -215,8 +215,11 @@ export async function runSendCommand( type: "single", data: { agentId: agentIdArg, - status: "sent", - message: "Message sent, not waiting for completion", + status: receipt.deliveryStatus ?? "sent", + message: + receipt.deliveryStatus === "queued" + ? "Message queued for delivery after two quiet seconds" + : "Message sent, not waiting for completion", }, schema: agentSendSchema, }; diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 0741b08147..94d20585e2 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -435,6 +435,8 @@ export interface DaemonClientTrace { export interface SendMessageOptions { sourceAgentId?: string; + sourceAgentServerId?: string; + sessionMode?: string; messageId?: string; /** What happens when the agent is mid-turn. The daemon interrupts the turn when omitted. */ activeTurnBehavior?: ActiveTurnBehavior; @@ -3732,26 +3734,34 @@ export class DaemonClient { async sendAgentMessage( agentId: string, text: string, - options?: SendMessageOptions, - ): Promise { + options: SendMessageOptions = {}, + ): Promise<{ deliveryStatus?: "queued" }> { if ( - options?.sourceAgentId && + options.sourceAgentId && this.lastServerInfoMessage?.features?.agentMessageProvenance !== true ) { throw new Error("Update the Paseo host to send messages with agent provenance."); } + if ( + options.sourceAgentServerId && + this.lastServerInfoMessage?.features?.agentMessageBatching !== true + ) { + throw new Error("Update the Paseo host to send messages with a qualified agent address."); + } const requestId = this.createRequestId(); - const messageId = options?.messageId ?? crypto.randomUUID(); + const messageId = options.messageId ?? crypto.randomUUID(); const message = SessionInboundMessageSchema.parse({ type: "send_agent_message_request", requestId, agentId, text, - sourceAgentId: options?.sourceAgentId, + sourceAgentId: options.sourceAgentId, + sourceAgentServerId: options.sourceAgentServerId, + sessionMode: options.sessionMode, ...(messageId ? { messageId } : {}), - ...(options?.activeTurnBehavior ? { activeTurnBehavior: options.activeTurnBehavior } : {}), - ...(options?.images ? { images: options.images } : {}), - ...(options?.attachments ? { attachments: options.attachments } : {}), + activeTurnBehavior: options.activeTurnBehavior, + images: options.images, + attachments: options.attachments, }); const payload = await this.sendRequest({ requestId, @@ -3770,10 +3780,15 @@ export class DaemonClient { if (!payload.accepted) { throw new Error(payload.error ?? "sendAgentMessage rejected"); } + return payload.deliveryStatus ? { deliveryStatus: payload.deliveryStatus } : {}; } - async sendMessage(agentId: string, text: string, options?: SendMessageOptions): Promise { - await this.sendAgentMessage(agentId, text, options); + async sendMessage( + agentId: string, + text: string, + options?: SendMessageOptions, + ): Promise<{ deliveryStatus?: "queued" }> { + return this.sendAgentMessage(agentId, text, options); } async rewindAgent( diff --git a/packages/protocol/src/agent-message.ts b/packages/protocol/src/agent-message.ts index 5caf63656d..81f871be26 100644 --- a/packages/protocol/src/agent-message.ts +++ b/packages/protocol/src/agent-message.ts @@ -5,6 +5,7 @@ export const AgentMessageSchema = z.object({ event: z.enum(["message", "finished", "errored", "permission-required", "closed"]), sender: z.object({ id: z.string().min(1), + serverId: z.string().min(1).optional(), title: z.string().min(1).optional(), }), text: z.string(), diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 4be3229b63..8eff2fbbd2 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -1387,6 +1387,8 @@ export const SendAgentMessageRequestSchema = z.object({ text: z.string(), /** Opaque sender identity for agent-originated prompts. */ sourceAgentId: z.string().min(1).optional(), + sourceAgentServerId: z.string().min(1).optional(), + sessionMode: z.string().optional(), messageId: z.string().optional(), // Client-provided ID for deduplication activeTurnBehavior: ActiveTurnBehaviorSchema.optional(), images: z.array(ImageAttachmentSchema).optional(), @@ -3776,6 +3778,7 @@ export const ServerInfoStatusPayloadSchema = z // COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30. canonicalSubmittedPrompts: z.boolean().optional(), agentMessageProvenance: z.boolean().optional(), + agentMessageBatching: z.boolean().optional(), // COMPAT(agentTurnIdentity): accept peers that observed pre-release v0.2.6 through 2027-01-31. agentTurnIdentity: z.boolean().optional(), // COMPAT(stableProjectIdentity): added in v0.1.109, remove gate after 2027-01-15. @@ -5026,6 +5029,7 @@ export const SendAgentMessageResponseMessageSchema = z.object({ requestId: z.string(), agentId: z.string(), accepted: z.boolean(), + deliveryStatus: z.literal("queued").optional(), error: z.string().nullable(), }), }); diff --git a/packages/server/src/server/agent/activity-curator.test.ts b/packages/server/src/server/agent/activity-curator.test.ts index 77814f1779..c98bbae54d 100644 --- a/packages/server/src/server/agent/activity-curator.test.ts +++ b/packages/server/src/server/agent/activity-curator.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it } from "vitest"; import { buildAgentForkContextAttachment, curateAgentActivity } from "./activity-curator.js"; import type { AgentTimelineItem } from "./agent-sdk-types.js"; import type { AgentTimelineRow } from "./agent-timeline-store-types.js"; -import { formatAgentMessage, projectAgentMessage } from "./agent-messages/index.js"; +import { formatAgentMessage, projectAgentMessages } from "./agent-messages/index.js"; function toolCallItem(params: { callId: string; @@ -81,10 +81,10 @@ describe("curateAgentActivity", () => { it("renders delivered agent messages with their event, sender, and body", () => { const delivered = (source: Parameters[0]["source"], text: string) => - projectAgentMessage({ + projectAgentMessages({ type: "user_message", text: formatAgentMessage({ id: `msg-${text.length}`, source, text }), - })!; + })[0]!; const timeline: AgentTimelineItem[] = [ delivered( { kind: "agent-message", agentId: "agt_lead", title: "CI lead" }, diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index e40a5f7720..ef5b361962 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -11991,3 +11991,81 @@ test.each([ expect(timelineStore.itemReads).toBe(readsBeforeDispatch); }, ); + +test.each(["live", "history", "accepted"] as const)( + "batched agent entries retain their identities through %s ingestion", + async (ingestion) => { + const workdir = mkdtempSync(join(tmpdir(), "agent-batch-provenance-")); + const entries = [ + { + id: "report", + source: { kind: "agent-message" as const, agentId: "worker", serverId: "remote-host" }, + text: "Result ", + }, + { + id: "finish", + source: { + kind: "agent-notification" as const, + agentId: "worker", + serverId: "remote-host", + event: "finished" as const, + }, + text: "Worker finished", + }, + ]; + const prompt = entries.map(formatAgentMessage).join("\n\n"); + const item: AgentTimelineItem = { + type: "user_message", + text: prompt, + clientMessageId: "report", + messageId: "provider-batch", + }; + const codex = fakeCodexEmitting( + ingestion === "history" ? { historyItems: [item] } : { turnItems: [item] }, + ); + const manager = new AgentManager({ clients: { codex }, logger }); + const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + try { + if (ingestion === "history") + await manager.hydrateTimelineFromProvider(agent.id, { force: true }); + else if (ingestion === "accepted") { + for await (const _event of manager.streamAgent(agent.id, prompt, { + clientMessageId: "report", + })) { + } + } else await manager.runAgent(agent.id, { text: "start" }); + expect( + manager + .getTimeline(agent.id) + .filter((row) => row.type === "tool_call") + .map((row) => ({ callId: row.callId, agentMessage: row.agentMessage })), + ).toEqual([ + { + callId: "paseo-agent-message:report", + agentMessage: { + event: "message", + sender: { id: "worker", serverId: "remote-host" }, + text: "Result ", + }, + }, + { + callId: "paseo-agent-message:finish", + agentMessage: { + event: "finished", + sender: { id: "worker", serverId: "remote-host" }, + text: "Worker finished", + }, + }, + ]); + expect(manager.getTimeline(agent.id).filter((row) => row.type === "user_message")).toEqual( + [], + ); + expect(manager.getAgent(agent.id)?.lastUserMessageAt).toBeNull(); + } finally { + await manager.closeAgent(agent.id); + rmSync(workdir, { recursive: true, force: true }); + } + }, +); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 5f1373ea5b..de7fe1d986 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -79,7 +79,7 @@ import { type PendingForegroundRun, } from "./agent-run-state.js"; import { invokeRewindCapability, type RewindMode } from "./rewind/rewind.js"; -import { parseAgentMessage, projectAgentMessage } from "./agent-messages/index.js"; +import { parseAgentMessage, projectAgentMessages } from "./agent-messages/index.js"; import { isStaleProviderSessionError } from "./stale-provider-session-error.js"; import { stripInternalPaseoMcpServer, withRuntimePaseoMcpServer } from "./runtime-mcp-config.js"; import { resolveCreateAgentTitles } from "./create-agent-title.js"; @@ -658,13 +658,13 @@ function buildExplicitTimelineSeedForRegister( function buildImportedTimelineRows(entries: readonly ImportedTimelineEntry[]): AgentTimelineRow[] { const rows: AgentTimelineRow[] = []; for (const entry of entries) { - const item = projectAgentMessage(entry.item); - if (!item) continue; - rows.push({ - seq: rows.length + 1, - timestamp: entry.timestamp ?? new Date().toISOString(), - item: limitAgentTimelineItemContent(item), - }); + for (const item of projectAgentMessages(entry.item)) { + rows.push({ + seq: rows.length + 1, + timestamp: entry.timestamp ?? new Date().toISOString(), + item: limitAgentTimelineItemContent(item), + }); + } } return rows; } @@ -4136,9 +4136,9 @@ export class AgentManager { for await (const rawEvent of agent.session.streamHistory()) { const event = limitAgentStreamEventContent(rawEvent); if (event.type === "timeline") { - const item = projectAgentMessage(event.item); - if (!item) continue; - historyEvents.push({ ...event, item }); + for (const item of projectAgentMessages(event.item)) { + historyEvents.push({ ...event, item }); + } } else if (event.type === "provider_subagent") { providerSubagentEvents.push(event); } @@ -4203,9 +4203,9 @@ export class AgentManager { if (event.type !== "timeline") { continue; } - const item = projectAgentMessage(event.item); - if (!item) continue; - historyEvents.push({ ...event, item }); + for (const item of projectAgentMessages(event.item)) { + historyEvents.push({ ...event, item }); + } } } catch (error) { this.logger.warn({ err: error, agentId: agent.id }, "Failed to hydrate provider history"); @@ -4537,42 +4537,32 @@ export class AgentManager { }): Promise { const { agent, event, options, flags } = params; - const item = projectAgentMessage(event.item); - if (!item) { - flags.shouldDispatchEvent = false; - flags.shouldNotifyWaiters = false; - return; - } - - event.item = item; - if ( - event.item.type === "user_message" && - event.item.clientMessageId && - this.reconcileSubmittedPromptEcho(agent, event.item, event.turnId) - ) { - flags.shouldDispatchEvent = false; - flags.shouldNotifyWaiters = false; - return; - } - - if (options?.fromHistory) { - this.recordTimeline( - agent.id, - event.item, - event.timestamp ? { timestamp: event.timestamp } : undefined, - ); - flags.shouldDispatchEvent = false; - flags.shouldNotifyWaiters = false; - return; - } - - this.recordAndDispatchTimelineItem(agent.id, event.item, event.provider, event.turnId); - if (event.item.type === "user_message") { - agent.lastUserMessageAt = new Date(); - this.emitState(agent); - } flags.shouldDispatchEvent = false; - flags.shouldNotifyWaiters = true; + flags.shouldNotifyWaiters = false; + for (const item of projectAgentMessages(event.item)) { + if ( + item.type === "user_message" && + item.clientMessageId && + this.reconcileSubmittedPromptEcho(agent, item, event.turnId) + ) + continue; + + if (options?.fromHistory) { + this.recordTimeline( + agent.id, + item, + event.timestamp ? { timestamp: event.timestamp } : undefined, + ); + continue; + } + + this.recordAndDispatchTimelineItem(agent.id, item, event.provider, event.turnId); + if (item.type === "user_message") { + agent.lastUserMessageAt = new Date(); + this.emitState(agent); + } + flags.shouldNotifyWaiters = true; + } } private onStreamTurnCompleted(params: { @@ -4818,16 +4808,17 @@ export class AgentManager { outcome: "accepted" | "rejected", options?: { messageId?: string; providerMessageId?: string; turnId?: string }, ): void { - const item = projectAgentMessage({ + const items = projectAgentMessages({ type: "user_message", text: submittedPromptText(prompt), clientMessageId, ...(options?.messageId ? { messageId: options.messageId } : {}), }); - if (!item) return; - // Human attempts stay in history on rejection; delivery notifications require acceptance. - if (outcome === "rejected" && item.type !== "user_message") return; - this.recordSubmittedPromptItem(agent, item, options); + for (const item of items) { + // Human attempts stay in history on rejection; delivery notifications require acceptance. + if (outcome === "rejected" && item.type !== "user_message") continue; + this.recordSubmittedPromptItem(agent, item, options); + } } private recordSubmittedPromptItem( diff --git a/packages/server/src/server/agent/agent-messages/delivery.test.ts b/packages/server/src/server/agent/agent-messages/delivery.test.ts new file mode 100644 index 0000000000..8cd480c1fa --- /dev/null +++ b/packages/server/src/server/agent/agent-messages/delivery.test.ts @@ -0,0 +1,285 @@ +import { expect, test } from "vitest"; +import { createAgentMessageDelivery, parseAgentMessages } from "./index.js"; +import type { AgentPromptInput } from "../agent-sdk-types.js"; + +function createClock() { + let now = 0; + const jobs = new Map<() => void, number>(); + return { + schedule(delayMs: number, callback: () => void) { + jobs.set(callback, now + delayMs); + return () => { + jobs.delete(callback); + }; + }, + advance(ms: number) { + now += ms; + for (const [callback, deadline] of jobs) { + if (deadline > now) continue; + jobs.delete(callback); + callback(); + } + }, + }; +} + +test("a queued child report and completion reach the recipient in one steer after two quiet seconds", async () => { + const clock = createClock(); + const sent: Array<{ recipientId: string; prompt: AgentPromptInput }> = []; + const delivery = createAgentMessageDelivery({ + schedule: clock.schedule, + dispatch: async (batch) => { + sent.push(batch); + return "steered"; + }, + onError: (error) => { + throw error; + }, + }); + const receipt = delivery.enqueue({ + recipientId: "parent", + messageId: "report", + prompt: "Results ", + source: { kind: "agent-message", agentId: "worker" }, + }); + expect(receipt).toMatchObject({ status: "queued", messageId: "report" }); + expect(sent).toEqual([]); + clock.advance(1000); + delivery.enqueue({ + recipientId: "parent", + messageId: "finish", + prompt: "Worker finished", + source: { kind: "agent-notification", agentId: "worker", event: "finished" }, + }); + clock.advance(1999); + expect(sent).toEqual([]); + clock.advance(1); + await delivery.close(); + expect( + sent.map((batch) => ({ + recipientId: batch.recipientId, + messages: parseAgentMessages(String(batch.prompt)), + })), + ).toEqual([ + { + recipientId: "parent", + messages: [ + { + id: "report", + source: { kind: "agent-message", agentId: "worker" }, + text: "Results ", + }, + { + id: "finish", + source: { kind: "agent-notification", agentId: "worker", event: "finished" }, + text: "Worker finished", + }, + ], + }, + ]); +}); + +test("recipient timers are independent, later messages form another batch, and close flushes pending work", async () => { + const clock = createClock(); + const sent: string[] = []; + const delivery = createAgentMessageDelivery({ + schedule: clock.schedule, + dispatch: async (request) => { + sent.push(`${request.recipientId}:${request.messageIds.join(",")}`); + return "steered"; + }, + onError: (error) => { + throw error; + }, + }); + const source = { kind: "agent-message" as const, agentId: "worker" }; + const a = delivery.enqueue({ recipientId: "a", messageId: "first", source, prompt: "one" }); + clock.advance(1000); + const b = delivery.enqueue({ recipientId: "b", messageId: "independent", source, prompt: "two" }); + clock.advance(1000); + await a.dispatched; + expect(sent).toEqual(["a:first"]); + delivery.enqueue({ recipientId: "a", messageId: "later", source, prompt: "three" }); + clock.advance(1000); + await b.dispatched; + expect(sent).toEqual(["a:first", "b:independent"]); + await delivery.close(); + expect(sent).toEqual(["a:first", "b:independent", "a:later"]); + expect(() => delivery.enqueue({ recipientId: "a", source, prompt: "closed" })).toThrow("closed"); +}); + +test("explicit interrupt and conflicting session modes retain ordered dispatch boundaries", async () => { + const sent: Array<{ + activeTurnBehavior: string; + sessionMode?: string; + messageIds: readonly string[]; + }> = []; + const delivery = createAgentMessageDelivery({ + dispatch: async (request) => { + sent.push(request); + request.onDispatch("steered"); + return "steered"; + }, + onError: (error) => { + throw error; + }, + }); + const source = { kind: "agent-message" as const, agentId: "worker" }; + delivery.enqueue({ + recipientId: "parent", + source, + prompt: "interrupt", + messageId: "interrupt", + activeTurnBehavior: "interrupt", + }); + delivery.enqueue({ + recipientId: "parent", + source, + prompt: "review", + messageId: "review", + sessionMode: "plan", + }); + delivery.enqueue({ + recipientId: "parent", + source, + prompt: "execute", + messageId: "execute", + sessionMode: "full-access", + }); + delivery.enqueue({ + recipientId: "parent", + source: { kind: "agent-notification", agentId: "worker", event: "finished" }, + prompt: "done", + messageId: "finished", + }); + await delivery.close(); + expect( + sent.map(({ activeTurnBehavior, sessionMode, messageIds }) => ({ + activeTurnBehavior, + sessionMode, + messageIds, + })), + ).toEqual([ + { activeTurnBehavior: "interrupt", sessionMode: undefined, messageIds: ["interrupt"] }, + { activeTurnBehavior: "steer", sessionMode: "plan", messageIds: ["review"] }, + { + activeTurnBehavior: "steer", + sessionMode: "full-access", + messageIds: ["execute", "finished"], + }, + ]); +}); + +test("dispatch failures reject waits before and after flush and do not poison later messages", async () => { + const clock = createClock(); + const failed = new Error("provider rejected"); + const errors: unknown[] = []; + const delivery = createAgentMessageDelivery({ + schedule: clock.schedule, + dispatch: async (request) => { + if (request.messageIds.includes("bad")) throw failed; + return "steered"; + }, + onError: (error) => { + errors.push(error); + }, + }); + const message = { + recipientId: "parent", + source: { kind: "agent-message" as const, agentId: "worker" }, + prompt: "report", + }; + const bad = delivery.enqueue({ ...message, messageId: "bad" }); + const before = expect( + delivery.waitForDispatch("parent", new AbortController().signal), + ).rejects.toBe(failed); + clock.advance(2000); + const after = expect( + delivery.waitForDispatch("parent", new AbortController().signal), + ).rejects.toBe(failed); + await Promise.all([before, after, expect(bad.dispatched).rejects.toBe(failed)]); + const next = delivery.enqueue({ ...message, messageId: "next" }); + await delivery.close(); + await expect(next.dispatched).resolves.toBe("steered"); + expect(errors).toEqual([failed]); +}); + +test("canceling a completion wait leaves queued delivery intact", async () => { + const clock = createClock(); + const delivery = createAgentMessageDelivery({ + schedule: clock.schedule, + dispatch: async () => "steered", + onError: (error) => { + throw error; + }, + }); + const receipt = delivery.enqueue({ + recipientId: "parent", + source: { kind: "agent-message", agentId: "worker" }, + prompt: "report", + }); + const controller = new AbortController(); + const wait = delivery.waitForDispatch("parent", controller.signal); + controller.abort(); + await expect(wait).rejects.toMatchObject({ name: "AbortError" }); + await delivery.close(); + await expect(receipt.dispatched).resolves.toBe("steered"); +}); + +test("batches retain each sender's attachment context and image order", async () => { + let prompt: AgentPromptInput = ""; + const delivery = createAgentMessageDelivery({ + dispatch: async (request) => { + prompt = request.prompt; + return "steered"; + }, + onError: (error) => { + throw error; + }, + }); + const firstImage = { type: "image" as const, data: "Zmlyc3Q=", mimeType: "image/png" }; + const secondImage = { type: "image" as const, data: "c2Vjb25k", mimeType: "image/png" }; + delivery.enqueue({ + recipientId: "parent", + messageId: "first", + source: { kind: "agent-message", agentId: "one" }, + prompt: [{ type: "text", text: "First" }, firstImage], + }); + delivery.enqueue({ + recipientId: "parent", + messageId: "second", + source: { kind: "agent-message", agentId: "two" }, + prompt: [ + { type: "text", text: "Second" }, + { + type: "github_issue", + mimeType: "application/github-issue", + number: 42, + title: "Context", + url: "https://example.com/42", + body: "Attached", + }, + secondImage, + ], + }); + await delivery.close(); + expect(prompt).toEqual([ + { + type: "text", + text: '\nFirst\n', + }, + firstImage, + { + type: "text", + text: '\nSecond\n\nGitHub Issue #42: Context\nhttps://example.com/42\n\nAttached\n', + }, + secondImage, + ]); + expect( + parseAgentMessages( + (prompt as Exclude) + .flatMap((block) => (block.type === "text" ? [block.text] : [])) + .join("\n"), + )?.map((message) => message.source?.agentId), + ).toEqual(["one", "two"]); +}); diff --git a/packages/server/src/server/agent/agent-messages/delivery.ts b/packages/server/src/server/agent/agent-messages/delivery.ts new file mode 100644 index 0000000000..e9313903d2 --- /dev/null +++ b/packages/server/src/server/agent/agent-messages/delivery.ts @@ -0,0 +1,189 @@ +import type { PromptDispatchDisposition } from "../agent-prompt.js"; +import { randomUUID } from "node:crypto"; +import type { AgentPromptInput, AgentPromptContentBlock } from "../agent-sdk-types.js"; +import { prepareAgentMessage, type AgentPromptSource } from "./envelope.js"; + +interface Message { + recipientId: string; + messageId?: string; + source: AgentPromptSource; + prompt: AgentPromptInput; + sessionMode?: string; + unarchive?: boolean; + clearPendingPermissions?: boolean; + activeTurnBehavior?: "steer" | "interrupt"; + beforeDispatch?: () => void; + onDispatch?: (disposition: PromptDispatchDisposition) => void; +} +interface Receipt { + status: "queued"; + messageId: string; + dispatched: Promise; +} +interface Admitted extends Message { + receipt: Receipt; + resolve(disposition: PromptDispatchDisposition): void; + reject(error: unknown): void; +} +interface DispatchRequest { + recipientId: string; + messageIds: readonly string[]; + prompt: AgentPromptInput; + activeTurnBehavior: "steer" | "interrupt"; + sessionMode?: string; + clearPendingPermissions: boolean; + unarchive: boolean; + onDispatch(disposition: PromptDispatchDisposition): void; +} +interface Dependencies { + dispatch(request: DispatchRequest): Promise; + onError(error: unknown, request: DispatchRequest): void; + /** Return a cancellation function. Production uses the daemon's event loop. */ + schedule?: (delayMs: number, callback: () => void) => () => void; +} +export interface AgentMessageDelivery { + enqueue(message: Message): Receipt; + /** Stop admission, flush pending messages, and await provider dispatch acknowledgments. */ + close(): Promise; + /** Existing completion waits include agent messages admitted before the wait. */ + waitForDispatch(recipientId: string, signal: AbortSignal): Promise; +} + +/** Recipient-owned agent delivery. Human prompts never enter this module. */ +export function createAgentMessageDelivery(dependencies: Dependencies): AgentMessageDelivery { + const schedule = + dependencies.schedule ?? + ((delayMs, callback) => { + const timer = setTimeout(callback, delayMs); + return () => clearTimeout(timer); + }); + const pending = new Map(); + const dispatching = new Map>(); + let closed = false; + let closing: Promise | undefined; + + function dispatch(recipientId: string, messages: Admitted[]): void { + const prompts = messages.map( + (message) => + prepareAgentMessage(message.prompt, message.source, message.receipt.messageId).prompt, + ); + const prompt: AgentPromptInput = prompts.every((entry) => typeof entry === "string") + ? prompts.join("\n\n") + : prompts.flatMap((entry): AgentPromptContentBlock[] => + typeof entry === "string" ? [{ type: "text", text: entry }] : entry, + ); + const request: DispatchRequest = { + recipientId, + messageIds: messages.map((message) => message.receipt.messageId), + prompt, + activeTurnBehavior: messages[0]!.activeTurnBehavior ?? "steer", + sessionMode: messages.find((message) => message.sessionMode)?.sessionMode, + clearPendingPermissions: messages.some((message) => message.clearPendingPermissions), + unarchive: messages.some((message) => message.unarchive !== false), + onDispatch: (disposition) => { + for (const message of messages) { + try { + message.onDispatch?.(disposition); + } catch (error) { + dependencies.onError(error, request); + } + } + }, + }; + const previous = dispatching.get(recipientId) ?? Promise.resolve(); + const dispatched = previous + .catch(() => undefined) + .then(async () => { + try { + for (const message of messages) message.beforeDispatch?.(); + const disposition = await dependencies.dispatch(request); + for (const message of messages) message.resolve(disposition); + return undefined; + } catch (error) { + for (const message of messages) message.reject(error); + dependencies.onError(error, request); + throw error; + } + }); + dispatching.set(recipientId, dispatched); + void dispatched + .finally(() => { + if (dispatching.get(recipientId) === dispatched) dispatching.delete(recipientId); + }) + .catch(() => undefined); + } + + function flush(recipientId: string): void { + const entry = pending.get(recipientId); + if (!entry) return; + pending.delete(recipientId); + entry.cancel(); + let group: Admitted[] = []; + let mode: string | undefined; + for (const message of entry.messages) { + // Explicitly conflicting send options require separate, ordered dispatches. + if ( + group.length && + ((message.activeTurnBehavior ?? "steer") !== (group[0]!.activeTurnBehavior ?? "steer") || + (mode && message.sessionMode && mode !== message.sessionMode)) + ) { + dispatch(recipientId, group); + group = []; + mode = undefined; + } + group.push(message); + mode ??= message.sessionMode; + } + if (group.length) dispatch(recipientId, group); + } + + return { + enqueue(message) { + if (closed) throw new Error("Agent message delivery is closed"); + let resolve!: Admitted["resolve"]; + let reject!: Admitted["reject"]; + const dispatched = new Promise((accepted, failed) => { + resolve = accepted; + reject = failed; + }); + void dispatched.catch(() => undefined); + const receipt: Receipt = { + status: "queued", + messageId: message.messageId ?? randomUUID(), + dispatched, + }; + const entry = pending.get(message.recipientId) ?? { messages: [], cancel: () => {} }; + entry.cancel(); + entry.messages.push({ ...message, receipt, resolve, reject }); + entry.cancel = schedule(2000, () => flush(message.recipientId)); + pending.set(message.recipientId, entry); + return receipt; + }, + async waitForDispatch(recipientId, signal) { + signal.throwIfAborted(); + const admitted = + pending.get(recipientId)?.messages.map((message) => message.receipt.dispatched) ?? []; + let aborted!: () => void; + const cancellation = new Promise((_resolve, reject) => { + aborted = () => reject(new DOMException("Wait aborted", "AbortError")); + signal.addEventListener("abort", aborted, { once: true }); + }); + try { + await Promise.race([ + Promise.all([dispatching.get(recipientId), ...admitted]), + cancellation, + ]); + } finally { + signal.removeEventListener("abort", aborted); + } + }, + close() { + if (closing) return closing; + closed = true; + for (const recipientId of pending.keys()) flush(recipientId); + // Each failure is reported and rejects its receipts; shutdown still drains other recipients. + closing = Promise.allSettled(dispatching.values()).then(() => undefined); + return closing; + }, + }; +} diff --git a/packages/server/src/server/agent/agent-messages/envelope.ts b/packages/server/src/server/agent/agent-messages/envelope.ts new file mode 100644 index 0000000000..2144d6ae02 --- /dev/null +++ b/packages/server/src/server/agent/agent-messages/envelope.ts @@ -0,0 +1,221 @@ +import { renderPromptAttachmentAsText } from "../prompt-attachments.js"; +import { randomUUID } from "node:crypto"; +import { z } from "zod"; +import type { + AgentPromptContentBlock, + AgentPromptInput, + AgentTimelineItem, +} from "../agent-sdk-types.js"; + +const agentIdSchema = z.string().min(1); +const sourceSchema = z.discriminatedUnion("kind", [ + z.object({ + kind: z.literal("agent-message"), + agentId: agentIdSchema, + serverId: z.string().min(1).optional(), + title: z.string().min(1).optional(), + }), + z.object({ + kind: z.literal("agent-notification"), + agentId: agentIdSchema, + serverId: z.string().min(1).optional(), + title: z.string().min(1).optional(), + event: z.enum(["finished", "errored", "permission-required", "closed"]), + }), +]); +export type AgentPromptSource = z.infer; + +interface AgentMessage { + id: string; + source: AgentPromptSource | null; + text: string; +} + +function escapeXml(value: string): string { + return value.replace(/[&<>"']/g, (character) => { + switch (character) { + case "&": + return "&"; + case "<": + return "<"; + case ">": + return ">"; + case '"': + return """; + default: + return "'"; + } + }); +} + +function unescapeXml(value: string): string | null { + if (/[<>]|&(?!amp;|lt;|gt;|quot;|apos;)/.test(value)) return null; + return value.replace(/&(amp|lt|gt|quot|apos);/g, (_, entity: string) => { + switch (entity) { + case "amp": + return "&"; + case "lt": + return "<"; + case "gt": + return ">"; + case "quot": + return '"'; + default: + return "'"; + } + }); +} + +export function formatAgentMessage(message: AgentMessage): string { + const { source } = message; + const attributes = source + ? `kind="${source.kind}" source-agent-id="${escapeXml(source.agentId)}"` + : 'kind="user-message"'; + const server = source?.serverId ? ` source-server-id="${escapeXml(source.serverId)}"` : ""; + const title = source?.title ? ` source-agent-title="${escapeXml(source.title)}"` : ""; + const event = source?.kind === "agent-notification" ? ` event="${source.event}"` : ""; + return `\n${escapeXml(message.text)}\n`; +} + +export function parseAgentMessage(text: string): AgentMessage | null { + const envelope = /^]*")+)>\n([\s\S]*)\n<\/paseo-system>$/.exec( + text, + ); + if (!envelope) return null; + const attributes = new Map(); + for (const match of envelope[1].matchAll(/\s+([a-z-]+)="([^"<>]*)"/g)) { + const value = unescapeXml(match[2]); + if (value === null || attributes.has(match[1])) return null; + attributes.set(match[1], value); + } + if (attributes.get("version") !== "1") return null; + const id = attributes.get("message-id"); + const body = unescapeXml(envelope[2]); + if (!id || body === null) return null; + if (attributes.get("kind") === "user-message") { + if (attributes.has("source-agent-id") || attributes.has("event")) return null; + return { id, source: null, text: body }; + } + const source = sourceSchema.safeParse({ + kind: attributes.get("kind"), + agentId: attributes.get("source-agent-id"), + serverId: attributes.get("source-server-id"), + event: attributes.get("event"), + title: attributes.get("source-agent-title"), + }); + if (!source.success) return null; + return { id, source: source.data, text: body }; +} + +/** Accept only a complete sequence of envelopes; prose around them stays user text. */ +export function parseAgentMessages(text: string): AgentMessage[] | null { + // Bodies are XML escaped, so a literal closing tag unambiguously ends an entry. + const parts = text.split(/(?<=<\/paseo-system>)\n+(?= message !== null) ? messages : null; +} + +function renderMessageText(block: AgentPromptContentBlock): string[] { + if (block.type === "image") return []; + if (block.type === "text") return [block.text]; + return [renderPromptAttachmentAsText(block)]; +} + +export function prepareAgentMessage( + prompt: AgentPromptInput, + source?: AgentPromptSource, + id?: string, +): { prompt: AgentPromptInput; messageId?: string } { + if (!source) { + const submittedText = + typeof prompt === "string" + ? prompt + : prompt + .flatMap((block) => + block.type === "text" && !("mimeType" in block) ? [block.text] : [], + ) + .join("\n") + .trim(); + // Quote reserved envelopes only when they arrived through the human send path. + // This provenance must survive provider history, not only the accepted local row. + if (!parseAgentMessages(submittedText) && !isSystemInjectedEnvelope(submittedText)) { + return { prompt, messageId: id }; + } + } + const messageId = id ?? randomUUID(); + const text = typeof prompt === "string" ? prompt : prompt.flatMap(renderMessageText).join("\n\n"); + // Render textual attachments before wrapping, so provider replay still contains one + // complete envelope instead of an envelope followed by provider-rendered context. + const envelope = formatAgentMessage({ id: messageId, source: source ?? null, text }); + return { + messageId, + prompt: + typeof prompt === "string" + ? envelope + : [{ type: "text", text: envelope }, ...prompt.filter((block) => block.type === "image")], + }; +} + +/** The same projection owns provider echoes, acceptance, import, and replay. */ +export function projectAgentMessages(item: AgentTimelineItem): AgentTimelineItem[] { + if (item.type === "user_message") { + const messages = parseAgentMessages(item.text); + if (messages) { + return messages.flatMap((message) => { + const projected = projectAgentMessage({ ...item, text: formatAgentMessage(message) }); + return projected ? [projected] : []; + }); + } + } + const projected = projectAgentMessage(item); + return projected ? [projected] : []; +} + +function projectAgentMessage(item: AgentTimelineItem): AgentTimelineItem | null { + if (item.type !== "user_message") return item; + const message = parseAgentMessage(item.text); + if (!message) { + // COMPAT(legacySystemEnvelope): added in v0.11.1; remove after 2027-04-08 once + // legacy notification history is retired. Schedules still use this envelope. + return isSystemInjectedEnvelope(item.text) ? null : item; + } + const { source } = message; + if (!source) { + return { ...item, text: message.text, clientMessageId: message.id }; + } + return { + type: "tool_call", + callId: `paseo-agent-message:${message.id}`, + agentMessage: { + event: source.kind === "agent-message" ? "message" : source.event, + sender: { + id: source.agentId, + ...(source.serverId ? { serverId: source.serverId } : {}), + ...(source.title ? { title: source.title } : {}), + }, + text: message.text, + }, + // COMPAT(agentMessageToolEnvelope): added in v0.11.1; remove after 2027-04-08 + // once clients can accept a dedicated timeline variant. Older clients still + // parse this existing shape and can expand the delivered body. + name: "agent_message", + status: "completed", + error: null, + detail: { type: "plain_text", text: message.text, icon: "bot" }, + }; +} + +/** + * Wrap a body in … so the receiving agent + * recognizes the prompt as system-injected context — not a user turn. + * Used by schedule fires; agent messages use the versioned envelope. + */ +export function formatSystemNotificationPrompt(reason: string): string { + return `\n${reason}\n`; +} + +const SYSTEM_ENVELOPE_PATTERN = /^\n[\s\S]*\n<\/paseo-system>$/; + +export function isSystemInjectedEnvelope(text: string): boolean { + return SYSTEM_ENVELOPE_PATTERN.test(text); +} diff --git a/packages/server/src/server/agent/agent-messages/index.ts b/packages/server/src/server/agent/agent-messages/index.ts index 1543bf33b4..3615993334 100644 --- a/packages/server/src/server/agent/agent-messages/index.ts +++ b/packages/server/src/server/agent/agent-messages/index.ts @@ -1,191 +1,11 @@ -import { renderPromptAttachmentAsText } from "../prompt-attachments.js"; -import { randomUUID } from "node:crypto"; -import { z } from "zod"; -import type { - AgentPromptContentBlock, - AgentPromptInput, - AgentTimelineItem, -} from "../agent-sdk-types.js"; - -const agentIdSchema = z.string().min(1); -const sourceSchema = z.discriminatedUnion("kind", [ - z.object({ - kind: z.literal("agent-message"), - agentId: agentIdSchema, - title: z.string().min(1).optional(), - }), - z.object({ - kind: z.literal("agent-notification"), - agentId: agentIdSchema, - title: z.string().min(1).optional(), - event: z.enum(["finished", "errored", "permission-required", "closed"]), - }), -]); -export type AgentPromptSource = z.infer; - -interface AgentMessage { - id: string; - source: AgentPromptSource | null; - text: string; -} - -function escapeXml(value: string): string { - return value.replace(/[&<>"']/g, (character) => { - switch (character) { - case "&": - return "&"; - case "<": - return "<"; - case ">": - return ">"; - case '"': - return """; - default: - return "'"; - } - }); -} - -function unescapeXml(value: string): string | null { - if (/[<>]|&(?!amp;|lt;|gt;|quot;|apos;)/.test(value)) return null; - return value.replace(/&(amp|lt|gt|quot|apos);/g, (_, entity: string) => { - switch (entity) { - case "amp": - return "&"; - case "lt": - return "<"; - case "gt": - return ">"; - case "quot": - return '"'; - default: - return "'"; - } - }); -} - -export function formatAgentMessage(message: AgentMessage): string { - const { source } = message; - const attributes = source - ? `kind="${source.kind}" source-agent-id="${escapeXml(source.agentId)}"` - : 'kind="user-message"'; - const title = source?.title ? ` source-agent-title="${escapeXml(source.title)}"` : ""; - const event = source?.kind === "agent-notification" ? ` event="${source.event}"` : ""; - return `\n${escapeXml(message.text)}\n`; -} - -export function parseAgentMessage(text: string): AgentMessage | null { - const envelope = /^]*")+)>\n([\s\S]*)\n<\/paseo-system>$/.exec( - text, - ); - if (!envelope) return null; - const attributes = new Map(); - for (const match of envelope[1].matchAll(/\s+([a-z-]+)="([^"<>]*)"/g)) { - const value = unescapeXml(match[2]); - if (value === null || attributes.has(match[1])) return null; - attributes.set(match[1], value); - } - if (attributes.get("version") !== "1") return null; - const id = attributes.get("message-id"); - const body = unescapeXml(envelope[2]); - if (!id || body === null) return null; - if (attributes.get("kind") === "user-message") { - if (attributes.has("source-agent-id") || attributes.has("event")) return null; - return { id, source: null, text: body }; - } - const source = sourceSchema.safeParse({ - kind: attributes.get("kind"), - agentId: attributes.get("source-agent-id"), - event: attributes.get("event"), - title: attributes.get("source-agent-title"), - }); - if (!source.success) return null; - return { id, source: source.data, text: body }; -} - -function renderMessageText(block: AgentPromptContentBlock): string[] { - if (block.type === "image") return []; - if (block.type === "text") return [block.text]; - return [renderPromptAttachmentAsText(block)]; -} - -export function prepareAgentMessage( - prompt: AgentPromptInput, - source?: AgentPromptSource, - id?: string, -): { prompt: AgentPromptInput; messageId?: string } { - if (!source) { - const submittedText = - typeof prompt === "string" - ? prompt - : prompt - .flatMap((block) => - block.type === "text" && !("mimeType" in block) ? [block.text] : [], - ) - .join("\n") - .trim(); - // Quote reserved envelopes only when they arrived through the human send path. - // This provenance must survive provider history, not only the accepted local row. - if (!parseAgentMessage(submittedText) && !isSystemInjectedEnvelope(submittedText)) { - return { prompt, messageId: id }; - } - } - const messageId = id ?? randomUUID(); - const text = typeof prompt === "string" ? prompt : prompt.flatMap(renderMessageText).join("\n\n"); - // Render textual attachments before wrapping, so provider replay still contains one - // complete envelope instead of an envelope followed by provider-rendered context. - const envelope = formatAgentMessage({ id: messageId, source: source ?? null, text }); - return { - messageId, - prompt: - typeof prompt === "string" - ? envelope - : [{ type: "text", text: envelope }, ...prompt.filter((block) => block.type === "image")], - }; -} - -/** The same projection owns provider echoes, acceptance, import, and replay. */ -export function projectAgentMessage(item: AgentTimelineItem): AgentTimelineItem | null { - if (item.type !== "user_message") return item; - const message = parseAgentMessage(item.text); - if (!message) { - // COMPAT(legacySystemEnvelope): added in v0.11.1; remove after 2027-04-08 once - // legacy notification history is retired. Schedules still use this envelope. - return isSystemInjectedEnvelope(item.text) ? null : item; - } - const { source } = message; - if (!source) { - return { ...item, text: message.text, clientMessageId: message.id }; - } - return { - type: "tool_call", - callId: `paseo-agent-message:${message.id}`, - agentMessage: { - event: source.kind === "agent-message" ? "message" : source.event, - sender: { id: source.agentId, ...(source.title ? { title: source.title } : {}) }, - text: message.text, - }, - // COMPAT(agentMessageToolEnvelope): added in v0.11.1; remove after 2027-04-08 - // once clients can accept a dedicated timeline variant. Older clients still - // parse this existing shape and can expand the delivered body. - name: "agent_message", - status: "completed", - error: null, - detail: { type: "plain_text", text: message.text, icon: "bot" }, - }; -} - -/** - * Wrap a body in … so the receiving agent - * recognizes the prompt as system-injected context — not a user turn. - * Used by schedule fires; agent messages use the versioned envelope. - */ -export function formatSystemNotificationPrompt(reason: string): string { - return `\n${reason}\n`; -} - -const SYSTEM_ENVELOPE_PATTERN = /^\n[\s\S]*\n<\/paseo-system>$/; - -export function isSystemInjectedEnvelope(text: string): boolean { - return SYSTEM_ENVELOPE_PATTERN.test(text); -} +export { + formatAgentMessage, + parseAgentMessage, + parseAgentMessages, + prepareAgentMessage, + projectAgentMessages, + formatSystemNotificationPrompt, + isSystemInjectedEnvelope, + type AgentPromptSource, +} from "./envelope.js"; +export { createAgentMessageDelivery, type AgentMessageDelivery } from "./delivery.js"; diff --git a/packages/server/src/server/agent/agent-prompt.test.ts b/packages/server/src/server/agent/agent-prompt.test.ts index dcdfdb1cbf..6ec66d10e0 100644 --- a/packages/server/src/server/agent/agent-prompt.test.ts +++ b/packages/server/src/server/agent/agent-prompt.test.ts @@ -1,10 +1,12 @@ +import { createTestAgentMessages } from "../test-utils/agent-messages.js"; import { formatSystemNotificationPrompt, isSystemInjectedEnvelope, parseAgentMessage, + parseAgentMessages, formatAgentMessage, prepareAgentMessage, - projectAgentMessage, + projectAgentMessages, } from "./agent-messages/index.js"; import { expect, it, test, vi } from "vitest"; import pino, { type Logger } from "pino"; @@ -143,6 +145,11 @@ function createFinishNotificationScenario( return { startWatchingChild() { setupFinishNotification({ + agentMessages: createTestAgentMessages({ + agentManager, + agentStorage, + logger: options?.logger ?? createTestLogger(), + }), agentManager, agentStorage, childAgentId: "child-agent", @@ -394,22 +401,25 @@ test("finish notifications report every concurrently pending permission", async scenario.requestChildPermission("permission-1"); scenario.requestChildPermission("permission-2"); - await vi.waitFor(() => expect(scenario.parentPrompts()).toHaveLength(2)); + await vi.waitFor(() => expect(scenario.parentPrompts()).toHaveLength(1)); expect( - scenario.parentPrompts().map((prompt) => { - const payload = parseAgentMessage(prompt)!.text.match( - /\n([\s\S]+?)\n<\/permission-request>/, - )?.[1]; - return JSON.parse(payload!).requestId; - }), + 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(3)); - expect(scenario.parentPrompts()[2]).toContain("finished."); + await vi.waitFor(() => expect(scenario.parentPrompts()).toHaveLength(2)); + expect(scenario.parentPrompts()[1]).toContain("finished."); }); test("finish notifications survive repeated permission cycles", async () => { @@ -464,10 +474,9 @@ test("finish notifications log a rejected parent prompt without an unhandled rej expect(captured.records).toEqual([ expect.objectContaining({ - msg: "Failed to notify caller agent", - childAgentId: "child-agent", - callerAgentId: "caller-agent", - reason: "finished", + msg: "Agent message batch dispatch failed", + agentId: "caller-agent", + messageIds: [expect.any(String)], err: expect.objectContaining({ message: "parent provider rejected replacement" }), }), ]); @@ -525,6 +534,11 @@ it("does not notify archived callers", async () => { Reflect.set(agentStorage, "get", agentStorageGetSpy); setupFinishNotification({ + agentMessages: createTestAgentMessages({ + agentManager, + agentStorage, + logger: createTestLogger(), + }), agentManager, agentStorage, childAgentId: "child-agent", @@ -808,6 +822,36 @@ test("agent envelopes round-trip opaque sender IDs and XML-sensitive messages", expect(parseAgentMessage(encoded.replace('version="1"', 'version="1" version="1"'))).toBeNull(); }); +test("a delivered batch projects each message and completion separately in order", () => { + const entries = [ + { + id: "report", + source: { kind: "agent-message" as const, agentId: "remote::worker", title: "Worker" }, + text: "Result ", + }, + { + id: "finished", + source: { + kind: "agent-notification" as const, + agentId: "remote::worker", + event: "finished" as const, + }, + text: "Finished & ready", + }, + ]; + const text = entries.map(formatAgentMessage).join("\n\n"); + expect(projectAgentMessages({ type: "user_message", text })).toEqual( + entries.flatMap((entry) => + projectAgentMessages({ type: "user_message", text: formatAgentMessage(entry) }), + ), + ); + // Human submissions containing the same envelopes must survive replay as human text. + const pasted = prepareAgentMessage(text, undefined, "human-batch"); + expect(projectAgentMessages({ type: "user_message", text: String(pasted.prompt) })).toEqual([ + { type: "user_message", text, clientMessageId: "human-batch" }, + ]); +}); + test("agent envelope includes rendered attachment context and preserves images", () => { const image = { type: "image" as const, data: "aW1hZ2U=", mimeType: "image/png" }; const attachment = { @@ -849,11 +893,11 @@ test("a human pasting an agent envelope stays a human message through provider r }); const delivery = prepareAgentMessage(pasted, undefined, "human-submission"); expect( - projectAgentMessage({ + projectAgentMessages({ type: "user_message", text: String(delivery.prompt), messageId: "provider-echo", - }), + })[0], ).toEqual({ type: "user_message", text: pasted, diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index 6b120abb1c..63c2309956 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -1,5 +1,9 @@ import type { WaitForFinishResult } from "@getpaseo/client/internal/daemon-client"; -import { prepareAgentMessage, type AgentPromptSource } from "./agent-messages/index.js"; +import { + prepareAgentMessage, + type AgentPromptSource, + type AgentMessageDelivery, +} from "./agent-messages/index.js"; import type { Logger } from "pino"; import type { @@ -316,6 +320,21 @@ async function resolvePromptSource( */ export async function sendPromptToAgent( params: SendPromptToAgentParams, +): Promise<{ disposition: PromptDispatchDisposition }> { + const source = await resolvePromptSource(params.source, params.agentManager, params.agentStorage); + const delivery = prepareAgentMessage(params.prompt, source, params.messageId); + return sendPreparedPromptToAgent({ + ...params, + prompt: delivery.prompt, + runOptions: delivery.messageId + ? { ...params.runOptions, clientMessageId: delivery.messageId } + : params.runOptions, + }); +} + +/** Dispatch daemon-prepared context without reclassifying it as human input. */ +export async function sendPreparedPromptToAgent( + params: Omit, ): Promise<{ disposition: PromptDispatchDisposition }> { const unarchive = params.unarchive ?? true; @@ -349,18 +368,12 @@ export async function sendPromptToAgent( await params.agentManager.setAgentMode(params.agentId, params.sessionMode); } - const source = await resolvePromptSource(params.source, params.agentManager, params.agentStorage); - const delivery = prepareAgentMessage(params.prompt, source, params.messageId); - const runOptions = delivery.messageId - ? { ...params.runOptions, clientMessageId: delivery.messageId } - : params.runOptions; - - return await startAgentRun(params.agentManager, params.agentId, delivery.prompt, params.logger, { + return await startAgentRun(params.agentManager, params.agentId, params.prompt, params.logger, { replaceRunning: true, onDispatch: params.onDispatch, activeTurnBehavior: params.activeTurnBehavior, clearPendingPermissions: params.clearPendingPermissions, - runOptions, + runOptions: params.runOptions, }); } @@ -401,6 +414,7 @@ export async function startCreatedAgentInitialPrompt( } export interface SetupFinishNotificationParams { + agentMessages: AgentMessageDelivery; agentManager: AgentManager; agentStorage: AgentStorage; childAgentId: string; @@ -490,6 +504,7 @@ const armedFinishNotifications = new WeakMap< /** Creation waits and accepted follow-ups share notification ownership. */ export function setupFinishWait(params: { + agentMessages: AgentMessageDelivery; agentManager: AgentManager; agentStorage: AgentStorage; originServerId?: string; @@ -545,10 +560,8 @@ export function setupFinishWait(params: { result.status ] as FinishNotificationReason; const permissionRequest = result.final?.pendingPermissions[0]; - await sendPromptToAgent({ - agentManager: params.agentManager, - agentStorage: params.agentStorage, - agentId: params.callerAgentId, + params.agentMessages.enqueue({ + recipientId: params.callerAgentId, prompt: formatFinishNotificationBody({ childAgentId: params.childAgentId, serverId: params.childServerId, @@ -560,12 +573,12 @@ export function setupFinishWait(params: { source: { kind: "agent-notification", agentId: params.childAgentId, + serverId: params.childServerId, title: result.final?.title ?? params.childAgentId, event: finishNotificationEvents[reason], }, activeTurnBehavior: "steer", unarchive: false, - logger: params.logger, }); if (result.status !== "permission" || !permissionRequest) return; await params.afterPermission(permissionRequest.id, controller.signal); @@ -677,19 +690,17 @@ export function setupFinishNotification(params: SetupFinishNotificationParams): permissionRequest, }); - await sendPromptToAgent({ - agentManager, - agentStorage, - agentId: callerAgentId, + params.agentMessages.enqueue({ + recipientId: callerAgentId, prompt: body, source: { kind: "agent-notification", agentId: childAgentId, + title, event: finishNotificationEvents[reason], }, activeTurnBehavior: "steer", unarchive: false, - logger, }); } diff --git a/packages/server/src/server/agent/create-agent/create.ts b/packages/server/src/server/agent/create-agent/create.ts index f9fe10bd09..79f674e389 100644 --- a/packages/server/src/server/agent/create-agent/create.ts +++ b/packages/server/src/server/agent/create-agent/create.ts @@ -1,4 +1,4 @@ -import type { AgentPromptSource } from "../agent-messages/index.js"; +import type { AgentPromptSource, AgentMessageDelivery } from "../agent-messages/index.js"; import type { Logger } from "pino"; import type { TerminalManager } from "../../../terminal/terminal-manager.js"; @@ -23,6 +23,7 @@ import { resolveRequiredProviderModel, type ResolvedProviderModel } from "../mcp import { resolveCreateAgentIntent } from "./intent.js"; export interface CreateAgentCommandDependencies { + agentMessages?: AgentMessageDelivery; agentManager: AgentManager; agentStorage: AgentStorage; logger: Logger; @@ -180,6 +181,7 @@ export async function createAgentCommand( input.callerAgentId && resolved.prompt !== undefined ? setupFinishNotification({ + agentMessages: requireAgentMessages(dependencies), agentManager: dependencies.agentManager, agentStorage: dependencies.agentStorage, childAgentId: snapshot.id, @@ -558,3 +560,8 @@ async function resolveMcpCwd(params: { createdWorktree, }; } + +function requireAgentMessages(dependencies: CreateAgentCommandDependencies): AgentMessageDelivery { + if (!dependencies.agentMessages) throw new Error("Agent message delivery is not configured"); + return dependencies.agentMessages; +} diff --git a/packages/server/src/server/agent/host-operations/index.ts b/packages/server/src/server/agent/host-operations/index.ts index b69f0775a9..294e869555 100644 --- a/packages/server/src/server/agent/host-operations/index.ts +++ b/packages/server/src/server/agent/host-operations/index.ts @@ -207,10 +207,17 @@ export function createHostResolver(input: { }, async sendPrompt(args) { const agentId = args.agentId as string; - if (args.sessionMode !== undefined) + // Human sends retain the established mode RPC on older connected hosts. + // Agent sends apply their requested mode when the recipient dispatches the batch. + if (!input.callerAgentId && args.sessionMode !== undefined) { await client.setAgentMode(agentId, args.sessionMode as string); - await client.sendMessage(agentId, args.prompt 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"), @@ -219,10 +226,11 @@ export function createHostResolver(input: { if (!agent) throw new Error(`Agent ${agentId} not found`); return result({ success: true, + ...receipt, status: agent.status, guidance: - input.callerAgentId && args.notifyOnFinish !== false - ? "You will get notified when the prompted agent finishes, errors, or needs permission. Do not poll for status; continue with other work until the notification arrives." + receipt.deliveryStatus === "queued" + ? "Queued for delivery after two quiet seconds. You will get notified when the prompted agent finishes, errors, or needs permission unless notifyOnFinish is false." : "Dispatch acknowledged. Observe completion using get_agent_status/get_agent_activity or SDK/RPC subscriptions and wait APIs.", }); }, diff --git a/packages/server/src/server/agent/mcp-server.test.ts b/packages/server/src/server/agent/mcp-server.test.ts index e774820dee..0599281f15 100644 --- a/packages/server/src/server/agent/mcp-server.test.ts +++ b/packages/server/src/server/agent/mcp-server.test.ts @@ -1,3 +1,5 @@ +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 { execFileSync } from "node:child_process"; @@ -123,6 +125,9 @@ interface RegisteredMcpToolWithHandler extends RegisteredMcpTool { }>; } +const serverDeliveries = new WeakMap(); +const toolDeliveries = new WeakMap(); + function lookupTool( server: Awaited>, name: string, @@ -143,7 +148,10 @@ function registeredTool( if (!handler) { throw new Error(`MCP tool has no callable handler: ${name}`); } - return { ...tool, handler }; + const registered = { ...tool, handler }; + const delivery = serverDeliveries.get(server); + if (delivery) toolDeliveries.set(registered, delivery); + return registered; } async function invokeToolWithParsedInput( @@ -152,7 +160,25 @@ async function invokeToolWithParsedInput( ) { const parsed = await tool.inputSchema.safeParseAsync(input); expect(parsed.success).toBe(true); - return tool.handler(parsed.data); + const response = await tool.handler(parsed.data); + if (response.structuredContent.deliveryStatus === "queued") { + await toolDeliveries + .get(tool)! + .waitForDispatch(input.agentId as string, new AbortController().signal); + } + return response; +} + +async function sendMcpAndAwaitDispatch( + client: Client, + server: object, + input: Record, +) { + const response = await client.callTool({ name: "send_agent_prompt", arguments: input }); + await serverDeliveries + .get(server)! + .waitForDispatch(input.agentId as string, new AbortController().signal); + return response; } function agentsOf(response: { @@ -812,11 +838,17 @@ function createPaseoWorktreeForMcpTest(options: { } // Typed in-memory workspace port for catalogue tests without workspace setup. -function createAgentMcpServer(options: Parameters[0]) { - return createAgentMcpServerWithDependencies({ +async function createAgentMcpServer( + options: Parameters[0], +) { + const messages = createTestAgentMessages(options); + const server = await createAgentMcpServerWithDependencies({ + agentMessages: () => messages, workspaceRegistry: { get: async () => null, list: async () => [], upsert: async () => {} }, ...options, }); + serverDeliveries.set(server, messages); + return server; } describe("bundled Paseo skill", () => { @@ -3601,7 +3633,9 @@ describe("send_agent_prompt MCP tool", () => { }); expect(response.structuredContent).toMatchObject({ success: true, - disposition: steered ? "steered" : "turn_started", + ...(agentScoped + ? { deliveryStatus: "queued" } + : { disposition: steered ? "steered" : "turn_started" }), }); expect(client.sessions[1]!.prompts).toHaveLength(steered ? 1 : 2); expect(client.sessions[1]!.steeredPrompts).toHaveLength(steered ? 1 : 0); @@ -3648,7 +3682,7 @@ describe("send_agent_prompt MCP tool", () => { }); mcpClient = await connectInMemoryMcpClient(server); const send = (input: Record) => - mcpClient!.callTool({ name: "send_agent_prompt", arguments: input }); + sendMcpAndAwaitDispatch(mcpClient!, server, input); await send({ agentId: child.id, prompt: "Original work", @@ -3660,10 +3694,7 @@ describe("send_agent_prompt MCP tool", () => { notifyOnFinish: true, }); if (rejected) { - expect(await sending).toMatchObject({ - isError: true, - content: [{ type: "text", text: expect.stringContaining("Steer rejected") }], - }); + await expect(sending).rejects.toThrow("Steer rejected"); // Flush asynchronous notification work before checking that a rejected // dispatch did not notify. Then prove a later successful send still can. await new Promise((resolve) => setTimeout(resolve, 50)); @@ -3677,7 +3708,7 @@ describe("send_agent_prompt MCP tool", () => { const response = await sending; expect(response.structuredContent).toMatchObject({ success: true, - disposition: "steered", + deliveryStatus: "queued", }); } await vi.waitFor(() => expect(client.sessions[0]!.prompts).toHaveLength(1)); @@ -3727,7 +3758,7 @@ describe("send_agent_prompt MCP tool", () => { }); mcpClient = await connectInMemoryMcpClient(server); const send = (input: Record) => - mcpClient!.callTool({ name: "send_agent_prompt", arguments: input }); + sendMcpAndAwaitDispatch(mcpClient!, server, input); await send({ agentId: child.id, prompt: "Original work", @@ -3738,10 +3769,7 @@ describe("send_agent_prompt MCP tool", () => { prompt: "New message", notifyOnFinish: true, }); - expect(await sending).toMatchObject({ - isError: true, - content: [{ type: "text", text: expect.stringContaining("Steer rejected") }], - }); + await expect(sending).rejects.toThrow("Steer rejected"); if (!finishDuringSteer) { await new Promise((resolve) => setTimeout(resolve, 50)); expect(client.sessions[0]!.prompts).toHaveLength(0); @@ -3799,7 +3827,7 @@ describe("send_agent_prompt MCP tool", () => { }); mcpClient = await connectInMemoryMcpClient(server); const send = (input: Record) => - mcpClient!.callTool({ name: "send_agent_prompt", arguments: input }); + sendMcpAndAwaitDispatch(mcpClient!, server, input); await send({ agentId: child.id, prompt: "Original work", @@ -3815,6 +3843,7 @@ describe("send_agent_prompt MCP tool", () => { }); for (const dispatchResult of [first, second]) { setupFinishNotification({ + agentMessages: serverDeliveries.get(server)!, agentManager: manager, agentStorage: storage, childAgentId: child.id, @@ -3891,11 +3920,14 @@ describe("send_agent_prompt MCP tool", () => { const response = await tool.handler(parsed.data as Record); + expect(response.structuredContent).toMatchObject({ deliveryStatus: "queued" }); + expect(spies.agentManager.subscribe).not.toHaveBeenCalled(); + await serverDeliveries + .get(server)! + .waitForDispatch("child-agent", new AbortController().signal); expect(spies.agentManager.subscribe).toHaveBeenCalled(); expect(spies.agentManager.waitForAgentEvent).not.toHaveBeenCalled(); - expect(response.structuredContent.guidance).toBe( - "You will get notified when the prompted agent finishes, errors, or needs permission. Do not poll for status; continue with other work until the notification arrives.", - ); + expect(response.structuredContent.guidance).toContain("You will get notified"); }); it("acknowledges top-level prompts without a completion wait", async () => { @@ -4039,10 +4071,10 @@ describe("send_agent_prompt MCP tool", () => { unsubscribe(); expect(response.structuredContent).toMatchObject({ success: true, - disposition: "turn_started", + deliveryStatus: "queued", }); if (!immediateFinish) { - expect(response.structuredContent.status).toBe("running"); + expect(response.structuredContent.deliveryStatus).toBe("queued"); if (failure) childClient.sessions[0]!.failTurn("Retry failed"); else childClient.sessions[0]!.finishTurn("Retry done"); } @@ -4105,8 +4137,7 @@ describe("send_agent_prompt MCP tool", () => { }); expect(response.structuredContent).toMatchObject({ success: true, - disposition: "out_of_band", - status: "running", + deliveryStatus: "queued", }); expect(childClient.sessions[0]!.prompts).toHaveLength(1); expect(parseAgentMessage(childClient.sessions[0]!.prompts[0])).toMatchObject({ @@ -4177,9 +4208,8 @@ describe("send_agent_prompt MCP tool", () => { }); const response = await pending; expect(response.structuredContent).toMatchObject({ - status: "running", - guidance: - "You will get notified when the prompted agent finishes, errors, or needs permission. Do not poll for status; continue with other work until the notification arrives.", + deliveryStatus: "queued", + guidance: expect.stringContaining("You will get notified"), }); expect(response.structuredContent).not.toHaveProperty("lastMessage"); @@ -4237,7 +4267,7 @@ describe("send_agent_prompt MCP tool", () => { }); expect(sent.structuredContent).toMatchObject({ success: true, - status: "running", + deliveryStatus: "queued", }); const childSession = childClient.sessions[0]!; await vi.waitFor(() => expect(childSession.prompts).toHaveLength(2)); @@ -4262,7 +4292,7 @@ describe("send_agent_prompt MCP tool", () => { } }); - it("reports a background prompt's accepted turn as running", async () => { + it("reports queue admission before the background prompt starts", async () => { const workdir = await mkdtemp(join(tmpdir(), "mcp-background-send-status-")); const storage = new AgentStorage(join(workdir, "agents"), logger); const parentClient = new HeldTurnAgentClient("claude", false); @@ -4308,7 +4338,7 @@ describe("send_agent_prompt MCP tool", () => { acknowledgeTurnStart(); const response = await pending; - expect(response.structuredContent).toMatchObject({ success: true, status: "running" }); + expect(response.structuredContent).toMatchObject({ success: true, deliveryStatus: "queued" }); expect(agentManager.getAgent(child.id)?.lifecycle).toBe("running"); childSession.finishTurn(); @@ -4378,7 +4408,7 @@ describe("send_agent_prompt MCP tool", () => { unsubscribe(); expect(response.structuredContent).toMatchObject({ success: true, - disposition: "turn_started", + deliveryStatus: "queued", }); await vi.waitFor(() => expect(agentManager.getAgent(child.id)?.lifecycle).toBe("idle")); await vi.waitFor(() => { @@ -4479,7 +4509,7 @@ describe("send_agent_prompt MCP tool to an OMP child", () => { expect(response.structuredContent).toMatchObject({ success: true, - disposition: "out_of_band", + deliveryStatus: "queued", status: "running", }); await vi.waitFor(() => { @@ -4547,8 +4577,8 @@ describe("send_agent_prompt MCP tool to an OMP child", () => { expect(response.structuredContent).toMatchObject({ success: true, - disposition: "out_of_band", - status: "running", + deliveryStatus: "queued", + status: "idle", }); await vi.waitFor(() => { expect(ompSession.prompts).toEqual([ diff --git a/packages/server/src/server/agent/providers/opencode/bridge.test.ts b/packages/server/src/server/agent/providers/opencode/bridge.test.ts index 8fc6c23d93..33f43c4f79 100644 --- a/packages/server/src/server/agent/providers/opencode/bridge.test.ts +++ b/packages/server/src/server/agent/providers/opencode/bridge.test.ts @@ -15,6 +15,7 @@ import { AgentManager } from "../../agent-manager.js"; import { AgentStorage } from "../../agent-storage.js"; import { ProviderSnapshotManager } from "../../provider-snapshot-manager.js"; import { createTestAgentClients } from "../../../test-utils/fake-agent-client.js"; +import { createTestAgentMessages } from "../../../test-utils/agent-messages.js"; import { createPaseoAgentToolManifest, createPaseoToolCatalog } from "../../tools/paseo-tools.js"; import { serializePaseoToolInputParameters } from "../../tools/paseo-tool-serialization.js"; import { @@ -83,7 +84,14 @@ describe("OpenCodeBridge", () => { const agentStorage = new AgentStorage(path.join(paseoHome, "agents"), logger); const agentManager = new AgentManager({ clients, registry: agentStorage, logger }); const providerSnapshotManager = new ProviderSnapshotManager({ logger, extraClients: clients }); - const dependencies = { agentManager, agentStorage, providerSnapshotManager, logger }; + const agentMessages = createTestAgentMessages({ agentManager, agentStorage, logger }); + const dependencies = { + agentManager, + agentStorage, + providerSnapshotManager, + logger, + agentMessages: () => agentMessages, + }; const bridge = new OpenCodeBridge({ paseoHome, logger }); await bridge.start(); bridge.setManifestCatalog(createPaseoAgentToolManifest(dependencies)); @@ -199,11 +207,13 @@ describe("OpenCodeBridge", () => { sessionID: "bound", ask: async () => undefined, }), - ).resolves.toMatchObject({ output: expect.stringContaining('"success": true') }); + ).resolves.toMatchObject({ output: expect.stringContaining('"deliveryStatus": "queued"') }); + await agentMessages.waitForDispatch(child.id, new AbortController().signal); } finally { await dispose(); } } finally { + await agentMessages.close(); await bridge.close(); await Promise.all( agentManager.listAgents().map((agent) => agentManager.closeAgent(agent.id)), diff --git a/packages/server/src/server/agent/tools/paseo-tools.ts b/packages/server/src/server/agent/tools/paseo-tools.ts index 97ded06e69..74c4361c9e 100644 --- a/packages/server/src/server/agent/tools/paseo-tools.ts +++ b/packages/server/src/server/agent/tools/paseo-tools.ts @@ -1,3 +1,4 @@ +import type { AgentMessageDelivery } from "../agent-messages/index.js"; import { createHostResolver, waitForPermissionChange, @@ -116,6 +117,7 @@ import type { ProviderPaseoToolsPolicy } from "@getpaseo/protocol/provider-confi import { isPaseoToolEnabled } from "../paseo-tool-policy.js"; export interface PaseoToolHostDependencies { + agentMessages?: () => AgentMessageDelivery; serverId?: string; hosts?: () => ToolHosts; agentManager: AgentManager; @@ -657,6 +659,10 @@ function createToolCatalog( return result.data; }; + function agentMessages(): AgentMessageDelivery { + if (!options.agentMessages) throw new Error("Agent message delivery is not configured"); + return options.agentMessages(); + } const localHostOperations = new Map(); const tools = new Map(); const hostTools = new Map([ @@ -758,6 +764,7 @@ function createToolCatalog( } if (childAgentId && callerAgentId && input.notifyOnFinish !== false) { setupFinishWait({ + agentMessages: agentMessages(), agentManager, agentStorage, originServerId: options.serverId, @@ -1213,7 +1220,7 @@ function createToolCatalog( .optional() .default(true) .describe( - "Get notified when the created agent finishes, errors, or needs permission. Set false only for truly fire-and-forget agents.", + "Get notified when the created agent finishes, errors, or needs permission. Set false for explicit reporting or no completion callback.", ), }; const canonicalTopLevelInputSchema = { @@ -1274,7 +1281,7 @@ function createToolCatalog( .optional() .default(true) .describe( - "Get notified when the prompted agent finishes, errors, or needs permission. Set false only for truly fire-and-forget prompts.", + "Get notified when the prompted agent finishes, errors, or needs permission. Set false for explicit reporting or no completion callback.", ), }; const topLevelSendAgentPromptInputSchema = { @@ -1941,11 +1948,13 @@ function createToolCatalog( { title: "Send agent prompt", description: - "Send a task and return after dispatch acknowledgement. Calling agents receive finish notifications by default.", + "Send a task. Agent callers receive an immediate queued receipt; delivery follows after two quiet seconds per recipient. Completion reports include the target’s last response by default. Paseo supplies caller identity.", inputSchema: sendAgentPromptInputSchema, outputSchema: { success: z.boolean(), disposition: z.enum(["turn_started", "out_of_band", "steered"]).optional(), + deliveryStatus: z.literal("queued").optional(), + messageId: z.string().optional(), status: AgentStatusEnum, guidance: z.string().optional(), }, @@ -1963,8 +1972,10 @@ function createToolCatalog( acceptDispatch = resolve; }); const notifying = Boolean(callerAgentId && notifyOnFinish); - if (callerAgentId && notifyOnFinish) { + function prepareDispatch(): void { + if (!callerAgentId || !notifyOnFinish) return; stopNotification = setupFinishNotification({ + agentMessages: agentMessages(), agentManager, agentStorage, childAgentId: agentId, @@ -1986,6 +1997,7 @@ function createToolCatalog( if (!callerAgentId || !notifyOnFinish) return; stopNotification?.(); stopNotification = setupFinishNotification({ + agentMessages: agentMessages(), agentManager, agentStorage, childAgentId: agentId, @@ -1996,6 +2008,41 @@ function createToolCatalog( }); } + 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, + }); + void receipt.dispatched.catch(() => { + acceptDispatch(false); + stopNotification?.(); + }); + return { + content: [], + structuredContent: { + success: true, + deliveryStatus: receipt.status, + messageId: receipt.messageId, + status: agentManager.getAgent(agentId)?.lifecycle ?? "idle", + guidance: notifying + ? `Queued for delivery after two quiet seconds. ${PROMPTED_AGENT_NOTIFICATION_GUIDANCE}` + : "Queued for delivery after two quiet seconds; provider dispatch has not occurred.", + }, + }; + } + const { disposition } = await sendPromptToAgent({ agentManager, agentStorage, diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 5125a0b923..09be15772e 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1393,6 +1393,10 @@ export async function createPaseoDaemon( const createAgentToolHostDependencies = ( runtime: PaseoToolRuntimeContext, ): PaseoToolHostDependencies => ({ + agentMessages: () => { + if (!wsServer) throw new Error("Daemon is not accepting agent messages"); + return wsServer.agentMessages; + }, serverId, hosts: () => { if (!wsServer) throw new Error("Daemon is not accepting host operations"); @@ -1830,6 +1834,7 @@ export async function createPaseoDaemon( scriptHealthMonitor.stop(); // Freeze both ingress and registration before taking the agent closure snapshot. wsServer?.prepareForShutdown(); + await wsServer?.agentMessages.close(); agentManager.prepareForShutdown(); await closeAllAgents(logger, agentManager); await withTimeout({ 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 0efd8d86d6..e24e431cd0 100644 --- a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts +++ b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts @@ -433,7 +433,7 @@ class RoutePeer { } } -import { mkdir, writeFile, readFile, chmod, stat, mkdtemp, rm } from "node:fs/promises"; +import { mkdir, writeFile, readFile, chmod, stat, mkdtemp, rm, readdir } from "node:fs/promises"; import { join, dirname, basename } from "node:path"; import { tmpdir, userInfo } from "node:os"; import { resolveDaemonVersion } from "../daemon-version.js"; @@ -1310,173 +1310,408 @@ test("cancelling a logical scope during admission settles it and releases its ow await delivery.close(); }); -test("agent MCP discovers hosts, creates children on self and B, and notifies their parent", async () => { - const parentPrompts: string[] = []; - const a = await daemon({ - agentClients: createTestAgentClients({ - onStartTurn: (prompt, config) => { - if (config.title === "Fleet parent") parentPrompts.push(JSON.stringify(prompt)); - }, - }), - }); - const b = await daemon(); - const { paseo, client: aClient } = await api(a); - await paseo.hosts.add({ connection: `127.0.0.1:${b.port}`, label: "Fleet B" }); - const directory = await online(paseo); - const workspace = await paseo.workspaces.create({ - source: { kind: "directory", path: a.staticDir }, - }); - const parent = await workspace.agents.create({ - config: { provider: "claude/claude-test" }, - title: "Fleet parent", - }); - const mcp = new Client({ name: "fleet-delegation", 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 result = await mcp.callTool({ name: "list_hosts", arguments: {} }); - expect(result.isError).not.toBe(true); - expect(result.structuredContent).toMatchObject({ - self: { serverId: directory.self.serverId }, - hosts: [expect.objectContaining({ serverId: directory.hosts[0]!.serverId, label: "Fleet B" })], - }); - const call = async (name: string, args: Record) => { - const response = await mcp.callTool({ name, arguments: args }); - expect(response.isError, JSON.stringify(response)).not.toBe(true); - return z.record(z.string(), z.unknown()).parse(response.structuredContent); - }; - const serverId = directory.hosts[0]!.serverId!; - const destination = await call("create_workspace", { - serverId, - isolation: "local", - path: b.staticDir, - title: "Destination workspace", - }); - expect(await call("list_workspaces", { serverId })).toMatchObject({ - serverId, - workspaces: [expect.objectContaining({ workspaceId: destination.workspaceId })], - }); - const self = await call("create_agent", { - title: "Self child", - provider: "claude/claude-test", - initialPrompt: "Finish local delegation", - }); - const remote = await call("create_agent", { - serverId, - workspaceId: destination.workspaceId, - title: "B child", - provider: "claude/claude-test", - initialPrompt: "Finish destination delegation", - }); - expect(self.serverId).toBe(directory.self.serverId); - expect(remote.serverId).toBe(serverId); - const bApi = await api(b); - expect( - (await bApi.paseo.agents.ref(String(remote.agentId)).refresh())?.agent.labels, - ).toMatchObject({ - "paseo.parent-agent-id": parent.id, - "paseo.parent-server-id": directory.self.serverId, - }); - await expect.poll(() => parentPrompts.join("\n")).toContain(`Agent ${self.agentId}`); - await expect.poll(() => parentPrompts.join("\n")).toContain(`Agent ${remote.agentId}`); - expect(parentPrompts.join("\n")).toContain(`Server: ${serverId}`); - expect(parentPrompts.join("\n")).toContain(`Server: ${directory.self.serverId}`); - - await aClient.createAgent({ - agentId: String(remote.agentId), - provider: "claude", - model: "claude-test", - cwd: a.staticDir, - workspaceId: workspace.id, - title: "Unrelated equal ID on A", - }); - expect(await call("get_agent_status", { serverId, agentId: remote.agentId })).toMatchObject({ - serverId, - snapshot: { id: remote.agentId, title: "B child", cwd: b.staticDir }, - }); - expect(await call("get_agent_status", { agentId: remote.agentId })).toMatchObject({ - serverId: directory.self.serverId, - snapshot: { title: "Unrelated equal ID on A" }, - }); - const beforeFollowup = parentPrompts.filter((prompt) => - prompt.includes(`Agent ${remote.agentId}`), - ).length; - expect(beforeFollowup).toBe(1); - expect( - await call("send_agent_prompt", { - serverId, - agentId: remote.agentId, - prompt: "REMOTE_FOLLOWUP_MARKER", - sessionMode: "full-access", - }), - ).toMatchObject({ success: true, serverId, guidance: expect.stringContaining("notified") }); - expect(await call("get_agent_activity", { serverId, agentId: remote.agentId })).toMatchObject({ - serverId, - agentId: remote.agentId, - currentModeId: "full-access", - content: expect.stringContaining("REMOTE_FOLLOWUP_MARKER"), - }); - expect(await call("list_agents", { serverId })).toMatchObject({ - serverId, - agents: expect.arrayContaining([ - expect.objectContaining({ id: remote.agentId, title: "B child", cwd: b.staticDir }), - ]), - }); - expect(await call("list_agents", {})).toMatchObject({ - serverId: directory.self.serverId, - agents: expect.arrayContaining([ - expect.objectContaining({ - id: remote.agentId, - title: "Unrelated equal ID on A", - cwd: a.staticDir, +test( + "agent MCP discovers hosts, creates children on self and B, and notifies their parent", + { timeout: 30000 }, + async () => { + const parentPrompts: string[] = []; + const a = await daemon({ + agentClients: createTestAgentClients({ + onStartTurn: (prompt, config) => { + if (config.title === "Fleet parent") parentPrompts.push(JSON.stringify(prompt)); + }, }), - ]), - }); - await expect - .poll(() => parentPrompts.filter((prompt) => prompt.includes(`Agent ${remote.agentId}`)).length) - .toBe(2); - expect(parentPrompts.findLast((prompt) => prompt.includes(`Agent ${remote.agentId}`))).toContain( - `Server: ${serverId}`, - ); - expect((await aClient.fetchAgent(String(remote.agentId)))?.agent.lastUserMessageAt).toBeNull(); - expect(JSON.stringify(await aClient.fetchAgentTimeline(String(remote.agentId)))).not.toContain( - "REMOTE_FOLLOWUP_MARKER", - ); - - for (const selector of [{}, { serverId: directory.self.serverId }]) { - expect(await call("get_agent_status", { ...selector, agentId: self.agentId })).toMatchObject({ - serverId: directory.self.serverId, - snapshot: { title: "Self child" }, }); - const previous = parentPrompts.filter((prompt) => - prompt.includes(`Agent ${self.agentId}`), + const b = await daemon(); + const { paseo, client: aClient } = await api(a); + await paseo.hosts.add({ connection: `127.0.0.1:${b.port}`, label: "Fleet B" }); + const directory = await online(paseo); + const workspace = await paseo.workspaces.create({ + source: { kind: "directory", path: a.staticDir }, + }); + const parent = await workspace.agents.create({ + config: { provider: "claude/claude-test" }, + title: "Fleet parent", + }); + const mcp = new Client({ name: "fleet-delegation", 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 result = await mcp.callTool({ name: "list_hosts", arguments: {} }); + expect(result.isError).not.toBe(true); + expect(result.structuredContent).toMatchObject({ + self: { serverId: directory.self.serverId }, + hosts: [ + expect.objectContaining({ serverId: directory.hosts[0]!.serverId, label: "Fleet B" }), + ], + }); + const call = async (name: string, args: Record) => { + const response = await mcp.callTool({ name, arguments: args }); + expect(response.isError, JSON.stringify(response)).not.toBe(true); + return z.record(z.string(), z.unknown()).parse(response.structuredContent); + }; + const serverId = directory.hosts[0]!.serverId!; + const destination = await call("create_workspace", { + serverId, + isolation: "local", + path: b.staticDir, + title: "Destination workspace", + }); + expect(await call("list_workspaces", { serverId })).toMatchObject({ + serverId, + workspaces: [expect.objectContaining({ workspaceId: destination.workspaceId })], + }); + const self = await call("create_agent", { + title: "Self child", + provider: "claude/claude-test", + initialPrompt: "Finish local delegation", + }); + const remote = await call("create_agent", { + serverId, + workspaceId: destination.workspaceId, + title: "B child", + provider: "claude/claude-test", + initialPrompt: "Finish destination delegation", + }); + expect(self.serverId).toBe(directory.self.serverId); + expect(remote.serverId).toBe(serverId); + const bApi = await api(b); + expect( + (await bApi.paseo.agents.ref(String(remote.agentId)).refresh())?.agent.labels, + ).toMatchObject({ + "paseo.parent-agent-id": parent.id, + "paseo.parent-server-id": directory.self.serverId, + }); + await expect + .poll(() => parentPrompts.join("\n"), { timeout: 10000 }) + .toContain(`Agent ${self.agentId}`); + await expect + .poll(() => parentPrompts.join("\n"), { timeout: 10000 }) + .toContain(`Agent ${remote.agentId}`); + expect(parentPrompts.join("\n")).toContain(`Server: ${serverId}`); + expect(parentPrompts.join("\n")).toContain(`Server: ${directory.self.serverId}`); + + await aClient.createAgent({ + agentId: String(remote.agentId), + provider: "claude", + model: "claude-test", + cwd: a.staticDir, + workspaceId: workspace.id, + title: "Unrelated equal ID on A", + }); + expect(await call("get_agent_status", { serverId, agentId: remote.agentId })).toMatchObject({ + serverId, + snapshot: { id: remote.agentId, title: "B child", cwd: b.staticDir }, + }); + expect(await call("get_agent_status", { agentId: remote.agentId })).toMatchObject({ + serverId: directory.self.serverId, + snapshot: { title: "Unrelated equal ID on A" }, + }); + const beforeFollowup = parentPrompts.filter((prompt) => + prompt.includes(`Agent ${remote.agentId}`), ).length; + expect(beforeFollowup).toBe(1); expect( await call("send_agent_prompt", { - ...selector, - agentId: self.agentId, - prompt: "SELF_FOLLOWUP_MARKER", + serverId, + agentId: remote.agentId, + prompt: "REMOTE_FOLLOWUP_MARKER", + sessionMode: "full-access", }), - ).toMatchObject({ - success: true, - serverId: directory.self.serverId, - disposition: "turn_started", + ).toMatchObject({ success: true, serverId, guidance: expect.stringContaining("notified") }); + await bApi.client.waitForFinish(String(remote.agentId), 10000); + expect(await call("get_agent_activity", { serverId, agentId: remote.agentId })).toMatchObject({ + serverId, + agentId: remote.agentId, + currentModeId: "full-access", + content: expect.stringContaining("REMOTE_FOLLOWUP_MARKER"), }); - expect(await call("get_agent_activity", { ...selector, agentId: self.agentId })).toMatchObject({ - serverId: directory.self.serverId, - content: expect.stringContaining("SELF_FOLLOWUP_MARKER"), + expect(await call("list_agents", { serverId })).toMatchObject({ + serverId, + agents: expect.arrayContaining([ + expect.objectContaining({ id: remote.agentId, title: "B child", cwd: b.staticDir }), + ]), }); - expect(await call("list_agents", selector)).toMatchObject({ + expect(await call("list_agents", {})).toMatchObject({ serverId: directory.self.serverId, agents: expect.arrayContaining([ - expect.objectContaining({ id: self.agentId, title: "Self child" }), + expect.objectContaining({ + id: remote.agentId, + title: "Unrelated equal ID on A", + cwd: a.staticDir, + }), ]), }); await expect - .poll(() => parentPrompts.filter((prompt) => prompt.includes(`Agent ${self.agentId}`)).length) - .toBe(previous + 1); - } -}); + .poll( + () => parentPrompts.filter((prompt) => prompt.includes(`Agent ${remote.agentId}`)).length, + { timeout: 10000 }, + ) + .toBe(2); + expect( + parentPrompts.findLast((prompt) => prompt.includes(`Agent ${remote.agentId}`)), + ).toContain(`Server: ${serverId}`); + expect((await aClient.fetchAgent(String(remote.agentId)))?.agent.lastUserMessageAt).toBeNull(); + expect(JSON.stringify(await aClient.fetchAgentTimeline(String(remote.agentId)))).not.toContain( + "REMOTE_FOLLOWUP_MARKER", + ); + + for (const selector of [{}, { serverId: directory.self.serverId }]) { + expect(await call("get_agent_status", { ...selector, agentId: self.agentId })).toMatchObject({ + serverId: directory.self.serverId, + snapshot: { title: "Self child" }, + }); + const previous = parentPrompts.filter((prompt) => + prompt.includes(`Agent ${self.agentId}`), + ).length; + expect( + await call("send_agent_prompt", { + ...selector, + agentId: self.agentId, + prompt: "SELF_FOLLOWUP_MARKER", + }), + ).toMatchObject({ + success: true, + serverId: directory.self.serverId, + deliveryStatus: "queued", + }); + await aClient.waitForFinish(String(self.agentId), 10000); + expect( + await call("get_agent_activity", { ...selector, agentId: self.agentId }), + ).toMatchObject({ + serverId: directory.self.serverId, + content: expect.stringContaining("SELF_FOLLOWUP_MARKER"), + }); + expect(await call("list_agents", selector)).toMatchObject({ + serverId: directory.self.serverId, + agents: expect.arrayContaining([ + expect.objectContaining({ id: self.agentId, title: "Self child" }), + ]), + }); + await expect + .poll( + () => parentPrompts.filter((prompt) => prompt.includes(`Agent ${self.agentId}`)).length, + { timeout: 10000 }, + ) + .toBe(previous + 1); + } + }, +); + +test( + "remote agent reports, local CLI sends and automatic completion share the recipient batch", + { timeout: 30000 }, + async () => { + const parentPrompts: string[] = []; + const childStarted = Promise.withResolvers(); + const finishChild = Promise.withResolvers(); + cleanup.push(async () => { + finishChild.resolve(); + }); + const a = await daemon({ + agentClients: createTestAgentClients({ + onStartTurn: (prompt, config) => { + if (config.title === "Batch parent") + parentPrompts.push(typeof prompt === "string" ? prompt : JSON.stringify(prompt)); + }, + }), + }); + const b = await daemon({ + agentClients: createTestAgentClients({ + onStartTurn: () => childStarted.resolve(), + beforeTurnComplete: () => finishChild.promise, + }), + }); + const origin = await api(a); + const destination = await api(b); + await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` }); + await destination.paseo.hosts.add({ connection: `127.0.0.1:${a.port}` }); + const aDirectory = await online(origin.paseo); + const bDirectory = await online(destination.paseo); + const aId = aDirectory.self.serverId; + const bId = bDirectory.self.serverId; + const aWorkspace = await origin.paseo.workspaces.create({ + source: { kind: "directory", path: a.staticDir }, + }); + const bWorkspace = await destination.paseo.workspaces.create({ + source: { kind: "directory", path: b.staticDir }, + }); + const parent = await aWorkspace.agents.create({ + config: { provider: "claude/claude-test", modeId: "full-access" }, + title: "Batch parent", + }); + const child = await bWorkspace.agents.create({ + config: { provider: "claude/claude-test", modeId: "full-access" }, + title: "Batch child", + labels: { "paseo.parent-agent-id": parent.id, "paseo.parent-server-id": aId }, + }); + const caller = async (host: TestPaseoDaemon, agentId: string) => { + const mcp = new Client({ name: "batching-agent", version: "1" }); + await mcp.connect( + new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${host.port}/mcp/agents?callerAgentId=${agentId}`), + ), + ); + cleanup.push(() => mcp.close()); + return mcp; + }; + const parentMcp = await caller(a, parent.id); + const childMcp = await caller(b, child.id); + const task = await parentMcp.callTool({ + name: "send_agent_prompt", + arguments: { serverId: bId, agentId: child.id, prompt: "Do the delegated task" }, + }); + expect(task.isError).not.toBe(true); + expect(task.content).toEqual([ + expect.objectContaining({ + type: "text", + text: expect.stringContaining('"deliveryStatus": "queued"'), + }), + ]); + expect(task.structuredContent).toMatchObject({ + success: true, + deliveryStatus: "queued", + serverId: bId, + }); + // An idle recipient must not report old idle as completion of its queued task. + expect((await destination.client.waitForFinish(child.id, 50)).status).toBe("timeout"); + await childStarted.promise; + const report = await childMcp.callTool({ + name: "send_agent_prompt", + arguments: { + serverId: aId, + agentId: parent.id, + prompt: "REMOTE_REPORT ", + notifyOnFinish: false, + }, + }); + expect(report.isError).not.toBe(true); + expect(report.structuredContent).toMatchObject({ deliveryStatus: "queued", serverId: aId }); + expect(parentPrompts).toEqual([]); + + const cli = await promisify(execFile)( + process.execPath, + [ + "--conditions=source", + "--import", + "tsx", + fileURLToPath(new URL("../../../../cli/src/index.ts", import.meta.url)), + "--host", + `127.0.0.1:${a.port}`, + "send", + parent.id, + "LOCAL_CLI_REPORT", + "--no-wait", + "--json", + ], + { + env: { ...process.env, PASEO_AGENT_ID: parent.id }, + timeout: 15000, + }, + ); + expect(JSON.parse(cli.stdout)).toMatchObject({ status: "queued" }); + expect(parentPrompts).toEqual([]); + const receiptStates = async () => { + const directory = join(a.paseoHome, "agent-requests"); + const receipts = await Promise.all( + (await readdir(directory)) + .filter((file) => file.endsWith(".json")) + .map(async (file) => JSON.parse(await readFile(join(directory, file), "utf8"))), + ); + return receipts + .filter( + (receipt) => receipt.agentId === parent.id && typeof receipt.fingerprint === "string", + ) + .map((receipt) => receipt.state) + .sort(); + }; + expect(await receiptStates()).toEqual(["pending", "pending"]); + finishChild.resolve(); + await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(1); + expect(parentPrompts[0]).toContain("REMOTE_REPORT <ready>"); + expect(parentPrompts[0]).toContain("LOCAL_CLI_REPORT"); + expect(parentPrompts[0]).toContain('event="finished"'); + expect(parentPrompts[0]).toContain(`source-agent-id="${child.id}"`); + await expect.poll(receiptStates, { timeout: 10000 }).toEqual(["completed", "completed"]); + const timeline = await origin.client.fetchAgentTimeline(parent.id); + const messages = timeline.entries.flatMap((row) => + row.item.type === "tool_call" && row.item.agentMessage ? [row.item.agentMessage] : [], + ); + expect( + messages.map((message) => [message.event, message.sender.id, message.sender.serverId]), + ).toEqual([ + ["message", child.id, bId], + ["message", parent.id, undefined], + ["finished", child.id, bId], + ]); + expect(messages[0]!.text).toBe("REMOTE_REPORT "); + expect(timeline.entries.filter((row) => row.item.type === "user_message")).toEqual([]); + + const human = await aWorkspace.agents.create({ + config: { provider: "claude/claude-test", modeId: "full-access" }, + title: "Human recipient", + }); + expect(await origin.client.sendAgentMessage(human.id, "HUMAN_PROMPT")).toEqual({}); + expect( + (await origin.client.fetchAgentTimeline(human.id)).entries.some( + (row) => row.item.type === "user_message" && row.item.text === "HUMAN_PROMPT", + ), + ).toBe(true); + }, +); + +test( + "remote top-level sends set the mode through the existing RPC before sending", + { timeout: 15000 }, + async () => { + const a = await daemon(); + const b = await daemon(); + const wire = await hostWire(b); + const origin = await api(a); + const destination = await api(b); + await origin.paseo.hosts.add({ connection: wire.connection }); + const serverId = (await online(origin.paseo)).hosts[0]!.serverId!; + const workspace = await destination.paseo.workspaces.create({ + source: { kind: "directory", path: b.staticDir }, + }); + const agent = await workspace.agents.create({ + config: { provider: "claude/claude-test", modeId: "default" }, + title: "Human mode recipient", + }); + const mcp = new Client({ name: "human-mode-send", version: "1" }); + cleanup.push(() => mcp.close()); + await mcp.connect( + new StreamableHTTPClientTransport(new URL(`http://127.0.0.1:${a.port}/mcp/agents`)), + ); + const response = await mcp.callTool({ + name: "send_agent_prompt", + arguments: { serverId, agentId: agent.id, prompt: "Human task", sessionMode: "full-access" }, + }); + expect(response.isError).not.toBe(true); + expect((await destination.client.fetchAgent(agent.id))?.agent.currentModeId).toBe( + "full-access", + ); + const requests = wire.frames + .filter((frame) => frame.direction === "toB" && !frame.binary) + .flatMap((frame) => { + const outer = JSON.parse(frame.bytes.toString()); + if (outer.type !== "session" || outer.message.type !== "hosts.route.frame.request") + return []; + const inner = JSON.parse(outer.message.data); + return inner.type === "session" ? [inner.message] : []; + }) + .filter( + (message) => + message.type === "set_agent_mode_request" || + message.type === "send_agent_message_request", + ); + expect(requests.map((message) => message.type)).toEqual([ + "set_agent_mode_request", + "send_agent_message_request", + ]); + expect(requests[1]).not.toHaveProperty("sessionMode"); + }, +); diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index b837502f0c..7d881286fb 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -1,3 +1,4 @@ +import { createAgentMessageDeliveryStub } from "./test-utils/session-stubs.js"; import { createMessageReceiptsStub, createTestCreationService, @@ -369,6 +370,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { const sessionOptions: SessionOptions = { messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: options.clientId ?? "test-client", onMessage: (message) => messages.push(message), diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 69f2c53b45..f49191e59e 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1,3 +1,4 @@ +import type { AgentMessageDelivery } from "./agent/agent-messages/index.js"; import { AgentDirectory } from "./agent/directory.js"; import { ConnectedHostSession } from "./connected-hosts/index.js"; import type { ConnectedHosts } from "./connected-hosts/index.js"; @@ -461,6 +462,7 @@ export interface SessionOptions { agentManager: AgentManager; agentStorage: AgentStorage; messageReceipts: Pick; + agentMessages: AgentMessageDelivery; creationService: Pick; projectRegistry: ProjectRegistry; workspaceRegistry: WorkspaceRegistry; @@ -813,6 +815,7 @@ export class Session { private readonly hubExecutionController: HubExecutionController | null; private readonly workspaceScripts: WorkspaceScriptsService; private readonly messageReceipts: Pick; + private readonly agentMessages: AgentMessageDelivery; private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch; private readonly creationService: Pick; @@ -896,6 +899,7 @@ export class Session { this.pushNotifications = pushNotifications; this.paseoHome = paseoHome; this.messageReceipts = options.messageReceipts; + this.agentMessages = options.agentMessages; this.creationService = options.creationService; this.projectIcons = new ProjectIconReader(paseoHome); this.worktreesRoot = worktreesRoot; @@ -8221,17 +8225,42 @@ export class Session { }, "agent.session.send_agent_message", ); + let queued!: () => void; + const admitted = new Promise((acceptAdmission) => { + queued = acceptAdmission; + }); + let deliveryStatus: "queued" | undefined; const send = async () => { + if (msg.sourceAgentId) { + const receipt = this.agentMessages.enqueue({ + recipientId: agentId, + messageId: msg.messageId, + prompt, + source: { + kind: "agent-message", + agentId: msg.sourceAgentId, + serverId: msg.sourceAgentServerId, + title: !msg.sourceAgentServerId + ? (this.agentManager.getAgent(msg.sourceAgentId)?.config.title ?? undefined) + : undefined, + }, + activeTurnBehavior: msg.activeTurnBehavior ?? "steer", + sessionMode: msg.sessionMode, + clearPendingPermissions: true, + }); + deliveryStatus = "queued"; + queued(); + await receipt.dispatched; + return; + } const result = await sendPromptToAgent({ agentManager: this.agentManager, agentStorage: this.agentStorage, agentId, prompt, - source: msg.sourceAgentId - ? { kind: "agent-message", agentId: msg.sourceAgentId } - : undefined, messageId: msg.messageId, activeTurnBehavior: msg.activeTurnBehavior ?? "interrupt", + sessionMode: msg.sessionMode, clearPendingPermissions: true, logger: this.sessionLogger, }); @@ -8243,23 +8272,30 @@ export class Session { ); } }; - if (msg.messageId) { - await this.messageReceipts.send({ - agentId, - messageId: msg.messageId, - request: { - prompt, - sourceAgentId: msg.sourceAgentId, - activeTurnBehavior: msg.activeTurnBehavior ?? "interrupt", - }, - prepare: async () => { - if (!msg.sourceAgentId) await this.prepareAgentMessage(agentId, msg.text); - }, - send, - }); - } else { - await send(); - } + const sent = msg.messageId + ? this.messageReceipts.send({ + agentId, + messageId: msg.messageId, + request: { + prompt, + sourceAgentId: msg.sourceAgentId, + sourceAgentServerId: msg.sourceAgentServerId, + sessionMode: msg.sessionMode, + activeTurnBehavior: + msg.activeTurnBehavior ?? (msg.sourceAgentId ? "steer" : "interrupt"), + }, + prepare: async () => { + if (!msg.sourceAgentId) await this.prepareAgentMessage(agentId, msg.text); + }, + send, + }) + : send(); + void sent.catch((error) => { + if (deliveryStatus === "queued") { + this.handleAgentRunError(agentId, error, "Failed to dispatch queued agent message"); + } + }); + await Promise.race([sent, admitted]); this.emit({ type: "send_agent_message_response", @@ -8267,6 +8303,7 @@ export class Session { requestId: msg.requestId, agentId, accepted: true, + ...(deliveryStatus ? { deliveryStatus } : {}), error: null, }, }); @@ -8308,40 +8345,6 @@ export class Session { } const agentId = resolved.agentId; - const live = this.agentManager.getAgent(agentId); - if (!live) { - const settled = await this.readSettledAgent(agentId); - if (!settled) { - this.emit({ - type: "wait_for_finish_response", - payload: { - requestId, - status: "error", - final: null, - error: `Agent not found: ${agentId}`, - lastMessage: null, - }, - }); - return; - } - const { record, lastMessage } = settled; - const final = this.buildStoredAgentPayload(record); - let status: "permission" | "error" | "idle"; - if (record.attentionReason === "permission") { - status = "permission"; - } else if (record.lastStatus === "error") { - status = "error"; - } else { - status = "idle"; - } - const error = resolveWaitForFinishError({ status, final }); - this.emit({ - type: "wait_for_finish_response", - payload: { requestId, status, final, error, lastMessage }, - }); - return; - } - const abortController = new AbortController(); const hasTimeout = typeof timeoutMs === "number" && timeoutMs > 0; const timeoutHandle = hasTimeout @@ -8351,6 +8354,45 @@ export class Session { : null; try { + await this.agentMessages.waitForDispatch( + agentId, + AbortSignal.any([abortController.signal, sourceSignal]), + ); + + const live = this.agentManager.getAgent(agentId); + if (!live) { + const settled = await this.readSettledAgent(agentId); + if (!settled) { + this.emit({ + type: "wait_for_finish_response", + payload: { + requestId, + status: "error", + final: null, + error: `Agent not found: ${agentId}`, + lastMessage: null, + }, + }); + return; + } + const { record, lastMessage } = settled; + const final = this.buildStoredAgentPayload(record); + let status: "permission" | "error" | "idle"; + if (record.attentionReason === "permission") { + status = "permission"; + } else if (record.lastStatus === "error") { + status = "error"; + } else { + status = "idle"; + } + const error = resolveWaitForFinishError({ status, final }); + this.emit({ + type: "wait_for_finish_response", + payload: { requestId, status, final, error, lastMessage }, + }); + return; + } + let result = await this.agentManager.waitForAgentEvent(agentId, { signal: AbortSignal.any([abortController.signal, sourceSignal]), waitForActive: true, diff --git a/packages/server/src/server/session.workspace-git-watch.test.ts b/packages/server/src/server/session.workspace-git-watch.test.ts index 836703fd30..c836e6db5d 100644 --- a/packages/server/src/server/session.workspace-git-watch.test.ts +++ b/packages/server/src/server/session.workspace-git-watch.test.ts @@ -1,3 +1,4 @@ +import { createAgentMessageDeliveryStub } from "./test-utils/session-stubs.js"; import { createMessageReceiptsStub, createTestCreationService, @@ -189,6 +190,7 @@ function createSessionForWorkspaceGitWatchTests(options?: { const session = new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, diff --git a/packages/server/src/server/session.workspace-resolution-invariants.test.ts b/packages/server/src/server/session.workspace-resolution-invariants.test.ts index e848e46f01..079dadc26f 100644 --- a/packages/server/src/server/session.workspace-resolution-invariants.test.ts +++ b/packages/server/src/server/session.workspace-resolution-invariants.test.ts @@ -1,3 +1,4 @@ +import { createAgentMessageDeliveryStub } from "./test-utils/session-stubs.js"; import { createMessageReceiptsStub, createTestCreationService, @@ -93,6 +94,7 @@ function createHarness(input: { const session = new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test", permissions: OWNER_PERMISSIONS, diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 89bc5a43e3..71759d33bf 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -1,3 +1,4 @@ +import { createAgentMessageDeliveryStub } from "./test-utils/session-stubs.js"; import { createMessageReceiptsStub, createTestCreationService, @@ -647,6 +648,7 @@ function createSessionForWorkspaceTests( const session = asTestSession( new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, @@ -1009,6 +1011,7 @@ test("create_agent_request keeps requested child cwd when grouped under an exist const session = asTestSession( new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test-client", serverId: "test-server", @@ -1164,6 +1167,7 @@ test("create_agent_request launches from an exact subdirectory in a created work const emitted: SessionOutboundMessage[] = []; const session = new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, @@ -1303,6 +1307,7 @@ test("create_agent_request does not title an existing workspace from the agent p const session = asTestSession( new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, @@ -1574,6 +1579,7 @@ test("archive emits an authoritative agent_update upsert for subscribed clients" const session = asTestSession( new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, @@ -2058,6 +2064,7 @@ test("close_items_request archives agents and kills terminals in one batch", asy const session = asTestSession( new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, @@ -2228,6 +2235,7 @@ test("close_items_request archives stored agents that are not currently loaded", const session = asTestSession( new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, @@ -2389,6 +2397,7 @@ test("close_items_request continues after an archive failure", async () => { const session = asTestSession( new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, @@ -3646,6 +3655,7 @@ test("workspace update stream keeps persisted workspace visible after agents sto const session = asTestSession( new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, diff --git a/packages/server/src/server/snapshot-mutation-ownership.test.ts b/packages/server/src/server/snapshot-mutation-ownership.test.ts index 2cfe5d3e18..4af5e45c5b 100644 --- a/packages/server/src/server/snapshot-mutation-ownership.test.ts +++ b/packages/server/src/server/snapshot-mutation-ownership.test.ts @@ -1,3 +1,4 @@ +import { createAgentMessageDeliveryStub } from "./test-utils/session-stubs.js"; import { createMessageReceiptsStub, createTestCreationService, @@ -104,6 +105,7 @@ describe("snapshot mutation ownership boundary", () => { const session = asInternals( new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, diff --git a/packages/server/src/server/test-utils/agent-messages.ts b/packages/server/src/server/test-utils/agent-messages.ts new file mode 100644 index 0000000000..b69da1f1c8 --- /dev/null +++ b/packages/server/src/server/test-utils/agent-messages.ts @@ -0,0 +1,43 @@ +import { createAgentMessageDelivery } from "../agent/agent-messages/index.js"; +import { + sendPreparedPromptToAgent, + waitForAgentRunStartWithTimeout, +} from "../agent/agent-prompt.js"; +import type { AgentManager } from "../agent/agent-manager.js"; +import type { AgentStorage } from "../agent/agent-storage.js"; +import type { Logger } from "pino"; + +/** Notification-format tests advance delivery immediately; debounce tests inject their own clock. */ +export function createTestAgentMessages(params: { + agentManager: AgentManager; + agentStorage: AgentStorage; + logger: Logger; +}) { + return createAgentMessageDelivery({ + schedule: (_delay, callback) => { + const timer = setTimeout(callback, 0); + return () => clearTimeout(timer); + }, + dispatch: async (request) => { + const result = await sendPreparedPromptToAgent({ + ...params, + agentId: request.recipientId, + prompt: request.prompt, + sessionMode: request.sessionMode, + unarchive: request.unarchive, + activeTurnBehavior: request.activeTurnBehavior, + clearPendingPermissions: request.clearPendingPermissions, + onDispatch: request.onDispatch, + runOptions: { clientMessageId: request.messageIds[0] }, + }); + if (result.disposition === "turn_started") + await waitForAgentRunStartWithTimeout(params.agentManager, request.recipientId); + return result.disposition; + }, + onError: (error, request) => + params.logger.error( + { err: error, agentId: request.recipientId, messageIds: request.messageIds }, + "Agent message batch dispatch failed", + ), + }); +} diff --git a/packages/server/src/server/test-utils/session-stubs.ts b/packages/server/src/server/test-utils/session-stubs.ts index c5cf271ad3..922f673935 100644 --- a/packages/server/src/server/test-utils/session-stubs.ts +++ b/packages/server/src/server/test-utils/session-stubs.ts @@ -1,3 +1,4 @@ +import { createAgentMessageDelivery } from "../agent/agent-messages/index.js"; import pino from "pino"; import { randomUUID } from "node:crypto"; import { afterEach, vi } from "vitest"; @@ -320,3 +321,15 @@ export function createTestCreationService(): SessionOptions["creationService"] { creationDirectories.push(directory); return new CreationService(directory, pino({ level: "silent" })); } + +/** Session fixtures without agent-originated sends still own the delivery lifetime. */ +export function createAgentMessageDeliveryStub(): SessionOptions["agentMessages"] { + return createAgentMessageDelivery({ + dispatch: async () => { + throw new Error("Unexpected agent message in session fixture"); + }, + onError: (error) => { + throw error; + }, + }); +} diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 8f36e1a2b8..dff09358a0 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1,3 +1,11 @@ +import { + createAgentMessageDelivery, + type AgentMessageDelivery, +} from "./agent/agent-messages/index.js"; +import { + sendPreparedPromptToAgent, + waitForAgentRunStartWithTimeout, +} from "./agent/agent-prompt.js"; import { createToolHosts } from "./agent/host-operations/index.js"; import { HOST_ROUTE_OPCODE } from "@getpaseo/protocol/binary-frames/index"; import { DaemonLinkEndpoint, VirtualHostSocket } from "./connected-hosts/index.js"; @@ -547,6 +555,7 @@ export class VoiceAssistantWebSocketServer { private readonly agentManager: AgentManager; private readonly agentStorage: AgentStorage; private readonly messageReceipts: MessageReceipts; + readonly agentMessages: AgentMessageDelivery; private readonly creationService: CreationService; private readonly projectRegistry: ProjectRegistry; private readonly workspaceRegistry: WorkspaceRegistry; @@ -693,6 +702,31 @@ export class VoiceAssistantWebSocketServer { this.orchestrationSkills = orchestrationSkills; this.agentManager = agentManager; this.agentStorage = agentStorage; + this.agentMessages = createAgentMessageDelivery({ + dispatch: async (batch) => { + const { disposition } = await sendPreparedPromptToAgent({ + agentManager, + agentStorage, + agentId: batch.recipientId, + prompt: batch.prompt, + activeTurnBehavior: batch.activeTurnBehavior, + sessionMode: batch.sessionMode, + clearPendingPermissions: batch.clearPendingPermissions, + unarchive: batch.unarchive, + runOptions: { clientMessageId: batch.messageIds[0] }, + onDispatch: batch.onDispatch, + logger: this.logger, + }); + if (disposition === "turn_started") + await waitForAgentRunStartWithTimeout(agentManager, batch.recipientId); + return disposition; + }, + onError: (err, batch) => + this.logger.error( + { err, agentId: batch.recipientId, messageIds: batch.messageIds }, + "Agent message batch dispatch failed", + ), + }); this.messageReceipts = new MessageReceipts(join(paseoHome, "agent-requests")); this.creationService = new CreationService( join(paseoHome, "creations"), @@ -1095,6 +1129,7 @@ export class VoiceAssistantWebSocketServer { public async close(): Promise { this.prepareForShutdown(); + await this.agentMessages.close(); await this.connectedHosts.close(); this.unsubscribeSpeechReadiness?.(); this.unsubscribeSpeechReadiness = null; @@ -1561,6 +1596,7 @@ export class VoiceAssistantWebSocketServer { agentManager: this.agentManager, agentStorage: this.agentStorage, messageReceipts: this.messageReceipts, + agentMessages: this.agentMessages, creationService: this.creationService, projectRegistry: this.projectRegistry, workspaceRegistry: this.workspaceRegistry, @@ -2047,6 +2083,7 @@ export class VoiceAssistantWebSocketServer { // COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30. canonicalSubmittedPrompts: true, agentMessageProvenance: true, + agentMessageBatching: true, // COMPAT(stableProjectIdentity): added in v0.1.109, remove gate after 2027-01-15. stableProjectIdentity: true, // COMPAT(workspaceScriptManagement): added in v0.1.105, remove gate after 2027-01-10. diff --git a/packages/server/src/server/wire-compat.test.ts b/packages/server/src/server/wire-compat.test.ts index a7b36f521d..c6896c3fb0 100644 --- a/packages/server/src/server/wire-compat.test.ts +++ b/packages/server/src/server/wire-compat.test.ts @@ -1,3 +1,4 @@ +import { createAgentMessageDeliveryStub } from "./test-utils/session-stubs.js"; import { createMessageReceiptsStub, createTestCreationService, @@ -214,6 +215,7 @@ function createSessionForWireCompatTest(options?: { const session = new Session({ messageReceipts: createMessageReceiptsStub(), + agentMessages: createAgentMessageDeliveryStub(), creationService: createTestCreationService(), clientId: "wire-compat-client", permissions: OWNER_PERMISSIONS, diff --git a/skills/paseo/SKILL.md b/skills/paseo/SKILL.md index 83d329f948..e92bdf91ee 100644 --- a/skills/paseo/SKILL.md +++ b/skills/paseo/SKILL.md @@ -61,9 +61,9 @@ Agent-scoped creation always creates your subagent. Omit `workspaceId` to use yo Detach is an explicit user action in the subagents track, not an agent tool. A cross-workspace child remains your subagent even though it also appears as a normal tab in its workspace. -Agent-scoped `create_agent` defaults `notifyOnFinish` to true. Set it to `false` only for truly fire-and-forget agents. +Agent-scoped `create_agent` defaults `notifyOnFinish` to true. Set it to `false` when you want explicit reporting or no completion callback. -**`send_agent_prompt`** — `{ agentId, prompt, activeTurnBehavior? }`. Use for follow-ups to an existing agent. Agent-scoped calls default to `steer`; top-level calls default to `interrupt`. Set `activeTurnBehavior` to `steer` or `interrupt` to override the default. Steering falls back to interruption when unavailable. CLI agent sends use the same defaults and accept `--behavior steer|interrupt`. Returns after dispatch acknowledgement. Agent-scoped calls default `notifyOnFinish` to true; top-level calls have no parent callback. +**`send_agent_prompt`** — `{ agentId, prompt, activeTurnBehavior? }`. Use for follow-ups to an existing agent. Agent-scoped calls default to `steer`; top-level calls default to `interrupt`. Set `activeTurnBehavior` to `steer` or `interrupt` to override the default. Steering falls back to interruption when unavailable. CLI agent sends use the same defaults and accept `--behavior steer|interrupt`. Agent-scoped calls return a queued receipt immediately; messages to each recipient deliver after two quiet seconds. Top-level calls return after dispatch acknowledgement. Agent-scoped calls default `notifyOnFinish` to true; top-level calls have no parent callback. **`update_agent`** — `{ agentId, name?, labels?, settings? }`. Use `settings` for runtime changes on an existing agent: `modeId`, `model`, `thinkingOptionId`, and provider-specific `features`. For Codex fast mode, pass `settings: { features: { "fast_mode": true } }`. @@ -108,9 +108,9 @@ Schedules have the full list/inspect/update/pause/resume/run-once/log/delete sur Agents take time — 10–30+ minutes is routine. Favor asynchronous workflows. -`create_agent` and `send_agent_prompt` return after startup or dispatch acknowledgement in both MCP scopes. Top-level callers have no parent callback; observe completion through status/activity tools or SDK/RPC subscriptions and wait APIs. +`create_agent` returns after startup. Agent-scoped `send_agent_prompt` returns after queue admission; provider dispatch follows after two quiet seconds. Top-level sends return after dispatch acknowledgement. Top-level callers have no parent callback; observe completion through status/activity tools or SDK/RPC subscriptions and wait APIs. -For agent-scoped `create_agent` and `send_agent_prompt`, leave `notifyOnFinish` omitted or set it to `true` unless the work is truly fire-and-forget. You will get notified when the target agent finishes, errors, or needs permission. Move on to other work. The notification arrives on its own. +For agent-scoped `create_agent` and `send_agent_prompt`, `notifyOnFinish` defaults to `true`. The caller receives the target’s completion, error, or permission report automatically, including its last response. Finish your turn with the result for automatic reporting. For a workflow that reports explicitly with `send_agent_prompt`, set `notifyOnFinish: false` when assigning the task. Paseo supplies caller identity for agent-scoped tools and CLI agent context; use that supplied identity. With finish notifications enabled, don't poll `list_agents` or `get_agent_status` to "check on" a running agent. The notification will tell you.