From c94c3910b9459c26b978aa317d3007241298687b Mon Sep 17 00:00:00 2001 From: Seth Date: Mon, 15 Jun 2026 13:30:20 -0700 Subject: [PATCH 1/3] feat(coding-agent): add daemon session messaging --- .../coding-agent/src/cli/daemon-command.ts | 92 ++++++++++++++++++ .../src/modes/daemon/agent-session-bus.ts | 97 +++++++++++++++++++ .../src/modes/daemon/daemon-mode.ts | 81 ++++++++++++++++ .../src/modes/daemon/daemon-protocol.ts | 11 +++ .../test/agent-session-bus.test.ts | 84 ++++++++++++++++ .../coding-agent/test/daemon-command.test.ts | 29 ++++++ 6 files changed, 394 insertions(+) create mode 100644 packages/coding-agent/src/modes/daemon/agent-session-bus.ts create mode 100644 packages/coding-agent/test/agent-session-bus.test.ts diff --git a/packages/coding-agent/src/cli/daemon-command.ts b/packages/coding-agent/src/cli/daemon-command.ts index b01efa8fc3..37b1b5ef04 100644 --- a/packages/coding-agent/src/cli/daemon-command.ts +++ b/packages/coding-agent/src/cli/daemon-command.ts @@ -32,6 +32,7 @@ const DAEMON_CLIENT_COMMANDS = new Set([ "kill", "rename", "prompt", + "send", "steer", "follow-up", "state", @@ -186,6 +187,9 @@ async function runDaemonClientCommand(parsed: ParsedDaemonClientCommand): Promis case "prompt": await runPrompt(client, parsed.positionals); return; + case "send": + await runSend(client, parsed.positionals, parsed.json); + return; case "steer": await runMessageCommand(client, "steer", parsed.positionals, parsed.json); return; @@ -762,6 +766,79 @@ async function runPrompt(client: DaemonClient, args: string[]): Promise { } } +async function runSend(client: DaemonClient, args: string[], json: boolean): Promise { + const parsed = parseSendArgs(args); + const response = await client.request({ + type: "send_message", + targetActiveSessionId: parsed.targetActiveSessionId, + fromActiveSessionId: parsed.fromActiveSessionId, + deliveryMode: parsed.deliveryMode, + message: parsed.message, + }); + const data = requireSuccess(response); + if (json) { + printJson(data); + return; + } + if (isAgentMessageReceipt(data)) { + const target = data.target.sessionName ?? data.target.activeSessionId; + console.log(`Sent to ${target}`); + return; + } + console.log("ok"); +} + +interface ParsedSendArgs { + targetActiveSessionId: string; + fromActiveSessionId?: string; + deliveryMode?: "auto" | "steer" | "follow_up"; + message: string; +} + +function parseSendArgs(args: string[]): ParsedSendArgs { + let fromActiveSessionId: string | undefined; + let deliveryMode: "auto" | "steer" | "follow_up" | undefined; + const positionals: string[] = []; + + for (let index = 0; index < args.length; index++) { + const arg = args[index]; + if (arg === "--from") { + const value = args[index + 1]; + if (!value) { + throw new Error("--from requires a session id or name"); + } + fromActiveSessionId = value; + index++; + continue; + } + if (arg === "--steer") { + deliveryMode = "steer"; + continue; + } + if (arg === "--follow-up") { + deliveryMode = "follow_up"; + continue; + } + if (arg === "--auto") { + deliveryMode = "auto"; + continue; + } + positionals.push(arg); + } + + const targetActiveSessionId = positionals[0]; + const message = positionals.slice(1).join(" ").trim(); + if (!targetActiveSessionId || !message) { + throw new Error("Usage: daemon send [--from ] [--steer|--follow-up] "); + } + return { + targetActiveSessionId, + fromActiveSessionId, + deliveryMode, + message, + }; +} + async function runMessageCommand( client: DaemonClient, type: "steer" | "follow_up", @@ -1314,6 +1391,18 @@ function isLiveSessionSummary(value: unknown): value is SessionSummary & { activ return isSessionSummary(value) && typeof value.activeSessionId === "string"; } +function isAgentMessageReceipt(value: unknown): value is { target: { activeSessionId: string; sessionName?: string } } { + if (!value || typeof value !== "object") { + return false; + } + const target = (value as { target?: unknown }).target; + return ( + !!target && + typeof target === "object" && + typeof (target as { activeSessionId?: unknown }).activeSessionId === "string" + ); +} + function printDaemonHelp(): void { console.log(`${chalk.bold("Usage:")} ${APP_NAME} daemon [options] [session name] @@ -1329,6 +1418,7 @@ ${chalk.bold("Commands:")} attach Attach an interactive terminal to a live session detach [session] Detach this client from one session or all sessions prompt Send a prompt, stream events, and exit when idle + send [options] Send an agent-to-agent message to another live session steer Queue a steering message follow-up Queue a follow-up message rename Rename a live session @@ -1345,6 +1435,7 @@ ${chalk.bold("Options:")} --cwd Working directory for the created session --foreground, --no-detach Keep daemon attached to this terminal for debugging --json Print raw JSON for commands with formatted output; attach streams raw protocol JSON + send options: --from , --steer, --follow-up Agent options such as --model, --provider, --tools, and --thinking apply to created sessions. ${chalk.bold("Examples:")} @@ -1359,6 +1450,7 @@ ${chalk.bold("Examples:")} ${APP_NAME} daemon --socket /tmp/prime-agent.sock list -a ${APP_NAME} daemon --socket /tmp/prime-agent.sock create scratch ${APP_NAME} daemon --socket /tmp/prime-agent.sock prompt "Say hello" + ${APP_NAME} daemon --socket /tmp/prime-agent.sock send --from planner worker "Use this context..." ${APP_NAME} daemon --socket /tmp/prime-agent.sock attach ${APP_NAME} daemon --socket /tmp/prime-agent.sock shutdown `); diff --git a/packages/coding-agent/src/modes/daemon/agent-session-bus.ts b/packages/coding-agent/src/modes/daemon/agent-session-bus.ts new file mode 100644 index 0000000000..846d2135fb --- /dev/null +++ b/packages/coding-agent/src/modes/daemon/agent-session-bus.ts @@ -0,0 +1,97 @@ +export type AgentSessionMessageDeliveryMode = "auto" | "steer" | "follow_up"; + +export interface AgentSessionMessageEndpoint { + activeSessionId: string; + sessionId: string; + sessionName?: string; +} + +export interface AgentSessionMessageSender extends Partial { + clientId?: string; +} + +export interface AgentSessionMessagePayload { + message: string; + from?: AgentSessionMessageSender; + target: AgentSessionMessageEndpoint; + deliveryMode: AgentSessionMessageDeliveryMode; +} + +export interface AgentSessionMessageReceipt { + target: AgentSessionMessageEndpoint; + from?: AgentSessionMessageSender; + message: string; + deliveredAt: string; + deliveryMode: AgentSessionMessageDeliveryMode; +} + +export function normalizeAgentSessionMessage(message: string): string { + const trimmed = message.trim(); + if (!trimmed) { + throw new Error("Agent session message cannot be empty"); + } + return trimmed; +} + +export function resolveAgentSessionMessageStreamingBehavior( + isTargetStreaming: boolean, + deliveryMode: AgentSessionMessageDeliveryMode | undefined, +): "steer" | "followUp" | undefined { + const mode = deliveryMode ?? "auto"; + if (!isTargetStreaming) { + return undefined; + } + if (mode === "steer") { + return "steer"; + } + return "followUp"; +} + +export function createAgentSessionMessagePrompt(payload: AgentSessionMessagePayload): string { + const lines = ["Agent-to-agent message received."]; + if (payload.from) { + lines.push(`From: ${formatAgentSessionMessageSender(payload.from)}`); + } + lines.push(`To: ${formatAgentSessionMessageEndpoint(payload.target)}`); + lines.push(""); + lines.push(payload.message); + return lines.join("\n"); +} + +export function createAgentSessionMessageReceipt( + payload: AgentSessionMessagePayload, + deliveredAt = new Date().toISOString(), +): AgentSessionMessageReceipt { + return { + target: payload.target, + from: payload.from, + message: payload.message, + deliveredAt, + deliveryMode: payload.deliveryMode, + }; +} + +function formatAgentSessionMessageSender(sender: AgentSessionMessageSender): string { + const parts: string[] = []; + if (sender.sessionName) { + parts.push(sender.sessionName); + } + if (sender.activeSessionId) { + parts.push(`active ${sender.activeSessionId}`); + } + if (sender.sessionId) { + parts.push(`session ${sender.sessionId}`); + } + if (sender.clientId && parts.length > 0) { + parts.push(`client ${sender.clientId}`); + } + if (sender.clientId && parts.length === 0) { + parts.push(`client ${sender.clientId}`); + } + return parts.length > 0 ? parts.join(", ") : "unknown sender"; +} + +function formatAgentSessionMessageEndpoint(endpoint: AgentSessionMessageEndpoint): string { + const name = endpoint.sessionName ? `${endpoint.sessionName}, ` : ""; + return `${name}active ${endpoint.activeSessionId}, session ${endpoint.sessionId}`; +} diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index 5d8c6b5a49..52cb1ffbcd 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-mode.ts @@ -37,6 +37,14 @@ import { type DaemonSocketClient, resolveActiveSessionState, } from "./active-session-state.js"; +import { + type AgentSessionMessageEndpoint, + type AgentSessionMessageSender, + createAgentSessionMessagePrompt, + createAgentSessionMessageReceipt, + normalizeAgentSessionMessage, + resolveAgentSessionMessageStreamingBehavior, +} from "./agent-session-bus.js"; import { serializeDaemonError } from "./daemon-errors.js"; import { bindActiveSessionState } from "./daemon-extension-binding.js"; import { @@ -85,6 +93,7 @@ const DAEMON_COMMAND_TYPES: ReadonlySet = new Set([ "prompt", "steer", "follow_up", + "send_message", "abort", "execute_bash", "abort_bash", @@ -671,6 +680,57 @@ class AgentDaemon { return success(command.id, "follow_up"); } + case "send_message": { + const targetState = this.getSessionState(command.targetActiveSessionId); + const fromState = command.fromActiveSessionId + ? this.getSessionState(command.fromActiveSessionId) + : undefined; + const message = normalizeAgentSessionMessage(command.message); + const payload = { + message, + from: this.createAgentSessionMessageSender(fromState, client.id), + target: this.createAgentSessionMessageEndpoint(targetState), + deliveryMode: command.deliveryMode ?? "auto", + }; + const streamingBehavior = resolveAgentSessionMessageStreamingBehavior( + targetState.runtime.session.isStreaming, + payload.deliveryMode, + ); + let responseSent = false; + const sendSuccessResponse = () => { + if (responseSent) { + return; + } + responseSent = true; + this.write(client, success(command.id, "send_message", createAgentSessionMessageReceipt(payload))); + }; + void targetState.runtime.session + .prompt(createAgentSessionMessagePrompt(payload), { + expandPromptTemplates: false, + streamingBehavior, + source: "rpc", + preflightResult: (didSucceed) => { + if (didSucceed) { + sendSuccessResponse(); + } + }, + }) + .then(() => { + sendSuccessResponse(); + }) + .catch((error) => { + if (responseSent) { + this.broadcastToSession( + targetState, + failure(undefined, "send_message", error, serializeDaemonError(error)), + ); + } else { + this.write(client, failure(command.id, "send_message", error, serializeDaemonError(error))); + } + }); + return undefined; + } + case "abort": { const state = this.getSessionState(command.activeSessionId); await state.runtime.session.abort(); @@ -1059,6 +1119,27 @@ class AgentDaemon { }; } + private createAgentSessionMessageEndpoint(state: ActiveSessionState): AgentSessionMessageEndpoint { + return { + activeSessionId: state.activeSessionId, + sessionId: state.runtime.session.sessionId, + ...(state.runtime.session.sessionName ? { sessionName: state.runtime.session.sessionName } : {}), + }; + } + + private createAgentSessionMessageSender( + state: ActiveSessionState | undefined, + clientId: string, + ): AgentSessionMessageSender { + if (!state) { + return { clientId }; + } + return { + ...this.createAgentSessionMessageEndpoint(state), + clientId, + }; + } + private detachClientFromSession(client: DaemonSocketClient, state: ActiveSessionState): void { detachClientFromActiveSession(client, state); this.write(client, { type: "session_detached", activeSessionId: state.activeSessionId }); diff --git a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts index 91419a1d1e..1653ef0ac2 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts @@ -15,6 +15,7 @@ import type { AgentConnectionSessionTreeNode, AgentConnectionState, } from "../agent-connection/types.js"; +import type { AgentSessionMessageDeliveryMode, AgentSessionMessageReceipt } from "./agent-session-bus.js"; import type { SessionSummary } from "./daemon-session-list.js"; /** @@ -172,6 +173,14 @@ export type DaemonCommand = } | { id?: string; type: "steer"; activeSessionId: string; message: string; images?: ImageContent[] } | { id?: string; type: "follow_up"; activeSessionId: string; message: string; images?: ImageContent[] } + | { + id?: string; + type: "send_message"; + targetActiveSessionId: string; + message: string; + fromActiveSessionId?: string; + deliveryMode?: AgentSessionMessageDeliveryMode; + } | { id?: string; type: "abort"; activeSessionId: string } | { id?: string; @@ -301,6 +310,8 @@ export type DaemonDeleteSavedSessionResult = DeleteSessionFileResult; export type DaemonResourceSnapshot = AgentConnectionResourceSnapshot; +export type DaemonAgentSessionMessageReceipt = AgentSessionMessageReceipt; + export type DaemonOutbound = | DaemonResponse | DaemonRequestProgress diff --git a/packages/coding-agent/test/agent-session-bus.test.ts b/packages/coding-agent/test/agent-session-bus.test.ts new file mode 100644 index 0000000000..b4b7358b06 --- /dev/null +++ b/packages/coding-agent/test/agent-session-bus.test.ts @@ -0,0 +1,84 @@ +import { describe, expect, it } from "vitest"; +import { + createAgentSessionMessagePrompt, + createAgentSessionMessageReceipt, + normalizeAgentSessionMessage, + resolveAgentSessionMessageStreamingBehavior, +} from "../src/modes/daemon/agent-session-bus.js"; + +describe("agent session bus", () => { + it("formats routed messages with sender and target context", () => { + const prompt = createAgentSessionMessagePrompt({ + message: "Use the latest benchmark notes.", + deliveryMode: "auto", + from: { + activeSessionId: "planner", + sessionId: "session-planner", + sessionName: "Planner", + clientId: "client-1", + }, + target: { + activeSessionId: "worker", + sessionId: "session-worker", + sessionName: "Worker", + }, + }); + + expect(prompt).toBe( + [ + "Agent-to-agent message received.", + "From: Planner, active planner, session session-planner, client client-1", + "To: Worker, active worker, session session-worker", + "", + "Use the latest benchmark notes.", + ].join("\n"), + ); + + expect( + createAgentSessionMessagePrompt({ + message: "hello", + deliveryMode: "auto", + from: { clientId: "client-only" }, + target: { + activeSessionId: "worker", + sessionId: "session-worker", + }, + }), + ).toContain("From: client client-only"); + }); + + it("uses follow-up delivery by default only when the target is streaming", () => { + expect(resolveAgentSessionMessageStreamingBehavior(false, "auto")).toBeUndefined(); + expect(resolveAgentSessionMessageStreamingBehavior(true, "auto")).toBe("followUp"); + expect(resolveAgentSessionMessageStreamingBehavior(true, "follow_up")).toBe("followUp"); + expect(resolveAgentSessionMessageStreamingBehavior(true, "steer")).toBe("steer"); + }); + + it("normalizes messages and creates receipts", () => { + const message = normalizeAgentSessionMessage(" hello from another session "); + const receipt = createAgentSessionMessageReceipt( + { + message, + deliveryMode: "follow_up", + target: { + activeSessionId: "target", + sessionId: "session-target", + }, + }, + "2026-06-15T12:00:00.000Z", + ); + + expect(message).toBe("hello from another session"); + expect(receipt).toEqual({ + target: { + activeSessionId: "target", + sessionId: "session-target", + }, + from: undefined, + message: "hello from another session", + deliveredAt: "2026-06-15T12:00:00.000Z", + deliveryMode: "follow_up", + }); + expect(() => normalizeAgentSessionMessage(" ")).toThrow("Agent session message cannot be empty"); + }); +}); diff --git a/packages/coding-agent/test/daemon-command.test.ts b/packages/coding-agent/test/daemon-command.test.ts index 430d3faef4..45cffc680a 100644 --- a/packages/coding-agent/test/daemon-command.test.ts +++ b/packages/coding-agent/test/daemon-command.test.ts @@ -8,6 +8,10 @@ const daemonClientMock = vi.hoisted(() => { name?: string; sessionPath?: string; config?: { extensionFlagValues?: Record }; + targetActiveSessionId?: string; + fromActiveSessionId?: string; + deliveryMode?: string; + message?: string; }; type Response = | { type: "response"; command: string; success: true } @@ -221,6 +225,31 @@ describe("daemon command", () => { sessionPath: "abc123", }); }); + + it("routes daemon send as an agent-to-agent message command", async () => { + await expect( + handleDaemonCommand([ + "daemon", + "--socket", + "/tmp/prime-agent.sock", + "send", + "--from", + "planner", + "--follow-up", + "worker", + "use this context", + ]), + ).resolves.toBe(true); + + const client = daemonClientMock.instances[0]; + expect(client?.requests[0]).toEqual({ + type: "send_message", + targetActiveSessionId: "worker", + fromActiveSessionId: "planner", + deliveryMode: "follow_up", + message: "use this context", + }); + }); }); async function flushPromises(): Promise { From 53d61ed711a15cfb0d10dad0ce9e29c87d8bcc3e Mon Sep 17 00:00:00 2001 From: Seth Date: Mon, 15 Jun 2026 21:24:08 -0700 Subject: [PATCH 2/3] Add safe agent messaging skill --- .../skills/agent-message/SKILL.md | 34 +++ .../skills/agent-message/pyproject.toml | 13 + .../src/agent_message/__init__.py | 44 +++ .../coding-agent/src/cli/daemon-command.ts | 31 ++ .../coding-agent/src/core/agent-messages.ts | 251 +++++++++++++++++ .../src/core/agent-session-services.ts | 2 + .../coding-agent/src/core/agent-session.ts | 69 ++++- packages/coding-agent/src/core/sdk.ts | 1 + .../src/modes/daemon/agent-session-bus.ts | 97 ------- .../src/modes/daemon/daemon-mode.ts | 264 ++++++++++++++---- .../src/modes/daemon/daemon-protocol.ts | 11 +- .../test/agent-session-bus.test.ts | 46 ++- .../coding-agent/test/builtin-skills.test.ts | 9 + .../coding-agent/test/daemon-command.test.ts | 24 ++ .../test/kernel-agent-message-skill.test.ts | 127 +++++++++ 15 files changed, 865 insertions(+), 158 deletions(-) create mode 100644 packages/coding-agent/skills/agent-message/SKILL.md create mode 100644 packages/coding-agent/skills/agent-message/pyproject.toml create mode 100644 packages/coding-agent/skills/agent-message/src/agent_message/__init__.py create mode 100644 packages/coding-agent/src/core/agent-messages.ts delete mode 100644 packages/coding-agent/src/modes/daemon/agent-session-bus.ts create mode 100644 packages/coding-agent/test/kernel-agent-message-skill.test.ts diff --git a/packages/coding-agent/skills/agent-message/SKILL.md b/packages/coding-agent/skills/agent-message/SKILL.md new file mode 100644 index 0000000000..f77c5f0280 --- /dev/null +++ b/packages/coding-agent/skills/agent-message/SKILL.md @@ -0,0 +1,34 @@ +--- +name: agent-message +description: Message other active Prime Agent sessions through the daemon. Use to discover active agents and send a direct text message without spoofing sender identity. +--- + +# Agent Message + +Send direct messages to other active Prime Agent sessions through the local +daemon. The daemon derives your sender identity from the current session; do +not try to include a `from` field. + +Call directly from the kernel: + +```python +agents = await agent_message.list_agents() +receipt = await agent_message.send("worker", "Please inspect the latest result.", mode="auto") +``` + +## API + +- `await agent_message.list_agents()` — returns `current` and `agents`, where + each agent includes active session id, session id, optional name, runtime + kind, cwd, streaming state, and pending message count. +- `await agent_message.send(target, message, mode="auto")` — sends one direct + text message to an active session. `target` is resolved by the daemon like + other live-session selectors. `mode` is `"auto"`, `"follow_up"`, or + `"steer"`. + +## Safety + +- Broadcast sends are not supported. +- Sender identity is daemon-derived and cannot be spoofed from Python. +- The daemon enforces message size, rate, and pending-queue limits before + accepting delivery. diff --git a/packages/coding-agent/skills/agent-message/pyproject.toml b/packages/coding-agent/skills/agent-message/pyproject.toml new file mode 100644 index 0000000000..67790f56d9 --- /dev/null +++ b/packages/coding-agent/skills/agent-message/pyproject.toml @@ -0,0 +1,13 @@ +[project] +name = "agent-message" +version = "0.1.0" +description = "Prime Agent session-to-session messaging skill" +requires-python = ">=3.10" +dependencies = [] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/agent_message"] diff --git a/packages/coding-agent/skills/agent-message/src/agent_message/__init__.py b/packages/coding-agent/skills/agent-message/src/agent_message/__init__.py new file mode 100644 index 0000000000..098801fdee --- /dev/null +++ b/packages/coding-agent/skills/agent-message/src/agent_message/__init__.py @@ -0,0 +1,44 @@ +"""Prime Agent session-to-session messaging skill. + +All routing and sender identity live in the TypeScript daemon. These functions +only call the host bridge exposed inside the Prime Agent IPython kernel. +""" + +from __future__ import annotations + +from typing import Any, Literal + +from rlm import host_request + +MessageMode = Literal["auto", "follow_up", "steer"] + + +async def list_agents() -> dict[str, Any]: + """List active daemon sessions addressable by agent_message.send().""" + return await host_request("agent_message.list") + + +async def send(target: str, message: str, mode: MessageMode = "auto") -> dict[str, Any]: + """Send one direct text message to another active Prime Agent session. + + Args: + target: Active session id, session id/name, or unambiguous suffix. + message: Text payload to deliver. + mode: "auto" queues as follow-up only if the target is streaming; + "follow_up" always uses follow-up when the target is streaming; + "steer" interrupts a streaming target. + """ + if not isinstance(target, str): + raise TypeError(f"target must be str, got {type(target).__name__}") + if not isinstance(message, str): + raise TypeError(f"message must be str, got {type(message).__name__}") + if mode not in ("auto", "follow_up", "steer"): + raise ValueError('mode must be "auto", "follow_up", or "steer"') + return await host_request( + "agent_message.send", + { + "target": target, + "message": message, + "mode": mode, + }, + ) diff --git a/packages/coding-agent/src/cli/daemon-command.ts b/packages/coding-agent/src/cli/daemon-command.ts index 37b1b5ef04..316a212a31 100644 --- a/packages/coding-agent/src/cli/daemon-command.ts +++ b/packages/coding-agent/src/cli/daemon-command.ts @@ -33,6 +33,7 @@ const DAEMON_CLIENT_COMMANDS = new Set([ "rename", "prompt", "send", + "agent-messages", "steer", "follow-up", "state", @@ -190,6 +191,9 @@ async function runDaemonClientCommand(parsed: ParsedDaemonClientCommand): Promis case "send": await runSend(client, parsed.positionals, parsed.json); return; + case "agent-messages": + await runAgentMessages(client, parsed.positionals, parsed.json); + return; case "steer": await runMessageCommand(client, "steer", parsed.positionals, parsed.json); return; @@ -766,6 +770,31 @@ async function runPrompt(client: DaemonClient, args: string[]): Promise { } } +async function runAgentMessages(client: DaemonClient, args: string[], json: boolean): Promise { + const subcommand = args[0]; + switch (subcommand) { + case "status": + await printResponseData(client, { type: "agent_messages_status" }, json); + return; + case "pause": + await printResponseData(client, { type: "agent_messages_pause" }, json); + return; + case "resume": + await printResponseData(client, { type: "agent_messages_resume" }, json); + return; + case "clear": { + const activeSessionId = args[1]; + if (!activeSessionId) { + throw new Error("Usage: daemon agent-messages clear "); + } + await printResponseData(client, { type: "agent_messages_clear", activeSessionId }, json); + return; + } + default: + throw new Error("Usage: daemon agent-messages "); + } +} + async function runSend(client: DaemonClient, args: string[], json: boolean): Promise { const parsed = parseSendArgs(args); const response = await client.request({ @@ -1419,6 +1448,7 @@ ${chalk.bold("Commands:")} detach [session] Detach this client from one session or all sessions prompt Send a prompt, stream events, and exit when idle send [options] Send an agent-to-agent message to another live session + agent-messages Safety controls: status, pause, resume, clear steer Queue a steering message follow-up Queue a follow-up message rename Rename a live session @@ -1436,6 +1466,7 @@ ${chalk.bold("Options:")} --foreground, --no-detach Keep daemon attached to this terminal for debugging --json Print raw JSON for commands with formatted output; attach streams raw protocol JSON send options: --from , --steer, --follow-up + agent-messages clear only clears one explicitly named session Agent options such as --model, --provider, --tools, and --thinking apply to created sessions. ${chalk.bold("Examples:")} diff --git a/packages/coding-agent/src/core/agent-messages.ts b/packages/coding-agent/src/core/agent-messages.ts new file mode 100644 index 0000000000..078d6e6d1d --- /dev/null +++ b/packages/coding-agent/src/core/agent-messages.ts @@ -0,0 +1,251 @@ +import { randomUUID } from "node:crypto"; +import type { HostRequestHandler } from "./kernel/index.js"; + +export const AGENT_MESSAGE_SKILL_NAME = "agent-message"; +export const AGENT_MESSAGE_IMPORT_NAME = "agent_message"; +export const AGENT_MESSAGE_SOURCE = "agent_message"; +export const DEFAULT_AGENT_MESSAGE_MAX_CHARS = 16_384; +export const DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION = 20; +export const DEFAULT_AGENT_MESSAGE_RATE_LIMIT_CAPACITY = 3; +export const DEFAULT_AGENT_MESSAGE_RATE_LIMIT_REFILL_MS = 1000; + +export type AgentSessionMessageDeliveryMode = "auto" | "steer" | "follow_up"; +export type AgentSessionMessageRuntimeKind = "top-level" | "subagent"; + +export interface AgentSessionMessageEndpoint { + activeSessionId: string; + sessionId: string; + sessionName?: string; + runtimeKind?: AgentSessionMessageRuntimeKind; +} + +export interface AgentSessionMessageSender extends Partial { + clientId?: string; +} + +export interface AgentSessionMessageAgentSummary extends AgentSessionMessageEndpoint { + cwd: string; + isStreaming: boolean; + pendingMessageCount: number; + parentActiveSessionId?: string; + rlmChildId?: string; +} + +export interface AgentSessionMessageListResult { + current?: AgentSessionMessageEndpoint; + agents: AgentSessionMessageAgentSummary[]; +} + +export interface AgentSessionMessagePayload { + id: string; + source: typeof AGENT_MESSAGE_SOURCE; + message: string; + from?: AgentSessionMessageSender; + target: AgentSessionMessageEndpoint; + deliveryMode: AgentSessionMessageDeliveryMode; +} + +export interface AgentSessionMessageReceipt { + id: string; + source: typeof AGENT_MESSAGE_SOURCE; + target: AgentSessionMessageEndpoint; + from?: AgentSessionMessageSender; + message: string; + deliveredAt: string; + deliveryMode: AgentSessionMessageDeliveryMode; +} + +export interface AgentSessionMessageSendInput { + target: string; + message: string; + deliveryMode?: AgentSessionMessageDeliveryMode; +} + +export interface AgentSessionMessageController { + listAgents(): AgentSessionMessageListResult; + sendAgentMessage(input: AgentSessionMessageSendInput): Promise; +} + +export interface AgentSessionMessageSafetyStatus { + paused: boolean; + maxMessageChars: number; + maxPendingPerSession: number; + rateLimitCapacity: number; + rateLimitRefillMs: number; +} + +export function createAgentSessionMessageId(): string { + return `agentmsg_${randomUUID()}`; +} + +export function normalizeAgentSessionMessage(message: string, maxChars = DEFAULT_AGENT_MESSAGE_MAX_CHARS): string { + const trimmed = message.trim(); + if (!trimmed) { + throw new Error("Agent session message cannot be empty"); + } + if (trimmed.length > maxChars) { + throw new Error(`Agent session message is too long: ${trimmed.length} chars exceeds ${maxChars}`); + } + return trimmed; +} + +export function normalizeAgentSessionMessageDeliveryMode(value: unknown): AgentSessionMessageDeliveryMode | undefined { + if (value === undefined || value === null) { + return undefined; + } + if (value === "auto" || value === "steer" || value === "follow_up") { + return value; + } + throw new Error('agent_message.send mode must be "auto", "steer", or "follow_up"'); +} + +export function assertDirectAgentMessageTarget(target: string): string { + const normalized = target.trim(); + if (!normalized) { + throw new Error("Agent message target cannot be empty"); + } + if (normalized === "*" || normalized.toLowerCase() === "all" || normalized.toLowerCase() === "broadcast") { + throw new Error("Broadcast agent messaging is not supported"); + } + return normalized; +} + +export function assertAgentMessageQueueCapacity( + pendingMessageCount: number, + maxPending = DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION, +): void { + if (pendingMessageCount >= maxPending) { + throw new Error( + `Target session has too many pending messages: ${pendingMessageCount} pending, limit is ${maxPending}`, + ); + } +} + +export function resolveAgentSessionMessageStreamingBehavior( + isTargetStreaming: boolean, + deliveryMode: AgentSessionMessageDeliveryMode | undefined, +): "steer" | "followUp" | undefined { + const mode = deliveryMode ?? "auto"; + if (!isTargetStreaming) { + return undefined; + } + if (mode === "steer") { + return "steer"; + } + return "followUp"; +} + +export function createAgentSessionMessagePrompt(payload: AgentSessionMessagePayload): string { + const lines = ["Agent-to-agent message received.", `Source: ${payload.source}`]; + if (payload.from) { + lines.push(`From: ${formatAgentSessionMessageSender(payload.from)}`); + } + lines.push(`To: ${formatAgentSessionMessageEndpoint(payload.target)}`); + lines.push(`Message id: ${payload.id}`); + lines.push(""); + lines.push(payload.message); + return lines.join("\n"); +} + +export function createAgentSessionMessageReceipt( + payload: AgentSessionMessagePayload, + deliveredAt = new Date().toISOString(), +): AgentSessionMessageReceipt { + return { + id: payload.id, + source: payload.source, + target: payload.target, + from: payload.from, + message: payload.message, + deliveredAt, + deliveryMode: payload.deliveryMode, + }; +} + +export interface AgentSessionMessageRateLimiterOptions { + capacity?: number; + refillMs?: number; + now?: () => number; +} + +export class AgentSessionMessageRateLimiter { + private readonly capacity: number; + private readonly refillMs: number; + private readonly now: () => number; + private readonly buckets = new Map(); + + constructor(options: AgentSessionMessageRateLimiterOptions = {}) { + this.capacity = options.capacity ?? DEFAULT_AGENT_MESSAGE_RATE_LIMIT_CAPACITY; + this.refillMs = options.refillMs ?? DEFAULT_AGENT_MESSAGE_RATE_LIMIT_REFILL_MS; + this.now = options.now ?? (() => Date.now()); + } + + tryConsume(key: string): { ok: true } | { ok: false; retryAfterMs: number } { + const now = this.now(); + const bucket = this.buckets.get(key) ?? { tokens: this.capacity, updatedAt: now }; + const elapsed = Math.max(0, now - bucket.updatedAt); + const refilledTokens = Math.floor(elapsed / this.refillMs); + if (refilledTokens > 0) { + bucket.tokens = Math.min(this.capacity, bucket.tokens + refilledTokens); + bucket.updatedAt += refilledTokens * this.refillMs; + } + if (bucket.tokens <= 0) { + this.buckets.set(key, bucket); + return { ok: false, retryAfterMs: Math.max(1, bucket.updatedAt + this.refillMs - now) }; + } + bucket.tokens -= 1; + this.buckets.set(key, bucket); + return { ok: true }; + } + + clear(key?: string): void { + if (key) { + this.buckets.delete(key); + return; + } + this.buckets.clear(); + } +} + +export function createAgentMessageHostHandlers( + controller: AgentSessionMessageController, +): Record { + return { + "agent_message.list": async () => controller.listAgents() as unknown as Record, + "agent_message.send": async (payload) => { + if (typeof payload.target !== "string") { + throw new Error("agent_message.send target must be a string"); + } + if (typeof payload.message !== "string") { + throw new Error("agent_message.send message must be a string"); + } + return (await controller.sendAgentMessage({ + target: payload.target, + message: payload.message, + deliveryMode: normalizeAgentSessionMessageDeliveryMode(payload.mode), + })) as unknown as Record; + }, + }; +} + +function formatAgentSessionMessageSender(sender: AgentSessionMessageSender): string { + const parts: string[] = []; + if (sender.sessionName) { + parts.push(sender.sessionName); + } + if (sender.activeSessionId) { + parts.push(`active ${sender.activeSessionId}`); + } + if (sender.sessionId) { + parts.push(`session ${sender.sessionId}`); + } + if (sender.clientId) { + parts.push(`client ${sender.clientId}`); + } + return parts.length > 0 ? parts.join(", ") : "unknown sender"; +} + +function formatAgentSessionMessageEndpoint(endpoint: AgentSessionMessageEndpoint): string { + const name = endpoint.sessionName ? `${endpoint.sessionName}, ` : ""; + return `${name}active ${endpoint.activeSessionId}, session ${endpoint.sessionId}`; +} diff --git a/packages/coding-agent/src/core/agent-session-services.ts b/packages/coding-agent/src/core/agent-session-services.ts index 46ed2458fd..2d296c12f2 100644 --- a/packages/coding-agent/src/core/agent-session-services.ts +++ b/packages/coding-agent/src/core/agent-session-services.ts @@ -2,6 +2,7 @@ import { join } from "node:path"; import type { ThinkingLevel } from "@earendil-works/pi-agent-core"; import type { Model } from "@earendil-works/pi-ai"; import { getAgentDir } from "../config.js"; +import type { AgentSessionMessageController } from "./agent-messages.js"; import { AuthStorage } from "./auth-storage.js"; import type { SessionStartEvent, ToolDefinition } from "./extensions/index.js"; import { ModelRegistry } from "./model-registry.js"; @@ -50,6 +51,7 @@ export interface AgentSessionCreationOptions { initialActiveToolNames?: string[]; allowedToolNames?: string[]; includeGoals?: boolean; + agentMessageController?: AgentSessionMessageController; rlmDepth?: number; rlmMaxDepth?: number; rlmSessionDir?: string; diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index cc9d44829d..55ba5b77a9 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -49,6 +49,16 @@ import { theme } from "../modes/interactive/theme/theme.js"; import { stripFrontmatter } from "../utils/frontmatter.js"; import { sleep } from "../utils/sleep.js"; import { ensureTool, MISSING_RIPGREP_MESSAGE } from "../utils/tools-manager.js"; +import { + AGENT_MESSAGE_SKILL_NAME, + type AgentSessionMessageController, + type AgentSessionMessageListResult, + type AgentSessionMessageReceipt, + assertDirectAgentMessageTarget, + createAgentMessageHostHandlers, + normalizeAgentSessionMessage, + normalizeAgentSessionMessageDeliveryMode, +} from "./agent-messages.js"; import { formatNoApiKeyFoundMessage, formatNoModelSelectedMessage } from "./auth-guidance.js"; import { type BashResult, executeBashWithOperations } from "./bash-executor.js"; import { @@ -298,6 +308,8 @@ export interface AgentSessionConfig { * Default: true. */ includeGoals?: boolean; + /** Daemon-backed agent-to-agent messaging bridge. Omitted for local-only sessions. */ + agentMessageController?: AgentSessionMessageController; /** * Override base tools (useful for custom runtimes). * @@ -684,6 +696,7 @@ export class AgentSession { private _initialActiveToolNames?: string[]; private _allowedToolNames?: Set; private _includeGoals: boolean; + private _agentMessageController?: AgentSessionMessageController; private _baseToolsOverride?: Record; private _sessionStartEvent: SessionStartEvent; private _extensionUIContext?: ExtensionUIContext; @@ -727,6 +740,7 @@ export class AgentSession { this._initialActiveToolNames = config.initialActiveToolNames; this._allowedToolNames = config.allowedToolNames ? new Set(config.allowedToolNames) : undefined; this._includeGoals = config.includeGoals ?? true; + this._agentMessageController = config.agentMessageController; this._baseToolsOverride = config.baseToolsOverride; this._sessionStartEvent = config.sessionStartEvent ?? { type: "session_start", reason: "startup" }; this._rlmDepth = config.rlmDepth ?? parseDepth(process.env.RLM_DEPTH, 0, "RLM_DEPTH"); @@ -1321,6 +1335,35 @@ export class AgentSession { } } + handleAgentMessageHostRequest( + type: string, + payload: Record = {}, + ): AgentSessionMessageListResult | Promise { + if (!this._agentMessageController) { + throw new Error("agent messaging is not available in this session"); + } + switch (type) { + case "agent_message.list": + return this._agentMessageController.listAgents(); + case "agent_message.send": { + if (typeof payload.target !== "string") { + throw new Error("agent_message.send target must be a string"); + } + if (typeof payload.message !== "string") { + throw new Error("agent_message.send message must be a string"); + } + const deliveryMode = normalizeAgentSessionMessageDeliveryMode(payload.mode); + return this._agentMessageController.sendAgentMessage({ + target: assertDirectAgentMessageTarget(payload.target), + message: normalizeAgentSessionMessage(payload.message), + ...(deliveryMode ? { deliveryMode } : {}), + }); + } + default: + throw new Error(`unknown agent message request type "${type}"`); + } + } + private _createGoalFromHost(objective: string, tokenBudget: number | undefined): GoalState { switch (this._goalState.status) { case "active": @@ -3467,11 +3510,14 @@ export class AgentSession { * skill is withheld when goals are disabled for this session. */ private _modelVisibleSkills(): Skill[] { - const skills = this._resourceLoader.getSkills().skills; - if (this._includeGoals) { - return skills; + let skills = this._resourceLoader.getSkills().skills; + if (!this._includeGoals) { + skills = skills.filter((skill) => skill.name !== GOAL_SKILL_NAME); } - return skills.filter((skill) => skill.name !== GOAL_SKILL_NAME); + if (!this._agentMessageController) { + skills = skills.filter((skill) => skill.name !== AGENT_MESSAGE_SKILL_NAME); + } + return skills; } /** Typed handlers for host requests arriving from the IPython kernel comm bridge. */ @@ -3486,6 +3532,21 @@ export class AgentSession { handlers[type] = async (payload) => this.handleGoalHostRequest(type, payload); } } + if (this._agentMessageController) { + Object.assign( + handlers, + createAgentMessageHostHandlers({ + listAgents: () => + this.handleAgentMessageHostRequest("agent_message.list") as AgentSessionMessageListResult, + sendAgentMessage: async (input) => + (await this.handleAgentMessageHostRequest("agent_message.send", { + target: input.target, + message: input.message, + mode: input.deliveryMode, + })) as AgentSessionMessageReceipt, + }), + ); + } return handlers; } diff --git a/packages/coding-agent/src/core/sdk.ts b/packages/coding-agent/src/core/sdk.ts index c6d9f3d0e7..4a8ce0af5d 100644 --- a/packages/coding-agent/src/core/sdk.ts +++ b/packages/coding-agent/src/core/sdk.ts @@ -349,6 +349,7 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} initialActiveToolNames, allowedToolNames, includeGoals, + agentMessageController: options.agentMessageController, extensionRunnerRef, rlmDepth: options.rlmDepth, rlmMaxDepth: options.rlmMaxDepth, diff --git a/packages/coding-agent/src/modes/daemon/agent-session-bus.ts b/packages/coding-agent/src/modes/daemon/agent-session-bus.ts deleted file mode 100644 index 846d2135fb..0000000000 --- a/packages/coding-agent/src/modes/daemon/agent-session-bus.ts +++ /dev/null @@ -1,97 +0,0 @@ -export type AgentSessionMessageDeliveryMode = "auto" | "steer" | "follow_up"; - -export interface AgentSessionMessageEndpoint { - activeSessionId: string; - sessionId: string; - sessionName?: string; -} - -export interface AgentSessionMessageSender extends Partial { - clientId?: string; -} - -export interface AgentSessionMessagePayload { - message: string; - from?: AgentSessionMessageSender; - target: AgentSessionMessageEndpoint; - deliveryMode: AgentSessionMessageDeliveryMode; -} - -export interface AgentSessionMessageReceipt { - target: AgentSessionMessageEndpoint; - from?: AgentSessionMessageSender; - message: string; - deliveredAt: string; - deliveryMode: AgentSessionMessageDeliveryMode; -} - -export function normalizeAgentSessionMessage(message: string): string { - const trimmed = message.trim(); - if (!trimmed) { - throw new Error("Agent session message cannot be empty"); - } - return trimmed; -} - -export function resolveAgentSessionMessageStreamingBehavior( - isTargetStreaming: boolean, - deliveryMode: AgentSessionMessageDeliveryMode | undefined, -): "steer" | "followUp" | undefined { - const mode = deliveryMode ?? "auto"; - if (!isTargetStreaming) { - return undefined; - } - if (mode === "steer") { - return "steer"; - } - return "followUp"; -} - -export function createAgentSessionMessagePrompt(payload: AgentSessionMessagePayload): string { - const lines = ["Agent-to-agent message received."]; - if (payload.from) { - lines.push(`From: ${formatAgentSessionMessageSender(payload.from)}`); - } - lines.push(`To: ${formatAgentSessionMessageEndpoint(payload.target)}`); - lines.push(""); - lines.push(payload.message); - return lines.join("\n"); -} - -export function createAgentSessionMessageReceipt( - payload: AgentSessionMessagePayload, - deliveredAt = new Date().toISOString(), -): AgentSessionMessageReceipt { - return { - target: payload.target, - from: payload.from, - message: payload.message, - deliveredAt, - deliveryMode: payload.deliveryMode, - }; -} - -function formatAgentSessionMessageSender(sender: AgentSessionMessageSender): string { - const parts: string[] = []; - if (sender.sessionName) { - parts.push(sender.sessionName); - } - if (sender.activeSessionId) { - parts.push(`active ${sender.activeSessionId}`); - } - if (sender.sessionId) { - parts.push(`session ${sender.sessionId}`); - } - if (sender.clientId && parts.length > 0) { - parts.push(`client ${sender.clientId}`); - } - if (sender.clientId && parts.length === 0) { - parts.push(`client ${sender.clientId}`); - } - return parts.length > 0 ? parts.join(", ") : "unknown sender"; -} - -function formatAgentSessionMessageEndpoint(endpoint: AgentSessionMessageEndpoint): string { - const name = endpoint.sessionName ? `${endpoint.sessionName}, ` : ""; - return `${name}active ${endpoint.activeSessionId}, session ${endpoint.sessionId}`; -} diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index 1173a91dde..bcecb765c2 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-mode.ts @@ -9,6 +9,27 @@ import { createServer, type Server, type Socket } from "node:net"; import { resolve } from "node:path"; import { VERSION } from "../../config.js"; +import { + AGENT_MESSAGE_SOURCE, + type AgentSessionMessageAgentSummary, + type AgentSessionMessageEndpoint, + type AgentSessionMessageListResult, + type AgentSessionMessagePayload, + AgentSessionMessageRateLimiter, + type AgentSessionMessageReceipt, + type AgentSessionMessageSender, + assertAgentMessageQueueCapacity, + assertDirectAgentMessageTarget, + createAgentSessionMessageId, + createAgentSessionMessagePrompt, + createAgentSessionMessageReceipt, + DEFAULT_AGENT_MESSAGE_MAX_CHARS, + DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION, + DEFAULT_AGENT_MESSAGE_RATE_LIMIT_CAPACITY, + DEFAULT_AGENT_MESSAGE_RATE_LIMIT_REFILL_MS, + normalizeAgentSessionMessage, + resolveAgentSessionMessageStreamingBehavior, +} from "../../core/agent-messages.js"; import { type AgentSessionRuntimeConfig, mergeAgentSessionRuntimeConfig } from "../../core/agent-session-config.js"; import { AgentSessionRuntime, @@ -37,14 +58,6 @@ import { type DaemonSocketClient, resolveActiveSessionState, } from "./active-session-state.js"; -import { - type AgentSessionMessageEndpoint, - type AgentSessionMessageSender, - createAgentSessionMessagePrompt, - createAgentSessionMessageReceipt, - normalizeAgentSessionMessage, - resolveAgentSessionMessageStreamingBehavior, -} from "./agent-session-bus.js"; import { serializeDaemonError } from "./daemon-errors.js"; import { bindActiveSessionState } from "./daemon-extension-binding.js"; import { @@ -94,6 +107,10 @@ const DAEMON_COMMAND_TYPES: ReadonlySet = new Set([ "steer", "follow_up", "send_message", + "agent_messages_status", + "agent_messages_pause", + "agent_messages_resume", + "agent_messages_clear", "abort", "execute_bash", "abort_bash", @@ -168,6 +185,8 @@ class AgentDaemon { private readonly sessions = new Map(); private readonly closingSessions = new Map>(); private readonly signalCleanupHandlers: Array<() => void> = []; + private readonly agentMessageRateLimiter = new AgentSessionMessageRateLimiter(); + private agentMessagesPaused = false; constructor( private readonly socketPath: string, @@ -327,13 +346,38 @@ class AgentDaemon { // visible in session lists again. sessionManager.appendSessionState({ status: "sleep" }); } + let stateRef: ActiveSessionState | undefined; const runtime = await createAgentSessionRuntime(this.options.createRuntime, { cwd: sessionManager.getCwd(), agentDir: config.agentDir, sessionManager, sessionConfig: config, + sessionOptions: { + agentMessageController: { + listAgents: () => { + if (!stateRef) { + throw new Error("Agent message state is not ready for this session yet"); + } + return this.createAgentMessageListResult(stateRef); + }, + sendAgentMessage: (input) => { + if (!stateRef) { + throw new Error("Agent message state is not ready for this session yet"); + } + return this.sendAgentSessionMessage({ + targetSelector: input.target, + message: input.message, + fromState: stateRef, + deliveryMode: input.deliveryMode, + origin: "agent", + }); + }, + }, + }, }); - return this.addRuntime(runtime, command.name); + const state = await this.addRuntime(runtime, command.name); + stateRef = state; + return state; } private findSessionBySessionFile(sessionFile: string | undefined): ActiveSessionState | undefined { @@ -398,6 +442,7 @@ class AgentDaemon { if (options.parentSession.sessionFile) { sessionManager.newSession({ parentSession: options.parentSession.sessionFile }); } + let stateRef: ActiveSessionState | undefined; const runtime = await createAgentSessionRuntime(this.options.createRuntime, { cwd: sessionManager.getCwd(), agentDir: parentState.runtime.services.agentDir, @@ -412,6 +457,26 @@ class AgentDaemon { allowedToolNames: options.allowedToolNames, customTools: options.customTools, includeGoals: options.includeGoals, + agentMessageController: { + listAgents: () => { + if (!stateRef) { + throw new Error("Agent message state is not ready for this session yet"); + } + return this.createAgentMessageListResult(stateRef); + }, + sendAgentMessage: (input) => { + if (!stateRef) { + throw new Error("Agent message state is not ready for this session yet"); + } + return this.sendAgentSessionMessage({ + targetSelector: input.target, + message: input.message, + fromState: stateRef, + deliveryMode: input.deliveryMode, + origin: "agent", + }); + }, + }, rlmDepth: options.rlmDepth, rlmMaxDepth: options.rlmMaxDepth, rlmSessionDir: options.sessionDir, @@ -430,7 +495,8 @@ class AgentDaemon { sessionDir: options.sessionDir, }, }); - await this.addRuntime(runtime); + const state = await this.addRuntime(runtime); + stateRef = state; return runtime; } @@ -682,54 +748,38 @@ class AgentDaemon { } case "send_message": { - const targetState = this.getSessionState(command.targetActiveSessionId); const fromState = command.fromActiveSessionId ? this.getSessionState(command.fromActiveSessionId) : undefined; - const message = normalizeAgentSessionMessage(command.message); - const payload = { - message, - from: this.createAgentSessionMessageSender(fromState, client.id), - target: this.createAgentSessionMessageEndpoint(targetState), - deliveryMode: command.deliveryMode ?? "auto", - }; - const streamingBehavior = resolveAgentSessionMessageStreamingBehavior( - targetState.runtime.session.isStreaming, - payload.deliveryMode, - ); - let responseSent = false; - const sendSuccessResponse = () => { - if (responseSent) { - return; - } - responseSent = true; - this.write(client, success(command.id, "send_message", createAgentSessionMessageReceipt(payload))); - }; - void targetState.runtime.session - .prompt(createAgentSessionMessagePrompt(payload), { - expandPromptTemplates: false, - streamingBehavior, - source: "rpc", - preflightResult: (didSucceed) => { - if (didSucceed) { - sendSuccessResponse(); - } - }, - }) - .then(() => { - sendSuccessResponse(); - }) - .catch((error) => { - if (responseSent) { - this.broadcastToSession( - targetState, - failure(undefined, "send_message", error, serializeDaemonError(error)), - ); - } else { - this.write(client, failure(command.id, "send_message", error, serializeDaemonError(error))); - } - }); - return undefined; + const receipt = await this.sendAgentSessionMessage({ + targetSelector: command.targetActiveSessionId, + message: command.message, + fromState, + clientId: client.id, + deliveryMode: command.deliveryMode, + origin: "cli", + }); + return success(command.id, "send_message", receipt); + } + + case "agent_messages_status": { + return success(command.id, "agent_messages_status", this.getAgentMessageSafetyStatus()); + } + + case "agent_messages_pause": { + this.agentMessagesPaused = true; + return success(command.id, "agent_messages_pause", this.getAgentMessageSafetyStatus()); + } + + case "agent_messages_resume": { + this.agentMessagesPaused = false; + return success(command.id, "agent_messages_resume", this.getAgentMessageSafetyStatus()); + } + + case "agent_messages_clear": { + const state = this.getSessionState(command.activeSessionId); + this.agentMessageRateLimiter.clear(state.activeSessionId); + return success(command.id, "agent_messages_clear", state.runtime.session.clearQueue()); } case "abort": { @@ -1130,10 +1180,12 @@ class AgentDaemon { } private createAgentSessionMessageEndpoint(state: ActiveSessionState): AgentSessionMessageEndpoint { + const metadata = state.runtime.metadata; return { activeSessionId: state.activeSessionId, sessionId: state.runtime.session.sessionId, ...(state.runtime.session.sessionName ? { sessionName: state.runtime.session.sessionName } : {}), + runtimeKind: metadata.kind, }; } @@ -1150,6 +1202,108 @@ class AgentDaemon { }; } + private createAgentMessageAgentSummary(state: ActiveSessionState): AgentSessionMessageAgentSummary { + const metadata = state.runtime.metadata; + return { + ...this.createAgentSessionMessageEndpoint(state), + cwd: state.runtime.cwd, + isStreaming: state.runtime.session.isStreaming, + pendingMessageCount: state.runtime.session.pendingMessageCount, + ...(metadata.parentActiveSessionId ? { parentActiveSessionId: metadata.parentActiveSessionId } : {}), + ...(metadata.rlmChildId ? { rlmChildId: metadata.rlmChildId } : {}), + }; + } + + private createAgentMessageListResult(current: ActiveSessionState): AgentSessionMessageListResult { + return { + current: this.createAgentSessionMessageEndpoint(current), + agents: [...this.sessions.values()].map((state) => this.createAgentMessageAgentSummary(state)), + }; + } + + private getAgentMessageSafetyStatus() { + return { + paused: this.agentMessagesPaused, + maxMessageChars: DEFAULT_AGENT_MESSAGE_MAX_CHARS, + maxPendingPerSession: DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION, + rateLimitCapacity: DEFAULT_AGENT_MESSAGE_RATE_LIMIT_CAPACITY, + rateLimitRefillMs: DEFAULT_AGENT_MESSAGE_RATE_LIMIT_REFILL_MS, + }; + } + + private async sendAgentSessionMessage(options: { + targetSelector: string; + message: string; + fromState?: ActiveSessionState; + clientId?: string; + deliveryMode?: AgentSessionMessagePayload["deliveryMode"]; + origin: "agent" | "cli"; + }): Promise { + if (this.agentMessagesPaused) { + throw new Error("Agent messaging is paused"); + } + const targetSelector = assertDirectAgentMessageTarget(options.targetSelector); + const targetState = this.getSessionState(targetSelector); + const message = normalizeAgentSessionMessage(options.message, DEFAULT_AGENT_MESSAGE_MAX_CHARS); + assertAgentMessageQueueCapacity( + targetState.runtime.session.pendingMessageCount, + DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION, + ); + const senderKey = options.fromState?.activeSessionId ?? `client:${options.clientId ?? "unknown"}`; + const rateLimit = this.agentMessageRateLimiter.tryConsume(senderKey); + if (!rateLimit.ok) { + throw new Error(`Agent messaging rate limit exceeded; retry after ${rateLimit.retryAfterMs}ms`); + } + const payload: AgentSessionMessagePayload = { + id: createAgentSessionMessageId(), + source: AGENT_MESSAGE_SOURCE, + message, + from: this.createAgentSessionMessageSender(options.fromState, options.clientId ?? options.origin), + target: this.createAgentSessionMessageEndpoint(targetState), + deliveryMode: options.deliveryMode ?? "auto", + }; + const streamingBehavior = resolveAgentSessionMessageStreamingBehavior( + targetState.runtime.session.isStreaming, + payload.deliveryMode, + ); + + return new Promise((resolveReceipt, rejectReceipt) => { + let settled = false; + const settleSuccess = () => { + if (settled) { + return; + } + settled = true; + resolveReceipt(createAgentSessionMessageReceipt(payload)); + }; + void targetState.runtime.session + .prompt(createAgentSessionMessagePrompt(payload), { + expandPromptTemplates: false, + streamingBehavior, + source: "rpc", + preflightResult: (didSucceed) => { + if (didSucceed) { + settleSuccess(); + } + }, + }) + .then(() => { + settleSuccess(); + }) + .catch((error) => { + if (settled) { + this.broadcastToSession( + targetState, + failure(undefined, "send_message", error, serializeDaemonError(error)), + ); + return; + } + settled = true; + rejectReceipt(error); + }); + }); + } + private detachClientFromSession(client: DaemonSocketClient, state: ActiveSessionState): void { detachClientFromActiveSession(client, state); this.write(client, { type: "session_detached", activeSessionId: state.activeSessionId }); diff --git a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts index dbf1844b97..5e6a2ef6a1 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts @@ -1,5 +1,10 @@ import type { AgentMessage, ThinkingLevel } from "@earendil-works/pi-agent-core"; import type { ImageContent, Transport } from "@earendil-works/pi-ai"; +import type { + AgentSessionMessageDeliveryMode, + AgentSessionMessageReceipt, + AgentSessionMessageSafetyStatus, +} from "../../core/agent-messages.js"; import type { AgentSessionRuntimeConfig } from "../../core/agent-session-config.js"; import type { SessionCwdIssue } from "../../core/session-cwd.js"; import type { DeleteSessionFileResult } from "../../core/session-file-actions.js"; @@ -15,7 +20,6 @@ import type { AgentConnectionSessionTreeNode, AgentConnectionState, } from "../agent-connection/types.js"; -import type { AgentSessionMessageDeliveryMode, AgentSessionMessageReceipt } from "./agent-session-bus.js"; import type { SessionSummary } from "./daemon-session-list.js"; /** @@ -181,6 +185,10 @@ export type DaemonCommand = fromActiveSessionId?: string; deliveryMode?: AgentSessionMessageDeliveryMode; } + | { id?: string; type: "agent_messages_status" } + | { id?: string; type: "agent_messages_pause" } + | { id?: string; type: "agent_messages_resume" } + | { id?: string; type: "agent_messages_clear"; activeSessionId: string } | { id?: string; type: "abort"; activeSessionId: string } | { id?: string; @@ -312,6 +320,7 @@ export type DaemonDeleteSavedSessionResult = DeleteSessionFileResult; export type DaemonResourceSnapshot = AgentConnectionResourceSnapshot; export type DaemonAgentSessionMessageReceipt = AgentSessionMessageReceipt; +export type DaemonAgentSessionMessageSafetyStatus = AgentSessionMessageSafetyStatus; export type DaemonOutbound = | DaemonResponse diff --git a/packages/coding-agent/test/agent-session-bus.test.ts b/packages/coding-agent/test/agent-session-bus.test.ts index b4b7358b06..c3764de36a 100644 --- a/packages/coding-agent/test/agent-session-bus.test.ts +++ b/packages/coding-agent/test/agent-session-bus.test.ts @@ -1,14 +1,20 @@ import { describe, expect, it } from "vitest"; import { + AGENT_MESSAGE_SOURCE, + AgentSessionMessageRateLimiter, + assertAgentMessageQueueCapacity, + assertDirectAgentMessageTarget, createAgentSessionMessagePrompt, createAgentSessionMessageReceipt, normalizeAgentSessionMessage, resolveAgentSessionMessageStreamingBehavior, -} from "../src/modes/daemon/agent-session-bus.js"; +} from "../src/core/agent-messages.js"; describe("agent session bus", () => { it("formats routed messages with sender and target context", () => { const prompt = createAgentSessionMessagePrompt({ + id: "agentmsg-1", + source: AGENT_MESSAGE_SOURCE, message: "Use the latest benchmark notes.", deliveryMode: "auto", from: { @@ -27,8 +33,10 @@ describe("agent session bus", () => { expect(prompt).toBe( [ "Agent-to-agent message received.", + "Source: agent_message", "From: Planner, active planner, session session-planner, client client-1", "To: Worker, active worker, session session-worker", + "Message id: agentmsg-1", "", "Use the latest benchmark notes.", ].join("\n"), @@ -36,6 +44,8 @@ describe("agent session bus", () => { expect( createAgentSessionMessagePrompt({ + id: "agentmsg-2", + source: AGENT_MESSAGE_SOURCE, message: "hello", deliveryMode: "auto", from: { clientId: "client-only" }, @@ -58,6 +68,8 @@ describe("agent session bus", () => { const message = normalizeAgentSessionMessage(" hello from another session "); const receipt = createAgentSessionMessageReceipt( { + id: "agentmsg-3", + source: AGENT_MESSAGE_SOURCE, message, deliveryMode: "follow_up", target: { @@ -70,6 +82,8 @@ describe("agent session bus", () => { expect(message).toBe("hello from another session"); expect(receipt).toEqual({ + id: "agentmsg-3", + source: AGENT_MESSAGE_SOURCE, target: { activeSessionId: "target", sessionId: "session-target", @@ -80,5 +94,35 @@ describe("agent session bus", () => { deliveryMode: "follow_up", }); expect(() => normalizeAgentSessionMessage(" ")).toThrow("Agent session message cannot be empty"); + expect(() => normalizeAgentSessionMessage("abcd", 3)).toThrow("Agent session message is too long"); + }); + + it("rejects broadcast-style targets and full target queues", () => { + expect(assertDirectAgentMessageTarget(" worker ")).toBe("worker"); + expect(() => assertDirectAgentMessageTarget("*")).toThrow("Broadcast agent messaging is not supported"); + expect(() => assertDirectAgentMessageTarget("all")).toThrow("Broadcast agent messaging is not supported"); + expect(() => assertAgentMessageQueueCapacity(20, 20)).toThrow("Target session has too many pending messages"); + expect(() => assertAgentMessageQueueCapacity(19, 20)).not.toThrow(); + }); + + it("rate limits senders with a token bucket", () => { + let now = 0; + const limiter = new AgentSessionMessageRateLimiter({ + capacity: 3, + refillMs: 1000, + now: () => now, + }); + + expect(limiter.tryConsume("sender")).toEqual({ ok: true }); + expect(limiter.tryConsume("sender")).toEqual({ ok: true }); + expect(limiter.tryConsume("sender")).toEqual({ ok: true }); + expect(limiter.tryConsume("sender")).toEqual({ ok: false, retryAfterMs: 1000 }); + + now = 1000; + expect(limiter.tryConsume("sender")).toEqual({ ok: true }); + expect(limiter.tryConsume("other")).toEqual({ ok: true }); + + limiter.clear("sender"); + expect(limiter.tryConsume("sender")).toEqual({ ok: true }); }); }); diff --git a/packages/coding-agent/test/builtin-skills.test.ts b/packages/coding-agent/test/builtin-skills.test.ts index a8690222e0..1881f03140 100644 --- a/packages/coding-agent/test/builtin-skills.test.ts +++ b/packages/coding-agent/test/builtin-skills.test.ts @@ -172,6 +172,15 @@ describe("builtin skills", () => { expect(goal?.kind === "python" && goal.python.importName).toBe("goal"); }); + it("ships the agent-message skill as a python skill importable as `agent_message`", () => { + const { skills } = loadSkillsFromDir({ dir: getBundledSkillsDir(), source: "builtin" }); + + const agentMessage = skills.find((s) => s.name === "agent-message"); + expect(agentMessage).toBeDefined(); + expect(agentMessage?.kind).toBe("python"); + expect(agentMessage?.kind === "python" && agentMessage.python.importName).toBe("agent_message"); + }); + it("ships the edit skill as a python skill importable as `edit`", () => { const { skills } = loadSkillsFromDir({ dir: getBundledSkillsDir(), source: "builtin" }); diff --git a/packages/coding-agent/test/daemon-command.test.ts b/packages/coding-agent/test/daemon-command.test.ts index 45cffc680a..98352d41fd 100644 --- a/packages/coding-agent/test/daemon-command.test.ts +++ b/packages/coding-agent/test/daemon-command.test.ts @@ -6,6 +6,7 @@ const daemonClientMock = vi.hoisted(() => { type Command = { type: string; name?: string; + activeSessionId?: string; sessionPath?: string; config?: { extensionFlagValues?: Record }; targetActiveSessionId?: string; @@ -250,6 +251,29 @@ describe("daemon command", () => { message: "use this context", }); }); + + it("routes daemon agent message safety controls", async () => { + await expect( + handleDaemonCommand(["daemon", "--socket", "/tmp/prime-agent.sock", "agent-messages", "status"]), + ).resolves.toBe(true); + await expect( + handleDaemonCommand(["daemon", "--socket", "/tmp/prime-agent.sock", "agent-messages", "pause"]), + ).resolves.toBe(true); + await expect( + handleDaemonCommand(["daemon", "--socket", "/tmp/prime-agent.sock", "agent-messages", "resume"]), + ).resolves.toBe(true); + await expect( + handleDaemonCommand(["daemon", "--socket", "/tmp/prime-agent.sock", "agent-messages", "clear", "worker"]), + ).resolves.toBe(true); + + const requests = daemonClientMock.instances.flatMap((client) => client.requests); + expect(requests).toEqual([ + { type: "agent_messages_status" }, + { type: "agent_messages_pause" }, + { type: "agent_messages_resume" }, + { type: "agent_messages_clear", activeSessionId: "worker" }, + ]); + }); }); async function flushPromises(): Promise { diff --git a/packages/coding-agent/test/kernel-agent-message-skill.test.ts b/packages/coding-agent/test/kernel-agent-message-skill.test.ts new file mode 100644 index 0000000000..123dcd775f --- /dev/null +++ b/packages/coding-agent/test/kernel-agent-message-skill.test.ts @@ -0,0 +1,127 @@ +import { mkdirSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { getBundledSkillsDir } from "../src/config.js"; +import type { PythonSkillRuntimeInfo } from "../src/core/skills.js"; +import { IpythonKernelProvisioner } from "../src/core/tools/ipython.js"; + +function bundledAgentMessageSkill(): PythonSkillRuntimeInfo { + const packagePath = join(getBundledSkillsDir(), "agent-message"); + return { + name: "agent-message", + importName: "agent_message", + packagePath, + pyprojectPath: join(packagePath, "pyproject.toml"), + }; +} + +describe("agent-message skill over the kernel host bridge", () => { + let tempDir: string; + let provisioner: IpythonKernelProvisioner | undefined; + + beforeEach(() => { + tempDir = join(tmpdir(), `pi-agent-message-skill-${Date.now()}-${Math.random().toString(36).slice(2)}`); + mkdirSync(tempDir, { recursive: true }); + }); + + afterEach(async () => { + await provisioner?.dispose(); + provisioner = undefined; + rmSync(tempDir, { recursive: true, force: true }); + }); + + it("lists agents and sends without exposing a spoofable sender", async () => { + const requests: Array<{ type: string; payload: Record }> = []; + provisioner = new IpythonKernelProvisioner(tempDir, { + pythonSkills: [bundledAgentMessageSkill()], + hostHandlers: { + "agent_message.list": async (payload) => { + requests.push({ type: "agent_message.list", payload }); + return { + current: { activeSessionId: "alpha", sessionId: "session-alpha" }, + agents: [ + { + activeSessionId: "alpha", + sessionId: "session-alpha", + cwd: tempDir, + isStreaming: false, + pendingMessageCount: 0, + }, + { + activeSessionId: "beta", + sessionId: "session-beta", + sessionName: "Beta", + cwd: tempDir, + isStreaming: true, + pendingMessageCount: 1, + }, + ], + }; + }, + "agent_message.send": async (payload) => { + requests.push({ type: "agent_message.send", payload }); + return { + id: "agentmsg-test", + source: "agent_message", + target: { activeSessionId: payload.target, sessionId: "session-beta" }, + from: { activeSessionId: "alpha", sessionId: "session-alpha" }, + message: payload.message, + deliveredAt: "2026-06-16T00:00:00.000Z", + deliveryMode: payload.mode, + }; + }, + }, + }); + + const manager = await provisioner.ensure(); + const result = await manager.execute(` +import json +agents = await agent_message.list_agents() +receipt = await agent_message.send("beta", "hello beta", mode="follow_up") +print(json.dumps({"agents": agents, "receipt": receipt}, sort_keys=True)) +`); + + expect(result.status).toBe("ok"); + const output = JSON.parse(result.stdout.trim()); + expect(output.agents.agents).toHaveLength(2); + expect(output.receipt).toMatchObject({ + id: "agentmsg-test", + source: "agent_message", + message: "hello beta", + deliveryMode: "follow_up", + }); + expect(requests[0]).toMatchObject({ type: "agent_message.list", payload: { type: "agent_message.list" } }); + expect(requests[1]).toMatchObject({ + type: "agent_message.send", + payload: { + type: "agent_message.send", + target: "beta", + message: "hello beta", + mode: "follow_up", + }, + }); + expect(requests[1].payload).not.toHaveProperty("from"); + }); + + it("validates mode before sending to the host", async () => { + provisioner = new IpythonKernelProvisioner(tempDir, { + pythonSkills: [bundledAgentMessageSkill()], + hostHandlers: { + "agent_message.send": async () => { + throw new Error("should not reach host"); + }, + }, + }); + + const manager = await provisioner.ensure(); + const result = await manager.execute(` +try: + await agent_message.send("beta", "hello", mode="broadcast") +except ValueError as error: + print(f"ValueError: {error}") +`); + expect(result.status).toBe("ok"); + expect(result.stdout.trim()).toBe('ValueError: mode must be "auto", "follow_up", or "steer"'); + }); +}); From cb2f183482ca03b9ead1ceb7113cf19d2e7f04db Mon Sep 17 00:00:00 2001 From: Seth Date: Mon, 15 Jun 2026 23:44:53 -0700 Subject: [PATCH 3/3] Forward agent message controller into daemon sessions --- .../src/core/agent-session-services.ts | 1 + packages/coding-agent/src/main.ts | 1 + .../test/agent-session-services.test.ts | 89 +++++++++++++++++++ 3 files changed, 91 insertions(+) create mode 100644 packages/coding-agent/test/agent-session-services.test.ts diff --git a/packages/coding-agent/src/core/agent-session-services.ts b/packages/coding-agent/src/core/agent-session-services.ts index 2d296c12f2..bae82b27fb 100644 --- a/packages/coding-agent/src/core/agent-session-services.ts +++ b/packages/coding-agent/src/core/agent-session-services.ts @@ -211,6 +211,7 @@ export async function createAgentSessionFromServices( initialActiveToolNames: options.initialActiveToolNames, allowedToolNames: options.allowedToolNames, includeGoals: options.includeGoals, + agentMessageController: options.agentMessageController, rlmDepth: options.rlmDepth, rlmMaxDepth: options.rlmMaxDepth, rlmSessionDir: options.rlmSessionDir, diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index 526cf7a571..fb3c54f29d 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -1092,6 +1092,7 @@ export async function main(args: string[], options?: MainOptions) { initialActiveToolNames: runtimeSessionOptions?.initialActiveToolNames, allowedToolNames: runtimeSessionOptions?.allowedToolNames, includeGoals: runtimeSessionOptions?.includeGoals, + agentMessageController: runtimeSessionOptions?.agentMessageController, rlmDepth: runtimeSessionOptions?.rlmDepth, rlmMaxDepth: runtimeSessionOptions?.rlmMaxDepth, rlmSessionDir: runtimeSessionOptions?.rlmSessionDir, diff --git a/packages/coding-agent/test/agent-session-services.test.ts b/packages/coding-agent/test/agent-session-services.test.ts new file mode 100644 index 0000000000..1f8c9d9e80 --- /dev/null +++ b/packages/coding-agent/test/agent-session-services.test.ts @@ -0,0 +1,89 @@ +import { existsSync, mkdirSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { registerFauxProvider } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it } from "vitest"; +import type { AgentSessionMessageController } from "../src/core/agent-messages.js"; +import { createAgentSessionFromServices, createAgentSessionServices } from "../src/core/agent-session-services.js"; +import { AuthStorage } from "../src/core/auth-storage.js"; +import { SessionManager } from "../src/core/session-manager.js"; + +describe("createAgentSessionFromServices", () => { + const cleanupPaths: string[] = []; + const unregisters: Array<() => void> = []; + + afterEach(() => { + while (unregisters.length > 0) { + unregisters.pop()?.(); + } + while (cleanupPaths.length > 0) { + const path = cleanupPaths.pop(); + if (path && existsSync(path)) { + rmSync(path, { recursive: true, force: true }); + } + } + }); + + it("forwards daemon-backed agent message controllers into AgentSession", async () => { + const tempDir = join(tmpdir(), `pi-session-services-${Date.now()}-${Math.random().toString(36).slice(2)}`); + mkdirSync(tempDir, { recursive: true }); + cleanupPaths.push(tempDir); + + const faux = registerFauxProvider(); + unregisters.push(() => faux.unregister()); + + const authStorage = AuthStorage.inMemory(); + authStorage.setRuntimeApiKey(faux.getModel().provider, "faux-key"); + const services = await createAgentSessionServices({ + cwd: tempDir, + agentDir: tempDir, + authStorage, + resourceLoaderOptions: { + noSkills: true, + noPromptTemplates: true, + noThemes: true, + }, + }); + services.modelRegistry.registerProvider(faux.getModel().provider, { + baseUrl: faux.getModel().baseUrl, + apiKey: "faux-key", + api: faux.api, + models: faux.models, + }); + + const agentMessageController: AgentSessionMessageController = { + listAgents: () => ({ + current: { activeSessionId: "current", sessionId: "session-current", runtimeKind: "top-level" }, + agents: [ + { + activeSessionId: "worker", + sessionId: "session-worker", + runtimeKind: "top-level", + cwd: tempDir, + isStreaming: false, + pendingMessageCount: 0, + }, + ], + }), + sendAgentMessage: async () => { + throw new Error("not used"); + }, + }; + + const { session } = await createAgentSessionFromServices({ + services, + sessionManager: SessionManager.create(tempDir, join(tempDir, "sessions")), + model: faux.getModel(), + agentMessageController, + }); + + try { + expect(session.handleAgentMessageHostRequest("agent_message.list")).toMatchObject({ + current: { activeSessionId: "current" }, + agents: [{ activeSessionId: "worker" }], + }); + } finally { + session.dispose(); + } + }); +});