mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-27 19:52:49 +00:00
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.
This commit is contained in:
Mohamed Boudra
@@ -73,7 +73,12 @@ Runtime ownership is resolved from explicit workspace ID and caller context, nev
|
||||
|
||||
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.
|
||||
|
||||
@@ -54,7 +54,7 @@ export interface ComposerSendClient {
|
||||
images: Array<{ data: string; mimeType: string }>;
|
||||
attachments: ReturnType<typeof splitComposerAttachmentsForSubmit>["attachments"];
|
||||
},
|
||||
) => Promise<void>;
|
||||
) => Promise<unknown>;
|
||||
uploadFile: (input: { fileName: string; mimeType: string; bytes: Uint8Array }) => Promise<{
|
||||
requestId: string;
|
||||
file: {
|
||||
|
||||
@@ -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<void> {
|
||||
|
||||
@@ -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,
|
||||
};
|
||||
|
||||
@@ -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;
|
||||
@@ -3688,26 +3690,34 @@ export class DaemonClient {
|
||||
async sendAgentMessage(
|
||||
agentId: string,
|
||||
text: string,
|
||||
options?: SendMessageOptions,
|
||||
): Promise<void> {
|
||||
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,
|
||||
@@ -3726,10 +3736,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<void> {
|
||||
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(
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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(),
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -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<typeof formatAgentMessage>[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" },
|
||||
|
||||
@@ -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 <ready>",
|
||||
},
|
||||
{
|
||||
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 <ready>",
|
||||
},
|
||||
},
|
||||
{
|
||||
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 });
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
@@ -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<void> {
|
||||
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(
|
||||
|
||||
@@ -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 <ready>",
|
||||
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 <ready>",
|
||||
},
|
||||
{
|
||||
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: '<paseo-system version="1" kind="agent-message" source-agent-id="one" message-id="first">\nFirst\n</paseo-system>',
|
||||
},
|
||||
firstImage,
|
||||
{
|
||||
type: "text",
|
||||
text: '<paseo-system version="1" kind="agent-message" source-agent-id="two" message-id="second">\nSecond\n\nGitHub Issue #42: Context\nhttps://example.com/42\n\nAttached\n</paseo-system>',
|
||||
},
|
||||
secondImage,
|
||||
]);
|
||||
expect(
|
||||
parseAgentMessages(
|
||||
(prompt as Exclude<AgentPromptInput, string>)
|
||||
.flatMap((block) => (block.type === "text" ? [block.text] : []))
|
||||
.join("\n"),
|
||||
)?.map((message) => message.source?.agentId),
|
||||
).toEqual(["one", "two"]);
|
||||
});
|
||||
@@ -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<PromptDispatchDisposition>;
|
||||
}
|
||||
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<PromptDispatchDisposition>;
|
||||
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<void>;
|
||||
/** Existing completion waits include agent messages admitted before the wait. */
|
||||
waitForDispatch(recipientId: string, signal: AbortSignal): Promise<void>;
|
||||
}
|
||||
|
||||
/** 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<string, { messages: Admitted[]; cancel(): void }>();
|
||||
const dispatching = new Map<string, Promise<void>>();
|
||||
let closed = false;
|
||||
let closing: Promise<void> | 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<PromptDispatchDisposition>((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<never>((_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;
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -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<typeof sourceSchema>;
|
||||
|
||||
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 `<paseo-system version="1" ${attributes} message-id="${escapeXml(message.id)}"${server}${title}${event}>\n${escapeXml(message.text)}\n</paseo-system>`;
|
||||
}
|
||||
|
||||
export function parseAgentMessage(text: string): AgentMessage | null {
|
||||
const envelope = /^<paseo-system((?:\s+[a-z-]+="[^"<>]*")+)>\n([\s\S]*)\n<\/paseo-system>$/.exec(
|
||||
text,
|
||||
);
|
||||
if (!envelope) return null;
|
||||
const attributes = new Map<string, string>();
|
||||
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+(?=<paseo-system )/);
|
||||
const messages = parts.map(parseAgentMessage);
|
||||
return messages.every((message) => 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 <paseo-system>…</paseo-system> 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 `<paseo-system>\n${reason}\n</paseo-system>`;
|
||||
}
|
||||
|
||||
const SYSTEM_ENVELOPE_PATTERN = /^<paseo-system>\n[\s\S]*\n<\/paseo-system>$/;
|
||||
|
||||
export function isSystemInjectedEnvelope(text: string): boolean {
|
||||
return SYSTEM_ENVELOPE_PATTERN.test(text);
|
||||
}
|
||||
@@ -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<typeof sourceSchema>;
|
||||
|
||||
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 `<paseo-system version="1" ${attributes} message-id="${escapeXml(message.id)}"${title}${event}>\n${escapeXml(message.text)}\n</paseo-system>`;
|
||||
}
|
||||
|
||||
export function parseAgentMessage(text: string): AgentMessage | null {
|
||||
const envelope = /^<paseo-system((?:\s+[a-z-]+="[^"<>]*")+)>\n([\s\S]*)\n<\/paseo-system>$/.exec(
|
||||
text,
|
||||
);
|
||||
if (!envelope) return null;
|
||||
const attributes = new Map<string, string>();
|
||||
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 <paseo-system>…</paseo-system> 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 `<paseo-system>\n${reason}\n</paseo-system>`;
|
||||
}
|
||||
|
||||
const SYSTEM_ENVELOPE_PATTERN = /^<paseo-system>\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";
|
||||
|
||||
@@ -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(
|
||||
/<permission-request>\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(
|
||||
/<permission-request>\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 <ready>",
|
||||
},
|
||||
{
|
||||
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,
|
||||
|
||||
@@ -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<SendPromptToAgentParams, "source" | "messageId">,
|
||||
): 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,
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -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";
|
||||
@@ -37,6 +37,7 @@ export interface CreateAgentSessionWorktreeResult {
|
||||
}
|
||||
|
||||
export interface CreateAgentCommandDependencies {
|
||||
agentMessages?: AgentMessageDelivery;
|
||||
agentManager: AgentManager;
|
||||
agentStorage: AgentStorage;
|
||||
logger: Logger;
|
||||
@@ -202,6 +203,7 @@ export async function createAgentCommand(
|
||||
input.callerAgentId &&
|
||||
(resolved.prompt !== undefined || resolved.setupContinuation !== undefined)
|
||||
? setupFinishNotification({
|
||||
agentMessages: requireAgentMessages(dependencies),
|
||||
agentManager: dependencies.agentManager,
|
||||
agentStorage: dependencies.agentStorage,
|
||||
childAgentId: snapshot.id,
|
||||
@@ -634,3 +636,8 @@ async function createMcpWorktree(
|
||||
throw toWorktreeRequestError(error);
|
||||
}
|
||||
}
|
||||
|
||||
function requireAgentMessages(dependencies: CreateAgentCommandDependencies): AgentMessageDelivery {
|
||||
if (!dependencies.agentMessages) throw new Error("Agent message delivery is not configured");
|
||||
return dependencies.agentMessages;
|
||||
}
|
||||
|
||||
@@ -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.",
|
||||
});
|
||||
},
|
||||
|
||||
@@ -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";
|
||||
@@ -124,6 +126,9 @@ interface RegisteredMcpToolWithHandler extends RegisteredMcpTool {
|
||||
}>;
|
||||
}
|
||||
|
||||
const serverDeliveries = new WeakMap<object, AgentMessageDelivery>();
|
||||
const toolDeliveries = new WeakMap<object, AgentMessageDelivery>();
|
||||
|
||||
function lookupTool(
|
||||
server: Awaited<ReturnType<typeof createAgentMcpServer>>,
|
||||
name: string,
|
||||
@@ -144,7 +149,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(
|
||||
@@ -153,7 +161,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<string, unknown>,
|
||||
) {
|
||||
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: {
|
||||
@@ -840,11 +866,17 @@ function createPaseoWorktreeForMcpTest(options: {
|
||||
}
|
||||
|
||||
// Typed in-memory workspace port for catalogue tests without workspace setup.
|
||||
function createAgentMcpServer(options: Parameters<typeof createAgentMcpServerWithDependencies>[0]) {
|
||||
return createAgentMcpServerWithDependencies({
|
||||
async function createAgentMcpServer(
|
||||
options: Parameters<typeof createAgentMcpServerWithDependencies>[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", () => {
|
||||
@@ -4311,7 +4343,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);
|
||||
@@ -4358,7 +4392,7 @@ describe("send_agent_prompt MCP tool", () => {
|
||||
});
|
||||
mcpClient = await connectInMemoryMcpClient(server);
|
||||
const send = (input: Record<string, unknown>) =>
|
||||
mcpClient!.callTool({ name: "send_agent_prompt", arguments: input });
|
||||
sendMcpAndAwaitDispatch(mcpClient!, server, input);
|
||||
await send({
|
||||
agentId: child.id,
|
||||
prompt: "Original work",
|
||||
@@ -4370,10 +4404,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));
|
||||
@@ -4387,7 +4418,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));
|
||||
@@ -4437,7 +4468,7 @@ describe("send_agent_prompt MCP tool", () => {
|
||||
});
|
||||
mcpClient = await connectInMemoryMcpClient(server);
|
||||
const send = (input: Record<string, unknown>) =>
|
||||
mcpClient!.callTool({ name: "send_agent_prompt", arguments: input });
|
||||
sendMcpAndAwaitDispatch(mcpClient!, server, input);
|
||||
await send({
|
||||
agentId: child.id,
|
||||
prompt: "Original work",
|
||||
@@ -4448,10 +4479,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);
|
||||
@@ -4509,7 +4537,7 @@ describe("send_agent_prompt MCP tool", () => {
|
||||
});
|
||||
mcpClient = await connectInMemoryMcpClient(server);
|
||||
const send = (input: Record<string, unknown>) =>
|
||||
mcpClient!.callTool({ name: "send_agent_prompt", arguments: input });
|
||||
sendMcpAndAwaitDispatch(mcpClient!, server, input);
|
||||
await send({
|
||||
agentId: child.id,
|
||||
prompt: "Original work",
|
||||
@@ -4525,6 +4553,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,
|
||||
@@ -4601,11 +4630,14 @@ describe("send_agent_prompt MCP tool", () => {
|
||||
|
||||
const response = await tool.handler(parsed.data as Record<string, unknown>);
|
||||
|
||||
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 () => {
|
||||
@@ -4749,10 +4781,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");
|
||||
}
|
||||
@@ -4815,8 +4847,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({
|
||||
@@ -4887,9 +4918,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");
|
||||
|
||||
@@ -4947,7 +4977,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));
|
||||
@@ -4972,7 +5002,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);
|
||||
@@ -5018,7 +5048,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();
|
||||
@@ -5088,7 +5118,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(() => {
|
||||
@@ -5189,7 +5219,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(() => {
|
||||
@@ -5257,8 +5287,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([
|
||||
|
||||
@@ -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<string, HostOperation>();
|
||||
const tools = new Map<string, PaseoToolDefinition>();
|
||||
const hostTools = new Map([
|
||||
@@ -758,6 +764,7 @@ function createToolCatalog(
|
||||
}
|
||||
if (childAgentId && callerAgentId && input.notifyOnFinish !== false) {
|
||||
setupFinishWait({
|
||||
agentMessages: agentMessages(),
|
||||
agentManager,
|
||||
agentStorage,
|
||||
originServerId: options.serverId,
|
||||
@@ -1260,7 +1267,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 = {
|
||||
@@ -1347,7 +1354,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 = {
|
||||
@@ -2095,11 +2102,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(),
|
||||
},
|
||||
@@ -2117,8 +2126,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,
|
||||
@@ -2140,6 +2151,7 @@ function createToolCatalog(
|
||||
if (!callerAgentId || !notifyOnFinish) return;
|
||||
stopNotification?.();
|
||||
stopNotification = setupFinishNotification({
|
||||
agentMessages: agentMessages(),
|
||||
agentManager,
|
||||
agentStorage,
|
||||
childAgentId: agentId,
|
||||
@@ -2150,6 +2162,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,
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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<string, unknown>) => {
|
||||
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<string, unknown>) => {
|
||||
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<void>();
|
||||
const finishChild = Promise.withResolvers<void>();
|
||||
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 <ready>",
|
||||
notifyOnFinish: false,
|
||||
},
|
||||
});
|
||||
expect(report.isError).not.toBe(true);
|
||||
expect(report.structuredContent).toMatchObject({ deliveryStatus: "queued", serverId: aId });
|
||||
expect(parentPrompts).toEqual([]);
|
||||
|
||||
const { resolve } = await import("node:path");
|
||||
const cli = await promisify(execFile)(
|
||||
process.execPath,
|
||||
[
|
||||
"--import",
|
||||
"tsx",
|
||||
resolve("packages/cli/src/index.ts"),
|
||||
"--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).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 <ready>");
|
||||
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");
|
||||
},
|
||||
);
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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";
|
||||
@@ -469,6 +470,7 @@ export interface SessionOptions {
|
||||
agentManager: AgentManager;
|
||||
agentStorage: AgentStorage;
|
||||
messageReceipts: Pick<MessageReceipts, "send">;
|
||||
agentMessages: AgentMessageDelivery;
|
||||
creationService: Pick<CreationService, "create" | "subscribe">;
|
||||
projectRegistry: ProjectRegistry;
|
||||
workspaceRegistry: WorkspaceRegistry;
|
||||
@@ -821,6 +823,7 @@ export class Session {
|
||||
private readonly hubExecutionController: HubExecutionController | null;
|
||||
private readonly workspaceScripts: WorkspaceScriptsService;
|
||||
private readonly messageReceipts: Pick<MessageReceipts, "send">;
|
||||
private readonly agentMessages: AgentMessageDelivery;
|
||||
private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch;
|
||||
private readonly creationService: Pick<CreationService, "create" | "subscribe">;
|
||||
|
||||
@@ -904,6 +907,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;
|
||||
@@ -8295,17 +8299,42 @@ export class Session {
|
||||
},
|
||||
"agent.session.send_agent_message",
|
||||
);
|
||||
let queued!: () => void;
|
||||
const admitted = new Promise<void>((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,
|
||||
});
|
||||
@@ -8317,23 +8346,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",
|
||||
@@ -8341,6 +8377,7 @@ export class Session {
|
||||
requestId: msg.requestId,
|
||||
agentId,
|
||||
accepted: true,
|
||||
...(deliveryStatus ? { deliveryStatus } : {}),
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
@@ -8382,40 +8419,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
|
||||
@@ -8425,6 +8428,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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<SessionInternals>(
|
||||
new Session({
|
||||
messageReceipts: createMessageReceiptsStub(),
|
||||
agentMessages: createAgentMessageDeliveryStub(),
|
||||
creationService: createTestCreationService(),
|
||||
clientId: "test-client",
|
||||
permissions: OWNER_PERMISSIONS,
|
||||
|
||||
@@ -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",
|
||||
),
|
||||
});
|
||||
}
|
||||
@@ -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;
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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<void> {
|
||||
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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user