Merge branch 'main' into fix/plugin-type-walk-node-builtins

This commit is contained in:
Gabriele Magno
2026-10-17 10:49:30 +02:00
committed by GitHub
258 changed files with 10031 additions and 8960 deletions
+6 -6
View File
@@ -1,6 +1,6 @@
{
"name": "@getpaseo/server",
"version": "0.11.2",
"version": "0.12.0-beta.1",
"description": "Paseo backend server",
"files": [
"dist/server",
@@ -107,11 +107,11 @@
"@agentclientprotocol/sdk": "^0.17.1",
"@anthropic-ai/claude-agent-sdk": "0.3.246",
"@anthropic-ai/sdk": "^0.104.2",
"@getpaseo/client": "0.11.2",
"@getpaseo/highlight": "0.11.2",
"@getpaseo/plugin": "0.11.2",
"@getpaseo/protocol": "0.11.2",
"@getpaseo/relay": "0.11.2",
"@getpaseo/client": "0.12.0-beta.1",
"@getpaseo/highlight": "0.12.0-beta.1",
"@getpaseo/plugin": "0.12.0-beta.1",
"@getpaseo/protocol": "0.12.0-beta.1",
"@getpaseo/relay": "0.12.0-beta.1",
"@isaacs/ttlcache": "^2.1.4",
"@modelcontextprotocol/sdk": "1.29.0",
"@opencode-ai/sdk": "1.14.46",
@@ -12,6 +12,13 @@ import {
toAgentPersistenceHandle,
} from "../persistence-hooks.js";
export class AgentNotFoundError extends Error {
constructor(agentId: string) {
super(`Agent not found: ${agentId}`);
this.name = "AgentNotFoundError";
}
}
interface PendingAgentInitialization {
promise: Promise<ManagedAgent>;
options: { broadcastTimeline: boolean };
@@ -94,7 +101,7 @@ export async function ensureAgentLoaded(
const initPromise = (async () => {
const record = await deps.agentStorage.get(agentId);
if (!record) {
throw new Error(`Agent not found: ${agentId}`);
throw new AgentNotFoundError(agentId);
}
const validProviders = deps.validProviders ?? deps.agentManager.getRegisteredProviderIds();
@@ -10929,6 +10929,41 @@ test("authoritative timeline records a daemon-handled submitted prompt before it
}
});
test("tryRunOutOfBand gives providers the command another agent sent, not its envelope", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-agent-out-of-band-"));
const seenPrompts: AgentPromptInput[] = [];
class SlashCommandSession extends TestAgentSession {
override tryHandleOutOfBand(prompt: AgentPromptInput) {
seenPrompts.push(prompt);
return prompt === "/handled" ? { run: async () => {} } : null;
}
}
const client = new (class extends TestAgentClient {
override async createSession(config: AgentSessionConfig) {
return new SlashCommandSession(config);
}
})();
const manager = new AgentManager({ clients: { codex: client }, logger });
const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
try {
const fromAgent = prepareAgentMessage("/handled", { kind: "agent-message", agentId: "sender" });
expect(manager.tryRunOutOfBand(agent.id, fromAgent.prompt)).toBe(true);
expect(seenPrompts).toEqual(["/handled"]);
// A human who pastes an envelope gets it quoted, and it stays a plain prompt.
const pasted = prepareAgentMessage(String(fromAgent.prompt));
expect(manager.tryRunOutOfBand(agent.id, pasted.prompt)).toBe(false);
expect(seenPrompts.at(-1)).toBe(pasted.prompt);
} finally {
await manager.closeAgent(agent.id);
rmSync(workdir, { recursive: true, force: true });
}
});
test("replaceAgentRun succeeds when foreground turn terminal event is never delivered", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-stale-fg-"));
const storagePath = join(workdir, "agents");
@@ -79,7 +79,7 @@ import {
type PendingForegroundRun,
} from "./agent-run-state.js";
import { invokeRewindCapability, type RewindMode } from "./rewind/rewind.js";
import { projectAgentMessage } from "./agent-messages/index.js";
import { parseAgentMessage, projectAgentMessage } 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";
@@ -2459,7 +2459,12 @@ export class AgentManager {
*/
tryRunOutOfBand(agentId: string, prompt: AgentPromptInput, options?: AgentRunOptions): boolean {
const agent = this.requireSessionAgent(agentId);
const handler = agent.session.tryHandleOutOfBand?.(prompt);
// Providers parse slash commands from the raw text, so a command another agent sent
// must reach them without its sender envelope.
const agentMessage = typeof prompt === "string" ? parseAgentMessage(prompt) : null;
const handler = agent.session.tryHandleOutOfBand?.(
agentMessage?.source ? agentMessage.text : prompt,
);
if (!handler) {
return false;
}
@@ -5,7 +5,7 @@ import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
import { describe, expect, it, vi } from "vitest";
import { realpathSync } from "node:fs";
import { access, mkdtemp, mkdir, rm, writeFile } from "node:fs/promises";
import { access, mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises";
import { join, resolve as resolvePath } from "node:path";
import { tmpdir } from "node:os";
import { z } from "zod";
@@ -15,6 +15,8 @@ import { createAgentMcpServer as createAgentMcpServerWithDependencies } from "./
import { AgentManager, type ManagedAgent } from "./agent-manager.js";
import { AgentStorage, type StoredAgentRecord } from "./agent-storage.js";
import { createTestAgentClients } from "../test-utils/fake-agent-client.js";
import { OmpAgentClient } from "./providers/omp/agent.js";
import { FakeOmp } from "./providers/omp/test-utils/fake-omp.js";
import type {
AgentClient,
AgentMode,
@@ -841,6 +843,49 @@ function createAgentMcpServer(options: Parameters<typeof createAgentMcpServerWit
});
}
describe("bundled Paseo skill", () => {
it("documents only advertised create_workspace input names", async () => {
const server = await createAgentMcpServer({
agentManager: new BoundaryAgentManagerFake() as AgentManager,
agentStorage: new BoundaryAgentStorageFake() as AgentStorage,
providerSnapshotManager:
new BoundaryProviderSnapshotManagerFake() as unknown as ProviderSnapshotManager,
logger: createTestLogger(),
});
const client = await connectInMemoryMcpClient(server);
try {
const skill = await readFile(
new URL("../../../../../skills/paseo/SKILL.md", import.meta.url),
"utf8",
);
const paragraph = skill.split("\n").find((line) => line.startsWith("**`create_workspace`**"));
expect(paragraph).toBeTypeOf("string");
// The final sentence describes output, rather than tool inputs.
const inputGuidance = paragraph!.split("Returns ")[0]!;
const documentedInputs = [...inputGuidance.matchAll(/`([a-zA-Z][a-zA-Z0-9]*)`/g)].map(
(match) => match[1]!,
);
expect(documentedInputs.length).toBeGreaterThan(0);
const { tools } = await client.listTools();
const tool = tools.find((entry) => entry.name === "create_workspace")!;
const properties = tool.inputSchema.properties!;
const advertisedNamesAndValues = new Set([
...Object.keys(properties),
...Object.values(properties).flatMap((property) =>
typeof property === "object" && property !== null && Array.isArray(property.enum)
? property.enum
: [],
),
]);
expect(documentedInputs.filter((name) => !advertisedNamesAndValues.has(name))).toEqual([]);
} finally {
await client.close();
await server.close();
}
});
});
describe("browser MCP tools", () => {
const logger = createTestLogger();
@@ -3856,8 +3901,7 @@ class HeldTurnAgentSession implements AgentSession {
}
tryHandleOutOfBand(prompt: AgentPromptInput) {
const text = typeof prompt === "string" ? (parseAgentMessage(prompt)?.text ?? prompt) : null;
return text === "/side-command" ? { run: async () => {} } : null;
return prompt === "/side-command" ? { run: async () => {} } : null;
}
async startTurn(prompt: AgentPromptInput): Promise<{ turnId: string }> {
@@ -4557,6 +4601,180 @@ describe("send_agent_prompt MCP tool", () => {
});
});
describe("send_agent_prompt MCP tool to an OMP child", () => {
const logger = createTestLogger();
const existingCwd = process.cwd();
async function setupOmpChild() {
const workdir = await mkdtemp(join(tmpdir(), "mcp-omp-child-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);
const parentClient = new HeldTurnAgentClient("claude", false);
const omp = new FakeOmp();
const agentManager = new AgentManager({
clients: {
claude: parentClient,
omp: new OmpAgentClient({
logger,
runtime: omp,
noTurnScheduler: { waitForSettle: async () => {} },
runtimeSettings: { command: { mode: "replace", argv: [process.execPath] } },
}),
},
registry: storage,
logger,
});
const parent = await agentManager.createAgent(
{ provider: "claude", cwd: existingCwd },
undefined,
{ workspaceId: "wks_parent" },
);
const child = await agentManager.createAgent(
{ provider: "omp", cwd: existingCwd, title: "OMP child" },
undefined,
{ workspaceId: "wks_parent" },
);
const server = await createAgentMcpServer({
agentManager,
agentStorage: storage,
callerAgentId: parent.id,
providerSnapshotManager: createOpenCodeManager().manager,
logger,
});
return {
agentManager,
storage,
workdir,
parentClient,
ompSession: omp.latestSession(),
childId: child.id,
tool: registeredTool(server, "send_agent_prompt"),
};
}
async function expectOneFinishNotification(
agentManager: AgentManager,
parentClient: HeldTurnAgentClient,
childId: string,
) {
await vi.waitFor(() => expect(agentManager.getAgent(childId)?.lifecycle).toBe("idle"));
await vi.waitFor(() => {
expect(parentClient.sessions[0]!.prompts).toHaveLength(1);
expect(parentClient.sessions[0]!.prompts[0]).toContain(
`Agent ${childId} (OMP child) finished.`,
);
});
}
it.each([
{ command: "follow-up", streamingBehavior: "followUp" },
{ command: "steer", streamingBehavior: "steer" },
] as const)(
"delivers /$command into a turn OMP started on its own, without aborting it",
async ({ command, streamingBehavior }) => {
const { agentManager, storage, workdir, parentClient, ompSession, childId, tool } =
await setupOmpChild();
try {
ompSession.beginTurn();
ompSession.streamAssistantText("background job finished, working on it");
await vi.waitFor(() => {
expect(agentManager.getAgent(childId)?.activeTurnId).toMatch(/^autonomous-/);
});
const response = await invokeToolWithParsedInput(tool, {
agentId: childId,
prompt: `/${command} then summarize the logs`,
});
expect(response.structuredContent).toMatchObject({
success: true,
disposition: "out_of_band",
status: "running",
});
await vi.waitFor(() => {
expect(ompSession.prompts).toEqual([
{ message: "then summarize the logs", imageCount: 0, streamingBehavior },
]);
});
expect(ompSession.abortRequested).toBe(false);
expect(agentManager.getAgent(childId)?.lifecycle).toBe("running");
ompSession.finishTurn();
await expectOneFinishNotification(agentManager, parentClient, childId);
} finally {
await removeAgentStateDir(agentManager, storage, workdir);
}
},
);
it.each(["before ack", "after ack"])(
"reports an idle follow-up rejected %s and notifies its parent",
async (order) => {
const { agentManager, storage, workdir, parentClient, ompSession, childId, tool } =
await setupOmpChild();
try {
ompSession.promptAck = { requestId: "follow-up-rejected" };
const rejection = {
type: "prompt_result",
id: "follow-up-rejected",
agentInvoked: false,
status: "error",
sessionSettled: true,
error: { message: "Provider rejected the follow-up", retryable: false },
} as const;
const promptStarted = ompSession.nextPrompt();
const response = invokeToolWithParsedInput(tool, {
agentId: childId,
prompt: "/follow-up check the build",
});
await promptStarted;
if (order === "before ack") ompSession.emit(rejection);
await response;
if (order === "after ack") ompSession.emit(rejection);
await vi.waitFor(() => expect(agentManager.getAgent(childId)?.lifecycle).toBe("error"));
expect(agentManager.getAgent(childId)?.lastError).toContain(
"Provider rejected the follow-up",
);
await vi.waitFor(() => expect(parentClient.sessions[0]!.prompts).toHaveLength(1));
expect(parentClient.sessions[0]!.prompts[0]).toContain(
`Agent ${childId} (OMP child) errored.`,
);
} finally {
await removeAgentStateDir(agentManager, storage, workdir);
}
},
);
it("lets an idle child start a run with the follow-up", async () => {
const { agentManager, storage, workdir, parentClient, ompSession, childId, tool } =
await setupOmpChild();
try {
const response = await invokeToolWithParsedInput(tool, {
agentId: childId,
prompt: "/follow-up check the build",
});
expect(response.structuredContent).toMatchObject({
success: true,
disposition: "out_of_band",
status: "running",
});
await vi.waitFor(() => {
expect(ompSession.prompts).toEqual([
{ message: "check the build", imageCount: 0, streamingBehavior: "followUp" },
]);
});
ompSession.beginTurn();
ompSession.streamAssistantText("Build is green");
await vi.waitFor(() => expect(agentManager.getAgent(childId)?.lifecycle).toBe("running"));
ompSession.finishTurn();
await expectOneFinishNotification(agentManager, parentClient, childId);
} finally {
await removeAgentStateDir(agentManager, storage, workdir);
}
});
});
describe("update_agent MCP tool", () => {
const logger = createTestLogger();
@@ -88,6 +88,7 @@ import {
type AgentStreamEvent,
type AgentTimelineItem,
type AgentUsage,
type AgentUsageSession,
type FetchCatalogOptions,
type ProviderRefreshContext,
type ImportableProviderSession,
@@ -112,6 +113,7 @@ import {
import { importSessionFromPersistence } from "../provider-session-import.js";
import {
checkProviderLaunchAvailable,
createProviderEnv,
createProviderEnvSpec,
resolveProviderLaunch,
type ProviderRuntimeSettings,
@@ -456,6 +458,8 @@ export type ACPCatalogModelResolver = (
interface ACPAgentClientOptions {
provider: string;
// Provider id usage sources match a session against; ACP sessions without one report no usage.
usageProvider?: string;
logger: Logger;
runtimeSettings?: ProviderRuntimeSettings;
defaultCommand: [string, ...string[]];
@@ -488,6 +492,7 @@ interface ACPAgentClientOptions {
interface ACPAgentSessionOptions {
provider: string;
usageProvider?: string;
logger: Logger;
runtimeSettings?: ProviderRuntimeSettings;
defaultCommand: [string, ...string[]];
@@ -910,6 +915,7 @@ function isACPCreateConfigUnattended(input: AgentCreateConfigUnattendedInput): b
export class ACPAgentClient implements AgentClient {
readonly provider: string;
private readonly usageProvider?: string;
readonly capabilities: AgentCapabilityFlags;
readonly resolveCreateConfig = resolveACPCreateConfig;
readonly isCreateConfigUnattended = isACPCreateConfigUnattended;
@@ -951,6 +957,7 @@ export class ACPAgentClient implements AgentClient {
constructor(options: ACPAgentClientOptions) {
this.provider = options.provider;
this.usageProvider = options.usageProvider;
this.terminateProcess = options.terminateProcess ?? terminateWithTreeKill;
this.capabilities = options.capabilities ?? DEFAULT_ACP_CAPABILITIES;
this.logger = options.logger.child({
@@ -989,6 +996,7 @@ export class ACPAgentClient implements AgentClient {
{ ...config, provider: this.provider },
{
provider: this.provider,
usageProvider: this.usageProvider,
logger: this.logger,
runtimeSettings: this.runtimeSettings,
defaultCommand: this.defaultCommand,
@@ -1046,6 +1054,7 @@ export class ACPAgentClient implements AgentClient {
const providerOptions = ACPProviderOptionsSchema.parse(mergedConfig.providerOptions ?? {});
const session = new ACPAgentSession(mergedConfig, {
provider: this.provider,
usageProvider: this.usageProvider,
logger: this.logger,
runtimeSettings: this.runtimeSettings,
defaultCommand: this.defaultCommand,
@@ -1715,6 +1724,9 @@ export class ACPAgentSession implements AgentSession, ACPClient {
) => Promise<void>;
private readonly agentId?: string;
private readonly launchEnv?: Record<string, string>;
private readonly usageProvider?: string;
private readonly usageSessionKey = randomUUID();
private readonly harnessEnvironment: Record<string, string>;
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
private readonly pendingPermissions = new Map<string, PendingPermission>();
private pendingUserMessage: PendingUserMessage | null = null;
@@ -1778,6 +1790,11 @@ export class ACPAgentSession implements AgentSession, ACPClient {
this.availableModes = options.defaultModes;
this.agentId = options.agentId;
this.launchEnv = options.launchEnv;
this.usageProvider = options.usageProvider;
this.harnessEnvironment = createProviderEnv({
runtimeSettings: options.runtimeSettings,
overlays: [options.launchEnv],
});
this.initialHandle = options.handle;
this.resumePurpose = options.resumePurpose ?? "interactive";
this.config = { ...config, provider: options.provider };
@@ -1794,6 +1811,16 @@ export class ACPAgentSession implements AgentSession, ACPClient {
return this.sessionId;
}
usageSession(): AgentUsageSession | null {
if (this.closed || !this.usageProvider) return null;
return {
provider: this.usageProvider,
model: this.currentModel ?? undefined,
env: this.harnessEnvironment,
sessionKey: this.usageSessionKey,
};
}
async initializeNewSession(): Promise<void> {
try {
const spawned = await this.spawnProcess();
@@ -4,7 +4,7 @@ import path from "node:path";
import { afterEach, describe, expect, test, vi } from "vitest";
import { createTestLogger } from "../../../../test-utils/test-logger.js";
import type { AgentStreamEvent } from "../../agent-sdk-types.js";
import type { AgentSession, AgentStreamEvent } from "../../agent-sdk-types.js";
import type { AgentTimelineRow } from "../../agent-manager.js";
import { projectTimelineRows } from "../../timeline-projection.js";
import { ClaudeAgentClient } from "./agent.js";
@@ -105,6 +105,39 @@ async function collectUntilTerminal(
return events;
}
async function collectTimelineAcrossTurns(
session: AgentSession,
prompt: string,
turnCount: number,
): Promise<AgentTimelineRow[]> {
const rows: AgentTimelineRow[] = [];
let currentTurnId: string | undefined;
let completedTurns = 0;
let unsubscribe = () => {};
const turnsCompleted = new Promise<void>((resolve) => {
unsubscribe = session.subscribe((event) => {
if (event.type === "turn_started") currentTurnId = event.turnId;
if (event.type === "timeline") {
rows.push({
seq: rows.length + 1,
timestamp: new Date(Date.UTC(2026, 9, 4, 0, 0, rows.length)).toISOString(),
item: event.item,
...(currentTurnId ? { turnId: currentTurnId } : {}),
});
}
if (event.type === "turn_completed" && ++completedTurns === turnCount) resolve();
});
});
try {
await session.startTurn(prompt);
await turnsCompleted;
return rows;
} finally {
unsubscribe();
await session.close();
}
}
describe("background Claude subagents", () => {
afterEach(() => {
queryFactory.mockReset();
@@ -431,6 +464,124 @@ describe("background Claude subagents", () => {
type: "sub_agent",
subAgentType: "general-purpose",
description: "Count files here",
log: "[Bash] ls -1",
},
});
});
test("keeps one logged Task card when the child streams on after the parent's turn ends", async () => {
// A child can stream across the main session's turn boundary. Claude opens the next
// main-session turn with system/init; bare child frames do not open a turn.
queryFactory.mockImplementation(() =>
buildQueryMock([
{ type: "system", subtype: "init", session_id: "bg-session", permissionMode: "default" },
{
type: "assistant",
message: {
model: "claude-haiku-4-5",
content: [
{
type: "tool_use",
id: "toolu_background",
name: "Agent",
input: {
subagent_type: "general-purpose",
description: "Count files here",
prompt: "Run ls and reply with the number of entries.",
run_in_background: true,
},
},
],
},
},
{
type: "system",
subtype: "task_started",
task_id: "bg-task",
tool_use_id: "toolu_background",
task_type: "local_agent",
subagent_type: "general-purpose",
description: "Count files here",
prompt: "Run ls and reply with the number of entries.",
},
{
type: "user",
message: {
role: "user",
content: [
{
type: "tool_result",
tool_use_id: "toolu_background",
content: "Async agent launched successfully.",
},
],
},
},
{
type: "assistant",
parent_tool_use_id: "toolu_background",
message: {
model: "claude-haiku-4-5",
content: [
{
type: "tool_use",
id: "toolu_ls",
name: "Bash",
input: { command: "ls -1", description: "List files" },
},
],
},
},
{ type: "result", subtype: "success", usage: {}, total_cost_usd: 0 },
{ type: "system", subtype: "init", session_id: "bg-session", permissionMode: "default" },
{
type: "user",
parent_tool_use_id: "toolu_background",
message: {
role: "user",
content: [{ type: "tool_result", tool_use_id: "toolu_ls", content: "a\nb\nc" }],
},
},
{
type: "assistant",
parent_tool_use_id: "toolu_background",
message: { model: "claude-haiku-4-5", content: [{ type: "text", text: "3" }] },
},
{
type: "system",
subtype: "task_updated",
task_id: "bg-task",
patch: { status: "completed" },
},
{
type: "system",
subtype: "task_notification",
task_id: "bg-task",
tool_use_id: "toolu_background",
status: "completed",
},
{ type: "result", subtype: "success", usage: {}, total_cost_usd: 0 },
]),
);
const session = await new ClaudeAgentClient({
logger: createTestLogger(),
queryFactory,
resolveBinary: async () => "/test/claude/bin",
}).createSession({ provider: "claude", cwd: process.cwd() });
const rows = await collectTimelineAcrossTurns(session, "delegate work", 2);
const cards = projectTimelineRows({ rows, mode: "projected" })
.map((entry) => entry.item)
.filter((item) => item.type === "tool_call" && item.callId === "toolu_background");
expect(cards).toHaveLength(1);
expect(cards[0]).toMatchObject({
status: "completed",
detail: {
type: "sub_agent",
subAgentType: "general-purpose",
description: "Count files here",
log: "[Bash] ls -1",
},
});
});
@@ -140,11 +140,11 @@ function imageMessages(items: AgentTimelineItem[]): string[] {
return items
.filter((item) => item.type === "assistant_message")
.map((item) => (item as { text: string }).text)
.filter((text) => text.startsWith("!["));
.filter((text) => text.trimStart().startsWith("!["));
}
function markdownImageSource(markdown: string): string {
const match = markdown.match(/^!\[[^\]]*]\((.*)\)$/);
const match = markdown.trim().match(/^!\[[^\]]*]\((.*)\)$/);
if (!match) {
throw new Error(`Expected markdown image, got: ${markdown}`);
}
@@ -0,0 +1,157 @@
import { expect, test } from "vitest";
import type { Query, SDKMessage } from "@anthropic-ai/claude-agent-sdk";
import { createTestLogger } from "../../../../test-utils/test-logger.js";
import type { AgentStreamEvent } from "../../agent-sdk-types.js";
import { ClaudeAgentClient } from "./agent.js";
import type { ClaudeQueryInput } from "./query.js";
function deferred<T>() {
let resolve!: (value: T) => void;
const promise = new Promise<T>((done) => {
resolve = done;
});
return { promise, resolve };
}
// The injected SDK adapter consumes real session input and holds a command open
// until the test releases it. Closing the query interrupts that command.
class ForegroundQuery {
readonly work = deferred<void>();
readonly closed = deferred<void>();
private stopped = false;
private readonly output: AsyncGenerator<SDKMessage>;
constructor(input: ClaudeQueryInput) {
this.output = this.events(input);
}
private async *events(input: ClaudeQueryInput): AsyncGenerator<SDKMessage> {
if (typeof input.prompt === "string") throw new Error("Expected streaming input");
for await (const _prompt of input.prompt) {
yield {
type: "system",
subtype: "init",
session_id: "settings-session",
model: "claude-haiku-5-5",
permissionMode: "auto",
} as SDKMessage;
yield {
type: "assistant",
message: {
content: [
{ type: "tool_use", id: "tool_sleep", name: "Bash", input: { command: "sleep 20" } },
],
},
} as SDKMessage;
await this.work.promise;
if (this.stopped) return;
yield { type: "assistant", message: { content: "DONE" } } as SDKMessage;
yield {
type: "result",
subtype: "success",
usage: { input_tokens: 1, cache_read_input_tokens: 0, output_tokens: 1 },
total_cost_usd: 0,
} as SDKMessage;
}
}
next() {
return this.output.next();
}
return() {
this.close();
return this.output.return();
}
close() {
this.stopped = true;
this.closed.resolve();
this.work.resolve();
}
async interrupt() {
this.close();
}
async setPermissionMode() {}
async setModel() {}
async supportedCommands() {
return [];
}
[Symbol.asyncIterator]() {
return this;
}
}
test.each(["mode", "model", "commands"] as const)(
"a pending thinking change preserves the current turn during a %s request",
async (control) => {
const queries: ForegroundQuery[] = [];
const efforts: ClaudeQueryInput["options"]["effort"][] = [];
const client = new ClaudeAgentClient({
logger: createTestLogger(),
resolveBinary: async () => "/test/claude",
queryFactory: (input) => {
const query = new ForegroundQuery(input);
queries.push(query);
efforts.push(input.options.effort);
return query as unknown as Query;
},
});
const session = await client.createSession({
provider: "claude",
cwd: process.cwd(),
model: "claude-haiku-5-5",
modeId: "auto",
thinkingOptionId: "low",
});
const running = deferred<void>();
let terminal = deferred<string>();
const events: AgentStreamEvent[] = [];
session.subscribe((event) => {
events.push(event);
if (
event.type === "timeline" &&
event.item.type === "tool_call" &&
event.item.status === "running"
)
running.resolve();
if (
event.type === "turn_completed" ||
event.type === "turn_failed" ||
event.type === "turn_canceled"
)
terminal.resolve(event.type);
});
try {
await session.startTurn("Run the foreground command");
await running.promise;
expect(await session.setThinkingOption?.("medium")).toMatchObject({
message: "Thinking level applies next turn",
});
if (control === "mode") await session.setMode("acceptEdits");
else if (control === "model") await session.setModel("claude-sonnet-5-6");
else await session.listCommands();
queries[0].work.resolve();
expect(
await Promise.race([
terminal.promise,
queries[0].closed.promise.then(() => "query closed before turn finished"),
]),
).toBe("turn_completed");
expect(
events.some(
(event) =>
event.type === "timeline" &&
event.item.type === "assistant_message" &&
event.item.text === "DONE",
),
).toBe(true);
terminal = deferred<string>();
await session.startTurn("Run another command with the new thinking level");
queries[1].work.resolve();
expect(await terminal.promise).toBe("turn_completed");
expect(efforts).toEqual(["low", "medium"]);
} finally {
await session.close();
}
},
);
@@ -2228,18 +2228,37 @@ class ClaudeAgentSession implements AgentSession {
getToolInput: (toolUseId) => this.toolUseCache.get(toolUseId)?.input ?? null,
readWorkflowResult: readClaudeWorkflowResultFile,
});
/**
* Subagent tool calls Claude has already answered while the child still runs, with the input
* they were launched with and the turn that answered them. A background child is answered at
* launch and streams its frames afterwards.
*/
private readonly settledSubagentToolCalls = new Map<
string,
{
item: Extract<AgentTimelineItem, { type: "tool_call" }>;
input: AgentMetadata | null;
turnId: string | null;
}
>();
private readonly sidechainTracker = new ClaudeSidechainTracker({
getToolInput: (toolUseId) => this.toolUseCache.get(toolUseId)?.input ?? null,
getToolInput: (toolUseId) =>
this.toolUseCache.get(toolUseId)?.input ??
this.settledSubagentToolCallInCurrentTurn(toolUseId)?.input ??
null,
// Releases that predate the task protocol announce nothing, so the tracker keeps deriving
// identity and status from frames for them. Detecting the capability beats comparing version
// strings: it reacts to what this session actually does.
isDescriptorOwnedElsewhere: () => this.taskProtocolSource.isActive,
// The parent's tool call owns its card once its result arrives. A background child streams
// its frames after Claude has already answered the call, and re-emitting the card from them
// would replace the settled card with an unlabeled, running one.
// A background child streams its frames after Claude has already answered the call, so its
// card is rebuilt from the settled call: a running one would replace the settled, labeled card.
// A card belongs to the turn that answered the call, so frames from a later turn leave it be.
needsSyntheticParentToolCard: (toolUseId) =>
this.toolUseCache.has(toolUseId) &&
(this.toolUseCache.has(toolUseId) ||
this.settledSubagentToolCallInCurrentTurn(toolUseId) !== undefined) &&
this.taskProtocolSource.needsSyntheticParentToolCard(toolUseId),
getSettledParentToolCall: (toolUseId) =>
this.settledSubagentToolCallInCurrentTurn(toolUseId)?.item,
});
private persistedHistory: PersistedTimelineEntry[] = [];
private persistedProviderSubagentEvents: Extract<
@@ -2591,13 +2610,13 @@ class ClaudeAgentSession implements AgentSession {
const normalized = isPermissionMode(modeId) ? modeId : "default";
assertClaudeModeCanRun(normalized, this.harnessEnvironment);
const previousMode = this.currentMode;
const launchesQuery = !this.query || this.queryRestartNeeded;
const previousQuery = this.query;
const activeQuery = await this.ensureQuery(normalized);
try {
await activeQuery.setPermissionMode(normalized);
} catch (error) {
// The query was launched in the rejected mode; relaunch in the current mode next time.
if (launchesQuery) this.queryRestartNeeded = true;
if (activeQuery !== previousQuery) this.queryRestartNeeded = true;
throw error;
}
if (normalized === "plan") {
@@ -2876,6 +2895,7 @@ class ClaudeAgentSession implements AgentSession {
this.cancelCurrentTurn = null;
this.turnState = "idle";
this.sidechainTracker.clear();
this.settledSubagentToolCalls.clear();
this.taskProtocolSource.reset();
this.input?.end();
this.query?.close?.();
@@ -3294,7 +3314,9 @@ class ClaudeAgentSession implements AgentSession {
}
private async ensureQuery(launchMode: PermissionMode = this.currentMode): Promise<Query> {
if (this.query && !this.queryRestartNeeded) {
// Controls use the running turn's query. Pending launch settings take effect
// after that turn finishes, when its active query handle has been cleared.
if (this.query && (!this.queryRestartNeeded || this.activeForegroundQuery === this.query)) {
return this.query;
}
@@ -4361,6 +4383,7 @@ class ClaudeAgentSession implements AgentSession {
for (const event of foldSubagentObservations(subagentObservations)) {
events.push({ type: "provider_subagent", provider: "claude", event });
}
this.releaseFinishedSubagentToolCalls(subagentObservations);
for (const observation of subagentObservations) {
if (observation.kind !== "declared") continue;
if (!this.taskProtocolSource.needsSyntheticParentToolCard(observation.id)) continue;
@@ -5548,28 +5571,22 @@ class ClaudeAgentSession implements AgentSession {
const { images, text } = splitClaudeToolResultImages(block.content);
const output = this.buildToolOutput(text, block, entry);
if (block.is_error) {
this.pushToolCall(
mapClaudeFailedToolCall({
const settledToolCall = block.is_error
? mapClaudeFailedToolCall({
name: toolName,
callId,
input: entry?.input ?? null,
output: output ?? null,
error: { ...block, content: text },
}),
items,
);
} else {
this.pushToolCall(
mapClaudeCompletedToolCall({
})
: mapClaudeCompletedToolCall({
name: toolName,
callId,
input: entry?.input ?? null,
output: output ?? null,
}),
items,
);
}
});
this.pushToolCall(settledToolCall, items);
this.rememberSettledSubagentToolCall(settledToolCall, entry);
for (const image of images) {
const imageItem = renderProviderImageOutputAsAssistantMarkdown(image, {
@@ -5585,6 +5602,34 @@ class ClaudeAgentSession implements AgentSession {
}
}
private rememberSettledSubagentToolCall(
item: Extract<AgentTimelineItem, { type: "tool_call" }> | null,
entry: ToolUseCacheEntry | undefined,
): void {
const subagentId = item ? this.taskProtocolSource.resolveSubagentId(item.callId) : undefined;
if (!item || !subagentId) return;
this.settledSubagentToolCalls.set(subagentId, {
item,
input: entry?.input ?? null,
turnId: this.activeForegroundTurnId ?? this.autonomousTurn?.id ?? null,
});
}
private settledSubagentToolCallInCurrentTurn(subagentId: string) {
const settled = this.settledSubagentToolCalls.get(subagentId);
const currentTurnId = this.activeForegroundTurnId ?? this.autonomousTurn?.id ?? null;
return settled && settled.turnId === currentTurnId ? settled : undefined;
}
/** A finished child streams no more frames, so its settled call has nothing left to update. */
private releaseFinishedSubagentToolCalls(observations: SubagentObservation[]): void {
for (const observation of observations) {
if (observation.kind === "status" && observation.status !== "running") {
this.settledSubagentToolCalls.delete(observation.id);
}
}
}
private buildToolOutput(
content: unknown,
block: ClaudeContentChunk,
@@ -18,6 +18,8 @@ interface ClaudeContentChunk {
[key: string]: unknown;
}
type ClaudeToolCallItem = Extract<AgentTimelineItem, { type: "tool_call" }>;
interface SubAgentActionEntry {
index: number;
toolName: string;
@@ -64,6 +66,7 @@ export class ClaudeSidechainTracker {
private readonly getToolInput: (toolUseId: string) => AgentMetadata | null | undefined;
private readonly isDescriptorOwnedElsewhere: () => boolean;
private readonly needsSyntheticParentToolCard: (toolUseId: string) => boolean;
private readonly getSettledParentToolCall: (toolUseId: string) => ClaudeToolCallItem | undefined;
constructor(input: {
getToolInput: (toolUseId: string) => AgentMetadata | null | undefined;
@@ -74,10 +77,17 @@ export class ClaudeSidechainTracker {
*/
isDescriptorOwnedElsewhere?: () => boolean;
needsSyntheticParentToolCard?: (toolUseId: string) => boolean;
/**
* The parent's tool call once Claude has answered it. A background child streams its frames
* after that answer, so its card keeps the settled status and output and gains the child's
* action log.
*/
getSettledParentToolCall?: (toolUseId: string) => ClaudeToolCallItem | undefined;
}) {
this.getToolInput = input.getToolInput;
this.isDescriptorOwnedElsewhere = input.isDescriptorOwnedElsewhere ?? (() => false);
this.needsSyntheticParentToolCard = input.needsSyntheticParentToolCard ?? (() => true);
this.getSettledParentToolCall = input.getSettledParentToolCall ?? (() => undefined);
}
handleMessage(message: SDKMessage, parentToolUseId: string): AgentStreamEvent[] {
@@ -119,17 +129,19 @@ export class ClaudeSidechainTracker {
return [];
}
const toolCall = mapClaudeRunningToolCall({
name: "Task",
callId: parentToolUseId,
input: null,
output: null,
});
const toolCall =
this.getSettledParentToolCall(parentToolUseId) ??
mapClaudeRunningToolCall({
name: "Task",
callId: parentToolUseId,
input: null,
output: null,
});
if (!toolCall) {
return [];
}
const detail: Extract<AgentTimelineItem, { type: "tool_call" }>["detail"] = {
const detail: ClaudeToolCallItem["detail"] = {
type: "sub_agent",
...(state.subAgentType ? { subAgentType: state.subAgentType } : {}),
...(state.description ? { description: state.description } : {}),
@@ -502,7 +502,7 @@ function asInternals(session: CodexTestSession): CodexSessionTestAccess {
}
function markdownImageSource(markdown: string): string {
const match = markdown.match(/^!\[[^\]]*]\((.*)\)$/);
const match = markdown.trim().match(/^!\[[^\]]*]\((.*)\)$/);
if (!match) {
throw new Error(`Expected markdown image, got: ${markdown}`);
}
@@ -6000,7 +6000,7 @@ describe("Codex app-server provider", () => {
turnId: "test-turn",
item: {
type: "assistant_message",
text: "![Image](file:///tmp/paseo%20image.png)",
text: "\n\n![Image](file:///tmp/paseo%20image.png)\n\n",
},
},
]);
@@ -6032,7 +6032,7 @@ describe("Codex app-server provider", () => {
turnId: "test-turn",
item: {
type: "assistant_message",
text: `![Image](${expectedPath})`,
text: `\n\n![Image](${expectedPath})\n\n`,
},
},
]);
@@ -1,10 +1,14 @@
import type { SessionConfigOption } from "@agentclientprotocol/sdk";
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { describe, expect, test, vi } from "vitest";
import { ACPAgentSession } from "./acp-agent.js";
import type { SpawnedACPProcess, SessionStateResponse } from "./acp-agent.js";
import type { AgentSessionConfig } from "../agent-sdk-types.js";
import { CURSOR_FAST_FEATURE_OPTION, CursorACPAgentClient } from "./cursor-acp-agent.js";
import { GenericACPAgentClient } from "./generic-acp-agent.js";
import { createTestLogger } from "../../../test-utils/test-logger.js";
function captureWarnings(): {
@@ -355,3 +359,70 @@ describe("CursorACPAgentClient session start", () => {
await expect(session.initializeNewSession()).rejects.toThrow("Invalid params");
});
});
describe("CursorACPAgentClient usage session", () => {
test("matches usage sources as cursor, with the agent's provider and launch env", async () => {
await withFakeACPAgent(async (command, cwd) => {
const client = new CursorACPAgentClient({
logger: createTestLogger(),
command,
env: { XDG_CONFIG_HOME: "/profile/config" },
});
const session = await client.createSession(
{ provider: "acp", cwd },
{ agentId: "agent-1", env: { HOME: "/agent/home" } },
);
const usage = session.usageSession?.();
expect(usage).toMatchObject({
provider: "cursor",
env: { XDG_CONFIG_HOME: "/profile/config", HOME: "/agent/home" },
});
expect(session.usageSession?.()?.sessionKey).toBe(usage?.sessionKey);
await session.close();
expect(session.usageSession?.()).toBeNull();
});
});
test("leaves other ACP agents without a usage session", async () => {
await withFakeACPAgent(async (command, cwd) => {
const client = new GenericACPAgentClient({ logger: createTestLogger(), command });
const session = await client.createSession({ provider: "acp", cwd });
try {
expect(session.usageSession?.()).toBeNull();
} finally {
await session.close();
}
});
});
});
async function withFakeACPAgent(
run: (command: [string, ...string[]], cwd: string) => Promise<void>,
): Promise<void> {
const testDir = await mkdtemp(path.join(tmpdir(), "paseo-cursor-usage-"));
try {
const scriptPath = path.join(testDir, "fake-acp-agent.cjs");
await writeFile(scriptPath, fakeACPAgentScript, "utf8");
await run([process.execPath, scriptPath], testDir);
} finally {
await rm(testDir, { recursive: true, force: true });
}
}
const fakeACPAgentScript = `
const readline = require("node:readline");
const rl = readline.createInterface({ input: process.stdin });
function write(message) {
process.stdout.write(JSON.stringify({ jsonrpc: "2.0", ...message }) + "\\n");
}
rl.on("line", (line) => {
const message = JSON.parse(line);
if (message.method === "initialize") {
write({ id: message.id, result: { protocolVersion: message.params?.protocolVersion ?? 1, agentCapabilities: {} } });
} else if (message.method === "session/new") {
write({ id: message.id, result: { sessionId: "session-1" } });
} else if (message.id !== undefined) {
write({ id: message.id, result: {} });
}
});
`;
@@ -92,6 +92,7 @@ export class CursorACPAgentClient extends GenericACPAgentClient {
env: options.env,
providerId: options.providerId,
label: options.label,
usageProvider: "cursor",
// cursor-agent publishes slash commands asynchronously via available_commands_update.
waitForInitialCommands: true,
initialCommandsWaitTimeoutMs: CURSOR_INITIAL_COMMANDS_WAIT_TIMEOUT_MS,
@@ -22,6 +22,7 @@ interface GenericACPAgentClientOptions {
env?: Record<string, string>;
providerId?: string;
label?: string;
usageProvider?: string;
waitForInitialCommands?: boolean;
initialCommandsWaitTimeoutMs?: number;
diagnosticPhaseTimeoutMs?: number;
@@ -41,6 +42,7 @@ export class GenericACPAgentClient extends ACPAgentClient {
constructor(options: GenericACPAgentClientOptions) {
super({
provider: "acp",
usageProvider: options.usageProvider,
logger: options.logger,
runtimeSettings: {
env: options.env,
@@ -39,6 +39,19 @@ const ABORTED_TERMINAL_RESPONSE: OmpAgentMessage = {
errorMessage: "Interrupted by user",
};
test("an idle follow-up is submitted as a prompt so OMP can start a turn", async () => {
const runtime = new FakeOmp();
const client = new OmpAgentClient({ logger: createTestLogger(), runtime });
const session = await client.createSession({ provider: "omp", cwd: process.cwd() });
onTestFinished(() => session.close());
const handler = session.tryHandleOutOfBand?.("/follow-up check the build");
expect(handler).not.toBeNull();
await handler!.run({ emit: () => {} });
expect(runtime.latestSession().prompts).toEqual([
{ message: "check the build", imageCount: 0, streamingBehavior: "followUp" },
]);
});
test("OMP ready timeout defaults to 20 seconds and RPC timeout overrides both", () => {
expect(resolveOmpProviderOptions({}).runtimeOptions).toMatchObject({
readyTimeoutMs: 20_000,
@@ -705,6 +705,7 @@ export class OmpAgentSession implements AgentSession {
private activeAssistantMessageId: string | null = null;
private activeTurnTerminalAssistantMessage: OmpAgentMessage | null = null;
private activeTurnStarted = false;
private outOfBandTurnStartAnnounced = false;
private activeTurnHasUserMessage = false;
private activeNoTurnPromptText: string | null = null;
private readonly pendingNoTurnOutputs: Array<{ turnId: string; message: string }> = [];
@@ -849,6 +850,20 @@ export class OmpAgentSession implements AgentSession {
}
async startTurn(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<StartTurnResult> {
return this.startPrompt({ prompt, options });
}
private async startPrompt({
prompt,
options,
streamingBehavior,
announceTurn = false,
}: {
prompt: AgentPromptInput;
options?: AgentRunOptions;
streamingBehavior?: "steer" | "followUp";
announceTurn?: boolean;
}): Promise<StartTurnResult> {
if (this.activeTurnId) {
throw new Error("An OMP turn is already active");
}
@@ -861,17 +876,26 @@ export class OmpAgentSession implements AgentSession {
this.activeAssistantMessageId = null;
this.activeTurnTerminalAssistantMessage = null;
this.activeTurnStarted = false;
this.outOfBandTurnStartAnnounced = false;
this.activeTurnHasUserMessage = false;
this.activePromptRequestId = null;
this.clearNoTurnBuffers();
this.activeNoTurnPromptText = payload.text;
this.usagePoller.startTurn();
if (announceTurn) {
this.outOfBandTurnStartAnnounced = true;
this.emit({ type: "turn_started", provider: this.provider, turnId });
}
void (async () => {
try {
if (this.runtimeDead) await this.replaceRuntime(this.currentModeId ?? "full");
if (this.closed) throw new Error("OMP session is closed");
const ack = await this.runtimeSession.prompt(payload.text, payload.images);
const ack = await this.runtimeSession.prompt({
message: payload.text,
images: payload.images,
streamingBehavior,
});
this.activePromptRequestId = ack.requestId ?? null;
const correlatedResult = ack.requestId
? this.pendingPromptResults.get(ack.requestId)
@@ -1191,6 +1215,7 @@ export class OmpAgentSession implements AgentSession {
this.activeAssistantMessageId = null;
this.activeTurnTerminalAssistantMessage = null;
this.activeTurnStarted = false;
this.outOfBandTurnStartAnnounced = false;
this.activeTurnHasUserMessage = false;
this.pendingClientMessages.length = 0;
this.clearNoTurnBuffers();
@@ -1235,12 +1260,13 @@ export class OmpAgentSession implements AgentSession {
if (!message) {
return null;
}
const streamingBehavior = commandName === "steer" ? "steer" : "followUp";
return {
run: async () => {
if (commandName === "steer") {
this.runtimeSession.steer(message);
if (!this.activeTurnId && !this.activeTurnStarted) {
await this.startPrompt({ prompt: message, streamingBehavior, announceTurn: true });
} else {
this.runtimeSession.followUp(message);
await this.runtimeSession.prompt({ message, streamingBehavior });
}
},
};
@@ -1804,11 +1830,11 @@ export class OmpAgentSession implements AgentSession {
case "turn_start":
this.activeTurnStarted = true;
this.clearNoTurnBuffers();
this.emit({
type: "turn_started",
provider: this.provider,
turnId,
});
if (this.outOfBandTurnStartAnnounced) {
this.outOfBandTurnStartAnnounced = false;
} else {
this.emit({ type: "turn_started", provider: this.provider, turnId });
}
return;
case "message_start":
this.handleMessageStart(event);
@@ -143,6 +143,26 @@ describe("OMP CLI runtime", () => {
await expect(result).rejects.toThrow("rejected");
await session.close();
});
test("sends the prompt streamingBehavior only when one is given", async () => {
const child = createOmpChild();
const session = await createRuntime(child, []).startSession({ cwd: "/workspace/project" });
for (const streamingBehavior of ["followUp", undefined] as const) {
const pending = capturePendingCommand(child, "prompt");
const result = session.prompt({ message: "after this, update the docs", streamingBehavior });
const command = await pending;
expect(withoutRequestId(command)).toEqual({
type: "prompt",
message: "after this, update the docs",
...(streamingBehavior ? { streamingBehavior } : {}),
});
child.stdout.write(
`${JSON.stringify({ id: command.id, type: "response", command: "prompt", success: true })}\n`,
);
await result;
}
await session.close();
});
test("uses the configured RPC timeout and attributes the pending phase", async () => {
vi.useFakeTimers();
const child = createOmpChild();
@@ -438,7 +458,7 @@ describe("OMP CLI runtime", () => {
replyToCommands(child, () => undefined);
const session = await createRuntime(child).startSession({ cwd: "/workspace/project" });
await expect(session.prompt("hello")).resolves.toEqual({ requestId: "req_1" });
await expect(session.prompt({ message: "hello" })).resolves.toEqual({ requestId: "req_1" });
});
test("reports OMP 18.3's late prompt rejection as a failed prompt result", async () => {
@@ -455,7 +475,7 @@ describe("OMP CLI runtime", () => {
const events: unknown[] = [];
session.onEvent((event) => events.push(event));
await expect(session.prompt("hello")).resolves.toEqual({ requestId: "req_1" });
await expect(session.prompt({ message: "hello" })).resolves.toEqual({ requestId: "req_1" });
expect(events).toContainEqual({
type: "prompt_result",
@@ -15,6 +15,7 @@ import {
type OmpRuntime,
type OmpRuntimeLaunch,
type OmpRuntimeSession,
type OmpPromptInput,
type OmpStartSessionInput,
} from "./runtime.js";
import {
@@ -142,14 +143,12 @@ class OmpCliRuntimeSession implements OmpRuntimeSession {
};
}
async prompt(
message: string,
images?: Array<{ type: "image"; data: string; mimeType: string }>,
): Promise<OmpPromptAck> {
async prompt({ message, images, streamingBehavior }: OmpPromptInput): Promise<OmpPromptAck> {
const { id: requestId, promise } = this.process.startRequest({
type: "prompt",
message,
...(images?.length ? { images } : {}),
...(streamingBehavior ? { streamingBehavior: streamingBehavior } : {}),
});
const ack = OmpPromptAckSchema.parse(await promise) ?? {};
return { requestId, ...ack };
@@ -282,13 +281,6 @@ class OmpCliRuntimeSession implements OmpRuntimeSession {
await this.request({ type: "steer", message, ...(images?.length ? { images } : {}) });
}
followUp(
message: string,
images?: Array<{ type: "image"; data: string; mimeType: string }>,
): void {
this.process.send({ type: "follow_up", message, ...(images?.length ? { images } : {}) });
}
async handoff(customInstructions?: string): Promise<void> {
await this.request({
type: "handoff",
@@ -535,6 +535,7 @@ export const OmpRpcCommandSchema = z.discriminatedUnion("type", [
type: z.literal("prompt"),
message: z.string(),
images: z.array(OmpImageContentSchema).optional(),
streamingBehavior: z.enum(["steer", "followUp"]).optional(),
}),
z.object({
...OmpCommandBase,
@@ -42,13 +42,18 @@ export interface OmpStartSessionInput {
extraArgs?: string[];
}
export interface OmpPromptInput {
message: string;
images?: Array<{ type: "image"; data: string; mimeType: string }>;
streamingBehavior?: "steer" | "followUp";
}
export interface OmpRuntimeSession {
readonly environment: Record<string, string>;
onEvent(callback: (event: OmpRuntimeEvent) => void): () => void;
prompt(
message: string,
images?: Array<{ type: "image"; data: string; mimeType: string }>,
): Promise<OmpPromptAck>;
// With streamingBehavior, a busy OMP queues the message instead of rejecting it, and an
// idle OMP starts a run with it.
prompt(input: OmpPromptInput): Promise<OmpPromptAck>;
compact(customInstructions?: string): Promise<void>;
setAutoCompaction(enabled: boolean): Promise<void>;
abort(): Promise<void>;
@@ -71,10 +76,6 @@ export interface OmpRuntimeSession {
message: string,
images?: Array<{ type: "image"; data: string; mimeType: string }>,
): Promise<void>;
followUp(
message: string,
images?: Array<{ type: "image"; data: string; mimeType: string }>,
): void;
handoff(customInstructions?: string): Promise<void>;
respondToExtensionUiRequest(
id: string,
@@ -2,6 +2,7 @@ import type {
OmpRuntime,
OmpRuntimeLaunch,
OmpRuntimeSession,
OmpPromptInput,
OmpStartSessionInput,
} from "../runtime.js";
import type {
@@ -133,7 +134,11 @@ export class FakeOmpSession implements OmpRuntimeSession {
readonly environment: Record<string, string>;
fastModeResult = { enabled: false, active: false };
readonly setFastModeRequests: boolean[] = [];
readonly prompts: Array<{ message: string; imageCount: number }> = [];
readonly prompts: Array<{
message: string;
imageCount: number;
streamingBehavior?: "steer" | "followUp";
}> = [];
readonly compactRequests: Array<{ customInstructions?: string }> = [];
readonly setAutoCompactionRequests: boolean[] = [];
readonly subagentSubscriptionRequests: FakeOmpSubagentSubscriptionLevel[] = [];
@@ -143,7 +148,6 @@ export class FakeOmpSession implements OmpRuntimeSession {
readonly handoffRequests: Array<{ customInstructions?: string }> = [];
readonly steerRequests: Array<{ message: string; imageCount: number }> = [];
steerError: Error | null = null;
readonly followUpRequests: Array<{ message: string; imageCount: number }> = [];
readonly hostToolSetRequests: OmpRpcHostToolDefinition[][] = [];
readonly hostToolResults: OmpRpcHostToolResult[] = [];
readonly hostToolUpdates: OmpRpcHostToolUpdate[] = [];
@@ -216,14 +220,19 @@ export class FakeOmpSession implements OmpRuntimeSession {
};
}
async prompt(
message: string,
images?: Array<{ type: "image"; data: string; mimeType: string }>,
): Promise<OmpPromptAck> {
if (this.state.isStreaming || this.state.isCompacting) {
async prompt({ message, images, streamingBehavior }: OmpPromptInput): Promise<OmpPromptAck> {
const busy = this.state.isStreaming || this.state.isCompacting;
if (busy && !streamingBehavior) {
throw new Error("Agent is already processing");
}
this.prompts.push({ message, imageCount: images?.length ?? 0 });
this.prompts.push({
message,
imageCount: images?.length ?? 0,
...(streamingBehavior ? { streamingBehavior: streamingBehavior } : {}),
});
if (busy) {
return {};
}
this.promptWaiters.shift()?.();
const heldPrompt = this.nextHeldPrompt;
if (heldPrompt) {
@@ -412,13 +421,6 @@ export class FakeOmpSession implements OmpRuntimeSession {
if (this.steerError) throw this.steerError;
}
followUp(
message: string,
images?: Array<{ type: "image"; data: string; mimeType: string }>,
): void {
this.followUpRequests.push({ message, imageCount: images?.length ?? 0 });
}
sendHostToolResult(result: OmpRpcHostToolResult): void {
this.hostToolResults.push(result);
this.hostToolResultWaiters.shift()?.(result);
@@ -0,0 +1,202 @@
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { expect, test } from "vitest";
import { createTestLogger } from "../../../../test-utils/test-logger.js";
import { createPaseoToolCatalog } from "../../tools/paseo-tools.js";
import { AgentManager } from "../../agent-manager.js";
import { AgentStorage } from "../../agent-storage.js";
import { ProviderSnapshotManager } from "../../provider-snapshot-manager.js";
import { OpenCodeBridge } from "./bridge.js";
import { ScheduleService } from "../../../schedule/service.js";
import { createWorkspaceScriptsService } from "../../../session/workspace-scripts/workspace-scripts-service.js";
import {
FileBackedProjectRegistry,
FileBackedWorkspaceRegistry,
} from "../../../workspace-registry.js";
import { createServiceProxySubsystem } from "../../../service-proxy.js";
import { WorkspaceScriptRuntimeStore } from "../../../workspace-script-runtime-store.js";
import { createTerminalManager } from "../../../../terminal/terminal-manager.js";
test("bridge preserves tool failure messages and classifies invalid requests as client errors", async () => {
const paseoHome = await mkdtemp(path.join(tmpdir(), "paseo-7045-"));
const logger = createTestLogger();
const providerSnapshotManager = new ProviderSnapshotManager({ logger });
const agentManager = new AgentManager({ logger });
const agentStorage = new AgentStorage(paseoHome, logger);
const unused = async (): Promise<never> => {
throw new Error("Unexpected workspace or agent creation");
};
const scheduleService = new ScheduleService({
paseoHome,
logger,
agentManager,
agentStorage,
createAgent: unused,
createDirectoryWorkspace: unused,
createPaseoWorktreeWorkspace: unused,
archiveWorkspace: unused,
runner: async () => ({ agentId: null, output: "done" }),
});
const workspaceScripts = createWorkspaceScriptsService({
serviceProxy: createServiceProxySubsystem({ logger }),
scriptRuntimeStore: new WorkspaceScriptRuntimeStore(),
terminalManager: createTerminalManager(),
workspaceRegistry: new FileBackedWorkspaceRegistry(
path.join(paseoHome, "workspaces.json"),
logger,
),
projectRegistry: new FileBackedProjectRegistry(path.join(paseoHome, "projects.json"), logger),
workspaceGitService: {
peekSnapshot: () => {
throw new Error("Unexpected git lookup");
},
},
getDaemonTcpPort: null,
getDaemonTcpHost: null,
serviceProxyPublicBaseUrl: null,
resolveScriptHealth: null,
logger,
emit: () => {},
publishStatusUpdate: () => {},
spawnWorkspaceScript: unused,
assertAutomationAllowed: unused,
});
const catalog = createPaseoToolCatalog({
workspaceScripts,
agentManager,
agentStorage,
scheduleService,
providerSnapshotManager,
logger,
});
const bridge = new OpenCodeBridge({ paseoHome, logger });
try {
await bridge.start();
bridge.setManifestCatalog(catalog);
bridge.bindSession({ sessionId: "diagnosis", env: {}, tools: catalog });
const config = JSON.parse(bridge.decorateServerEnv({}).OPENCODE_CONFIG_CONTENT);
const { baseUrl, token } = config.plugin[0][1];
const completed = await scheduleService.create({
prompt: "Run once",
name: "Daily check",
cadence: { type: "cron", expression: "0 0 * * *" },
target: { type: "new-agent", config: { provider: "codex", cwd: paseoHome } },
maxRuns: 1,
runOnCreate: true,
});
await scheduleService.tick();
expect((await scheduleService.inspect(completed.id)).status).toBe("completed");
const manifest = await fetch(`${baseUrl}/_internal/opencode/tools`, {
headers: { Authorization: `Bearer ${token}` },
});
const { tools } = await manifest.json();
expect(
tools.find((tool: { name: string }) => tool.name === "create_agent").inputSchema.properties,
).not.toHaveProperty("background");
expect(
tools.find((tool: { name: string }) => tool.name === "update_schedule").inputSchema
.properties,
).not.toHaveProperty("every");
await writeFile(
path.join(paseoHome, "schedules", "broken.json"),
JSON.stringify({ id: "broken", prompt: 17 }),
);
const observed = [];
for (const [tool, body, expected] of [
[
"create_agent",
JSON.stringify({ title: "Probe", provider: "opencode", initialPrompt: "Reply OK" }),
400,
],
["create_agent", "{", 400],
["nonexistent_tool", "{}", 404],
["inspect_schedule", JSON.stringify({ id: "missing" }), 400],
...["pause_schedule", "resume_schedule", "run_schedule_once"].map(
(action) => [action, JSON.stringify({ id: completed.id }), 400] as const,
),
[
"create_schedule",
JSON.stringify({ prompt: "Probe", provider: "codex", cron: "not-a-cron" }),
400,
],
[
"create_schedule",
JSON.stringify({
prompt: "Probe",
provider: "codex",
cron: "0 0 * * *",
timezone: "Mars/Phobos",
}),
400,
],
["update_schedule", JSON.stringify({ id: completed.id, name: "" }), 200],
["update_schedule", JSON.stringify({ id: completed.id, cron: "0 0 30 2 *" }), 400],
["update_schedule", JSON.stringify({ id: completed.id, expiresIn: "not-a-duration" }), 400],
["send_agent_prompt", JSON.stringify({ agentId: "missing", prompt: "Hello" }), 400],
["inspect_schedule", JSON.stringify({ id: "broken" }), 500],
["list_workspace_scripts", JSON.stringify({ workspaceId: "wks_typo" }), 500],
[
"start_workspace_script",
JSON.stringify({ workspaceId: "wks_typo", scriptName: "web" }),
500,
],
] as const) {
const response = await fetch(
`${baseUrl}/_internal/opencode/sessions/diagnosis/tools/${tool}`,
{
method: "POST",
headers: { Authorization: `Bearer ${token}`, "Content-Type": "application/json" },
body,
},
);
observed.push({
tool,
body,
expected,
status: response.status,
payload: await response.json(),
});
}
for (const result of observed) expect.soft(result.status, result.body).toBe(result.expected);
expect((await scheduleService.inspect(completed.id)).name).toBeNull();
expect(
tools.find((tool: { name: string }) => tool.name === "update_schedule").inputSchema.properties
.name,
).not.toHaveProperty("minLength");
for (const result of observed.slice(14)) {
expect.soft(result.payload).toEqual({ error: "Workspace not found: wks_typo" });
}
expect.soft(observed[12].payload).toEqual({ error: "Agent not found: missing" });
expect.soft(observed[13].payload).toEqual({ error: expect.stringContaining('"prompt"') });
expect.soft(observed[13].payload).not.toHaveProperty("issues");
expect(observed[3].payload).toEqual({ error: "Schedule not found: missing" });
for (const result of observed.slice(4, 7))
expect(result.payload).toEqual({ error: `Schedule ${completed.id} is already completed` });
for (const [index, field] of [
[7, "cron"],
[8, "timezone"],
[10, "cron"],
[11, "expiresIn"],
] as const) {
expect(observed[index].payload).toMatchObject({
error: "Invalid tool input",
issues: expect.arrayContaining([expect.objectContaining({ path: [field] })]),
});
}
expect(observed[0].payload).toMatchObject({
error: "Invalid tool input",
issues: expect.arrayContaining([
expect.objectContaining({
path: ["provider"],
message: expect.stringContaining("provider/model"),
}),
]),
});
} finally {
await bridge.close();
await providerSnapshotManager.shutdown();
await rm(paseoHome, { recursive: true, force: true });
}
});
@@ -13,8 +13,11 @@ export default async function paseoPlugin(input, options) {
});
const payload = await response.json().catch(() => ({}));
if (!response.ok) {
const error = new Error(payload.error ?? `Paseo OpenCode bridge failed: ${response.status}`);
const detail = formatErrorDetail(payload.issues ?? payload.details);
const message = payload.error ?? `Paseo OpenCode bridge failed: ${response.status}`;
const error = new Error(detail ? `${message}: ${detail}` : message);
error.status = response.status;
if (detail) error.details = detail;
throw error;
}
return payload;
@@ -79,6 +82,17 @@ export default async function paseoPlugin(input, options) {
};
}
function formatErrorDetail(details) {
if (!Array.isArray(details)) return typeof details === "string" ? details : "";
return details
.map((issue) => {
const path =
Array.isArray(issue?.path) && issue.path.length > 0 ? issue.path.join(".") : "input";
return `${path}: ${issue?.message ?? "Invalid value"}`;
})
.join("; ");
}
function logPluginError(stage, context, error) {
console.error(`[paseo-opencode-plugin] ${stage} failed`, {
...context,
@@ -9,6 +9,7 @@ import { afterEach, describe, expect, test, vi } from "vitest";
import { createTestLogger } from "../../../../test-utils/test-logger.js";
import type { PaseoToolCatalog } from "../../tools/types.js";
import { PaseoToolInputError } from "../../tools/types.js";
import { createPaseoToolCatalog } from "../../tools/paseo-tools.js";
import { AgentManager } from "../../agent-manager.js";
import { AgentStorage } from "../../agent-storage.js";
@@ -36,8 +37,12 @@ function createCatalog(): PaseoToolCatalog {
description: "Returns the supplied value.",
inputSchema: { value: z.string() },
async handler(input: unknown) {
const parsed = z.object({ value: z.string() }).parse(input);
return { content: [{ type: "text", text: parsed.value }] };
const parsed = z
.object({ value: z.string().min(2, "value must be at least 2 characters") })
.strict()
.safeParse(input);
if (!parsed.success) throw new PaseoToolInputError(parsed.error.message, parsed.error.issues);
return { content: [{ type: "text", text: parsed.data.value }] };
},
};
const tools = new Map([[tool.name, tool]]);
@@ -193,6 +198,13 @@ describe("OpenCodeBridge", () => {
expect(asked).toEqual([
{ permission: "paseo_echo_context", patterns: ["*"], always: ["*"], metadata: {} },
]);
asked.length = 0;
await expect(
hooks.tool.paseo_echo_context.execute(
{ value: "x" },
{ sessionID: "ses_one", ask: async () => undefined },
),
).rejects.toThrow("value: value must be at least 2 characters");
executedInputs.length = 0;
await expect(
@@ -357,12 +369,17 @@ describe("OpenCodeBridge", () => {
body: JSON.stringify(input),
},
);
expect(response.status).toBe(500);
expect(response.status).toBe(400);
const payload = await response.json();
expect(payload.error).toContain("provider must be provider/model");
expect(payload).toMatchObject({ error: "Invalid tool input", issues: expect.any(Array) });
await expect(
tools.get("paseo_create_agent")!.execute(input, { sessionID: "caller" }),
).rejects.toMatchObject({ message: payload.error, status: 500 });
).rejects.toMatchObject({
status: 400,
message: expect.stringContaining(
"Invalid tool input: provider: provider must be provider/model, for example codex/gpt-5.4",
),
});
} finally {
await dispose();
}
@@ -550,6 +567,104 @@ describe("OpenCodeBridge", () => {
}
});
test("returns actionable 4xx responses for invalid bridge requests", async () => {
const paseoHome = await mkdtemp(path.join(tmpdir(), "paseo-opencode-bridge-errors-"));
temporaryDirectories.push(paseoHome);
const bridge = new OpenCodeBridge({ paseoHome, logger: createTestLogger() });
await bridge.start();
const catalog = createCatalog();
bridge.bindSession({ sessionId: "ses_errors", env: {}, tools: catalog });
const expectedErrorCatalog: PaseoToolCatalog = {
tools: new Map([
[
"expected_error",
{
name: "expected_error",
description: "Throws an expected tool error.",
inputSchema: {},
async handler() {
throw new PaseoToolInputError("Agent not found: missing-agent");
},
},
],
[
"internal_error",
{
name: "internal_error",
description: "Throws an internal error.",
inputSchema: {},
async handler() {
throw new Error("Provider unavailable");
},
},
],
]),
getTool(name) {
return this.tools.get(name);
},
async executeTool(name, input, context) {
const tool = this.tools.get(name);
if (!tool) throw new Error(`Unknown tool: ${name}`);
return tool.handler(input, context ?? {});
},
};
bridge.bindSession({ sessionId: "ses_expected_error", env: {}, tools: expectedErrorCatalog });
try {
const plugin = readPluginOptions(bridge.decorateServerEnv({}));
const headers = {
Authorization: `Bearer ${plugin.token}`,
"Content-Type": "application/json",
};
const invalidInput = await fetch(
`${plugin.baseUrl}/_internal/opencode/sessions/ses_errors/tools/echo_context`,
{ method: "POST", headers, body: JSON.stringify({ value: "ok", extra: true }) },
);
expect(invalidInput.status).toBe(400);
expect(await invalidInput.json()).toEqual({
error: "Invalid tool input",
issues: [
{
code: "unrecognized_keys",
keys: ["extra"],
path: [],
message: 'Unrecognized key: "extra"',
},
],
});
const unknownTool = await fetch(
`${plugin.baseUrl}/_internal/opencode/sessions/ses_errors/tools/missing`,
{ method: "POST", headers, body: "{}" },
);
expect(unknownTool.status).toBe(404);
expect(await unknownTool.json()).toEqual({ error: "Paseo tool not found: missing" });
const malformedBody = await fetch(
`${plugin.baseUrl}/_internal/opencode/sessions/ses_errors/tools/echo_context`,
{ method: "POST", headers, body: "{" },
);
expect(malformedBody.status).toBe(400);
expect(await malformedBody.json()).toEqual({ error: "Malformed JSON request body" });
const expectedError = await fetch(
`${plugin.baseUrl}/_internal/opencode/sessions/ses_expected_error/tools/expected_error`,
{ method: "POST", headers, body: "{}" },
);
expect(expectedError.status).toBe(400);
expect(await expectedError.json()).toEqual({ error: "Agent not found: missing-agent" });
const internalError = await fetch(
`${plugin.baseUrl}/_internal/opencode/sessions/ses_expected_error/tools/internal_error`,
{ method: "POST", headers, body: "{}" },
);
expect(internalError.status).toBe(500);
expect(await internalError.json()).toEqual({ error: "Provider unavailable" });
} finally {
await bridge.close();
}
});
test("preserves user OpenCode config while installing one content-addressed plugin", async () => {
const paseoHome = await mkdtemp(path.join(tmpdir(), "paseo-opencode-bridge-config-"));
temporaryDirectories.push(paseoHome);
@@ -11,11 +11,22 @@ import {
addModelVisibleStructuredContent,
serializePaseoToolInputParameters,
} from "../../tools/paseo-tool-serialization.js";
import type { PaseoToolCatalog } from "../../tools/types.js";
import { PaseoToolInputError, type PaseoToolCatalog } from "../../tools/types.js";
const INTERNAL_PREFIX = "/_internal/opencode";
const MAX_REQUEST_BYTES = 1024 * 1024;
class OpenCodeBridgeClientError extends Error {
constructor(
readonly status: number,
message: string,
readonly details?: unknown,
) {
super(message);
this.name = "OpenCodeBridgeClientError";
}
}
interface OpenCodeBridgeOptions {
paseoHome: string;
logger: Logger;
@@ -187,10 +198,26 @@ export class OpenCodeBridge {
sendJson(response, 404, { error: "Not found" });
} catch (error) {
this.logger.warn({ err: error }, "OpenCode bridge request failed");
const failure = error instanceof Error ? error : new Error(String(error));
if (!response.headersSent) {
sendJson(response, 500, { error: error instanceof Error ? error.message : String(error) });
if (error instanceof OpenCodeBridgeClientError) {
sendJson(response, error.status, {
error: error.message,
...(error.details === undefined ? {} : { details: error.details }),
});
} else if (error instanceof PaseoToolInputError) {
sendJson(
response,
error.status,
error.issues
? { error: "Invalid tool input", issues: error.issues }
: { error: error.message },
);
} else {
sendJson(response, 500, { error: failure.message });
}
} else {
response.destroy(error instanceof Error ? error : new Error(String(error)));
response.destroy(failure);
}
}
}
@@ -224,6 +251,9 @@ export class OpenCodeBridge {
sendJson(input.response, 403, { error: "Paseo tools are disabled for this session" });
return;
}
if (!binding.tools.getTool(input.toolName)) {
throw new OpenCodeBridgeClientError(404, `Paseo tool not found: ${input.toolName}`);
}
const body = await readJsonBody(input.request);
const controller = new AbortController();
input.request.once("aborted", () =>
@@ -349,11 +379,17 @@ async function readJsonBody(request: IncomingMessage): Promise<unknown> {
for await (const chunk of request) {
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
size += buffer.byteLength;
if (size > MAX_REQUEST_BYTES) throw new Error("OpenCode bridge request body is too large");
if (size > MAX_REQUEST_BYTES) {
throw new OpenCodeBridgeClientError(413, "OpenCode bridge request body is too large");
}
chunks.push(buffer);
}
if (chunks.length === 0) return {};
return JSON.parse(Buffer.concat(chunks).toString("utf8")) as unknown;
try {
return JSON.parse(Buffer.concat(chunks).toString("utf8")) as unknown;
} catch {
throw new OpenCodeBridgeClientError(400, "Malformed JSON request body");
}
}
function sendJson(response: ServerResponse, status: number, body: unknown): void {
@@ -15,12 +15,14 @@ export default {
if (allowMissing && response.status === 404) return null;
if (!response.ok) {
const payload = await response.json().catch(() => null);
const detail = formatErrorDetail(payload?.issues ?? payload?.details);
const message =
typeof payload?.error === "string" && payload.error
? payload.error
: `Paseo tool bridge returned HTTP ${response.status}`;
const error = new Error(message);
const error = new Error(detail ? `${message}: ${detail}` : message);
error.status = response.status;
if (detail) error.details = detail;
throw error;
}
return response.json();
@@ -78,6 +80,17 @@ export default {
},
};
function formatErrorDetail(details) {
if (!Array.isArray(details)) return typeof details === "string" ? details : "";
return details
.map((issue) => {
const path =
Array.isArray(issue?.path) && issue.path.length > 0 ? issue.path.join(".") : "input";
return `${path}: ${issue?.message ?? "Invalid value"}`;
})
.join("; ");
}
// OpenCode tool content is text or a file; MCP images arrive as base64 data.
function toOpenCodeContent(part) {
if (part.type === "image") {
@@ -92,7 +92,7 @@ test("a tool's image output is shown as an image, not as base64 in the tool outp
},
{
type: "assistant_message",
text: expect.stringMatching(/^!\[Image\]\(file:\/\/.*paseo-attachments.*\.png\)$/),
text: expect.stringMatching(/^\n\n!\[Image\]\(file:\/\/.*paseo-attachments.*\.png\)\n\n$/),
},
]);
expect(timeline.messages([screenshot])).toEqual([]);
@@ -1,5 +1,7 @@
import { existsSync, rmSync } from "node:fs";
import path from "node:path";
import MarkdownIt from "markdown-it";
import { projectTimelineRows } from "../timeline-projection.js";
import { describe, expect, test } from "vitest";
import {
@@ -44,7 +46,7 @@ describe("isProviderImageMarkdown", () => {
);
expect(markdown).toBe(
`![Image](file:///C:/Users/me/AppData/Local/Temp/paseo-attachments/${HASH}.png)`,
`\n\n![Image](file:///C:/Users/me/AppData/Local/Temp/paseo-attachments/${HASH}.png)\n\n`,
);
expect(isProviderImageMarkdown(markdown)).toBe(true);
});
@@ -53,20 +55,20 @@ describe("isProviderImageMarkdown", () => {
const markdown = renderImageMarkdown("/home/user/Projects/Project With Spaces/screenshot.png");
expect(markdown).toBe(
"![Image](file:///home/user/Projects/Project%20With%20Spaces/screenshot.png)",
"\n\n![Image](file:///home/user/Projects/Project%20With%20Spaces/screenshot.png)\n\n",
);
});
test("encodes URI-significant characters in POSIX file paths", () => {
const markdown = renderImageMarkdown("/tmp/screenshot#1?draft.png");
expect(markdown).toBe("![Image](file:///tmp/screenshot%231%3Fdraft.png)");
expect(markdown).toBe("\n\n![Image](file:///tmp/screenshot%231%3Fdraft.png)\n\n");
});
test("preserves double-leading slashes in POSIX file paths", () => {
const markdown = renderImageMarkdown("//tmp/screenshot#1.png");
expect(markdown).toBe("![Image](file:////tmp/screenshot%231.png)");
expect(markdown).toBe("\n\n![Image](file:////tmp/screenshot%231.png)\n\n");
});
test.each([
@@ -77,7 +79,7 @@ describe("isProviderImageMarkdown", () => {
"file://server/share/shot%3Fdraft.png",
],
])("encodes %s image paths as file URIs", (_label, imagePath, expectedSource) => {
expect(renderImageMarkdown(imagePath)).toBe(`![Image](${expectedSource})`);
expect(renderImageMarkdown(imagePath)).toBe(`\n\n![Image](${expectedSource})\n\n`);
});
test("rejects user-authored markdown that is not a materialized attachment", () => {
@@ -113,3 +115,52 @@ describe("materializeProviderImage", () => {
}
});
});
describe("provider image paragraph boundaries", () => {
test.each(["", "\n"])("keeps image outputs separate from prose ending in %j", (ending) => {
const image = renderProviderImageOutputAsAssistantMarkdown({
url: "https://example.com/chart.png",
});
if (!image) throw new Error("Expected image output");
const projected = projectTimelineRows({
mode: "projected",
rows: [
{
seq: 1,
timestamp: "2026-10-16T15:08:23Z",
item: { type: "assistant_message", text: `Before${ending}`, messageId: "commentary" },
},
{ seq: 2, timestamp: "2026-10-16T15:08:24Z", item: image },
{
seq: 3,
timestamp: "2026-10-16T15:08:25Z",
item: { type: "assistant_message", text: "After" },
},
],
});
expect(projected).toHaveLength(1);
const item = projected[0]!.item;
if (item.type !== "assistant_message") throw new Error("Expected assistant message");
const paragraphs = new MarkdownIt()
.parse(item.text, {})
.filter((token) => token.type === "inline");
expect(paragraphs.map((token) => token.content)).toEqual([
"Before",
"![Image](https://example.com/chart.png)",
"After",
]);
});
});
test("materialized image output retains paragraph boundaries and attachment recognition", () => {
const item = renderProviderImageOutputAsAssistantMarkdown(
{ data: "YWJjMTIz", mimeType: "image/png" },
{ materialize: () => ({ path: `/tmp/paseo-attachments/${HASH}.png` }) },
);
expect(item).toEqual({
type: "assistant_message",
text: `\n\n![Image](file:///tmp/paseo-attachments/${HASH}.png)\n\n`,
});
if (!item || item.type !== "assistant_message") throw new Error("Expected image output");
expect(isProviderImageMarkdown(item.text)).toBe(true);
});
@@ -108,7 +108,7 @@ const PROVIDER_IMAGE_MARKDOWN = new RegExp(
);
export function isProviderImageMarkdown(text: string): boolean {
return PROVIDER_IMAGE_MARKDOWN.test(text);
return PROVIDER_IMAGE_MARKDOWN.test(text.trimStart());
}
interface RenderProviderImageOutputOptions {
@@ -178,7 +178,8 @@ export function renderProviderImageOutputAsAssistantMarkdown(
const altText = escapeMarkdownImageAlt(nonEmptyString(image.altText) ?? "Image");
return {
type: "assistant_message",
text: `![${altText}](${escapeMarkdownImageSource(source)})`,
// Image outputs remain standalone paragraphs when adjacent assistant chunks merge.
text: `\n\n![${altText}](${escapeMarkdownImageSource(source)})\n\n`,
};
}
@@ -208,6 +209,6 @@ export function renderProviderImageOutputAsAssistantMarkdown(
const altText = escapeMarkdownImageAlt(nonEmptyString(image.altText) ?? "Image");
return {
type: "assistant_message",
text: `![${altText}](${escapeMarkdownImageSource(materialized.path)})`,
text: `\n\n![${altText}](${escapeMarkdownImageSource(materialized.path)})\n\n`,
};
}
@@ -25,7 +25,7 @@ import {
import { curateAgentActivity } from "../activity-curator.js";
import { selectItemsByProjectedLimit } from "../timeline-projection.js";
import type { AgentStorage } from "../agent-storage.js";
import { ensureAgentLoaded } from "../agent-loading.js";
import { AgentNotFoundError, ensureAgentLoaded } from "../agent-loading.js";
import { isStoredAgentProviderAvailable } from "../../persistence-hooks.js";
import {
archiveByScope,
@@ -40,7 +40,12 @@ import { everyMsToFiveFieldCron } from "@getpaseo/protocol/schedule/cadence";
import { expandUserPath, isSameOrDescendantPath, resolvePathFromBase } from "../../path-utils.js";
import type { TerminalManager } from "../../../terminal/terminal-manager.js";
import type { CreatePaseoWorktreeWorkflowFn } from "../../worktree-session.js";
import type { ScheduleService } from "../../schedule/service.js";
import { computeNextRunAt, validateScheduleCadence } from "../../schedule/cron.js";
import {
ScheduleNotFoundError,
ScheduleStateConflictError,
type ScheduleService,
} from "../../schedule/service.js";
import {
ScheduleRunSchema,
ScheduleSummarySchema,
@@ -91,12 +96,13 @@ import {
} from "../../worktree/commands.js";
import { registerBrowserTools } from "../../browser-tools/tools.js";
import type { BrowserToolsBroker } from "../../browser-tools/broker.js";
import type {
PaseoToolCatalog,
PaseoToolConfig,
PaseoToolDefinition,
PaseoToolExecutionContext,
PaseoToolResult,
import {
PaseoToolInputError,
type PaseoToolCatalog,
type PaseoToolConfig,
type PaseoToolDefinition,
type PaseoToolExecutionContext,
type PaseoToolResult,
} from "./types.js";
import type { ProviderPaseoToolsPolicy } from "@getpaseo/protocol/provider-config";
import { isPaseoToolEnabled } from "../paseo-tool-policy.js";
@@ -229,7 +235,7 @@ function assertOptionsAbsent(
message: string,
): void {
if (options.some(([, value]) => value !== undefined)) {
throw new Error(message);
throw new PaseoToolInputError(message);
}
}
@@ -265,7 +271,7 @@ function resolveWorkspaceWorktreeTarget(input: WorkspaceWorktreeOptions): Worksp
};
case "checkout-branch":
if (!input.branch) {
throw new Error("branch is required for checkout-branch mode");
throw new PaseoToolInputError("branch is required for checkout-branch mode");
}
assertOptionsAbsent(
[
@@ -279,7 +285,7 @@ function resolveWorkspaceWorktreeTarget(input: WorkspaceWorktreeOptions): Worksp
return { action: "checkout", refName: input.branch };
case "checkout-pr":
if (input.prNumber === undefined) {
throw new Error("prNumber is required for checkout-pr mode");
throw new PaseoToolInputError("prNumber is required for checkout-pr mode");
}
assertOptionsAbsent(
[
@@ -355,7 +361,7 @@ function resolveScheduleProviderAndModel(params: {
const provider = providerInput.slice(0, slashIndex).trim();
const model = providerInput.slice(slashIndex + 1).trim();
if (!provider || !model) {
throw new Error("provider must be <provider> or <provider>/<model>");
throw new PaseoToolInputError("provider must be <provider> or <provider>/<model>");
}
return {
@@ -372,7 +378,7 @@ function resolveScheduleUpdateProviderAndModel(params: {
const modelInput = typeof params.model === "string" ? params.model.trim() : params.model;
if (params.model !== undefined && modelInput === "") {
throw new Error("model cannot be empty");
throw new PaseoToolInputError("model cannot be empty");
}
if (!providerInput) {
@@ -390,13 +396,13 @@ function resolveScheduleUpdateProviderAndModel(params: {
const provider = providerInput.slice(0, slashIndex).trim();
const modelFromProvider = providerInput.slice(slashIndex + 1).trim();
if (!provider || !modelFromProvider) {
throw new Error("provider must be <provider> or <provider>/<model>");
throw new PaseoToolInputError("provider must be <provider> or <provider>/<model>");
}
if (params.model === null) {
throw new Error("provider specifies a model but model is null");
throw new PaseoToolInputError("provider specifies a model but model is null");
}
if (typeof modelInput === "string" && modelInput !== modelFromProvider) {
throw new Error("Conflicting model values provided");
throw new PaseoToolInputError("Conflicting model values provided");
}
return {
@@ -438,26 +444,82 @@ function normalizeScheduleTimeZoneArg(value: string | undefined): string | undef
return normalizeScheduleCadenceArg(value);
}
function addScheduleCadenceIssue(
input: { cron?: string; timezone?: string },
context: z.RefinementCtx,
): void {
const expression = input.cron?.trim();
if (!expression) {
return;
}
const cadence: ScheduleCadence = {
type: "cron",
expression,
...(input.timezone !== undefined ? { timezone: input.timezone.trim() } : {}),
};
try {
validateScheduleCadence(cadence);
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
context.addIssue({
code: "custom",
path: [message.startsWith("Invalid cron time zone:") ? "timezone" : "cron"],
message,
});
return;
}
try {
computeNextRunAt(cadence, new Date());
} catch (error) {
context.addIssue({
code: "custom",
path: ["cron"],
message: error instanceof Error ? error.message : String(error),
});
}
}
function parseToolArguments<T>(schema: z.ZodType<T>, input: unknown): T {
const result = schema.safeParse(input);
if (!result.success) {
throw new PaseoToolInputError(result.error.message, result.error.issues);
}
return result.data;
}
function addScheduleDurationIssue(value: string, context: z.RefinementCtx): void {
try {
parseDurationString(value);
} catch (error) {
context.addIssue({
code: "custom",
message: error instanceof Error ? error.message : String(error),
});
}
}
const scheduleDurationInputSchema = z.string().superRefine(addScheduleDurationIssue);
function resolveScheduleUpdateCadence(input: ScheduleUpdateToolInput): ScheduleCadence | undefined {
const every = normalizeScheduleCadenceArg(input.every);
const cron = normalizeScheduleCadenceArg(input.cron);
const timeZone = normalizeScheduleTimeZoneArg(input.timezone);
if (every !== undefined && cron !== undefined) {
throw new Error("Specify at most one of every or cron");
throw new PaseoToolInputError("Specify at most one of every or cron");
}
if (timeZone !== undefined && cron === undefined) {
throw new Error("timezone can only be used with cron");
throw new PaseoToolInputError("timezone can only be used with cron");
}
if (every !== undefined) {
// COMPAT(scheduleEveryInput): accept the old hidden field and canonicalize it before write.
// Added in v0.2.0; remove after 2027-01-17.
const everyMs = parseDurationString(every);
const everyMs = parseDurationString(parseToolArguments(scheduleDurationInputSchema, every));
const expression = everyMsToFiveFieldCron(everyMs);
if (expression) {
return { type: "cron", expression };
}
throw new Error(`${every} cannot be represented faithfully by five-field cron`);
throw new PaseoToolInputError(`${every} cannot be represented faithfully by five-field cron`);
}
if (cron !== undefined) {
return {
@@ -471,7 +533,7 @@ function resolveScheduleUpdateCadence(input: ScheduleUpdateToolInput): ScheduleC
function resolveScheduleUpdateExpiresAt(input: ScheduleUpdateToolInput): string | null | undefined {
if (input.expiresIn !== undefined && input.clearExpires) {
throw new Error("Specify at most one of expiresIn or clearExpires");
throw new PaseoToolInputError("Specify at most one of expiresIn or clearExpires");
}
if (input.expiresIn !== undefined) {
return new Date(Date.now() + parseDurationString(input.expiresIn)).toISOString();
@@ -593,7 +655,11 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
typeof (inputSchema as { safeParseAsync?: unknown }).safeParseAsync === "function"
? (inputSchema as z.ZodType)
: z.object(inputSchema as z.ZodRawShape).passthrough();
return schema.parseAsync(input);
const result = await schema.safeParseAsync(input);
if (!result.success) {
throw new PaseoToolInputError(result.error.message, result.error.issues);
}
return result.data;
};
const tools = new Map<string, PaseoToolDefinition>();
@@ -627,7 +693,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
): Promise<PaseoToolResult> {
const tool = tools.get(name);
if (!tool) {
throw new Error(`Paseo tool not found: ${name}`);
throw new PaseoToolInputError(`Paseo tool not found: ${name}`);
}
return tool.handler(await parseToolInput(tool, input), context);
},
@@ -639,7 +705,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
}): ScheduleCadence => {
const expression = input.cron?.trim() ?? "";
if (!expression) {
throw new Error("cron is required");
throw new PaseoToolInputError("cron is required");
}
const timezone = normalizeScheduleTimeZoneArg(input.timezone);
return {
@@ -661,7 +727,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
}
const parentAgent = agentManager.getAgent(callerAgentId);
if (!parentAgent) {
throw new Error(`Parent agent ${callerAgentId} not found`);
throw new PaseoToolInputError(`Parent agent ${callerAgentId} not found`);
}
return parentAgent;
};
@@ -690,9 +756,9 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
const trimmedCwd = requestedCwd?.trim();
if (!trimmedCwd) {
if (opts?.required) {
throw new Error("cwd is required");
throw new PaseoToolInputError("cwd is required");
}
throw new Error("cwd is required outside an agent-scoped session");
throw new PaseoToolInputError("cwd is required outside an agent-scoped session");
}
return expandUserPath(trimmedCwd);
@@ -707,11 +773,12 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
}
if (!options.ensureWorkspaceForCreate) {
throw new Error(
callerAgentId
? `Caller agent ${callerAgentId} has no workspace and workspace minting is not configured`
: "workspaceId is required outside an agent-scoped session",
);
if (callerAgentId) {
throw new Error(
`Caller agent ${callerAgentId} has no workspace and workspace minting is not configured`,
);
}
throw new PaseoToolInputError("workspaceId is required outside an agent-scoped session");
}
return options.ensureWorkspaceForCreate(resolvedCwd);
@@ -726,11 +793,11 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
if (callerAgentId) {
const callerAgent = resolveCallerAgent();
if (!callerAgent?.workspaceId) {
throw new Error(`Caller agent ${callerAgentId} has no current workspace`);
throw new PaseoToolInputError(`Caller agent ${callerAgentId} has no current workspace`);
}
return callerAgent.workspaceId;
}
throw new Error("workspaceId is required outside an agent-scoped session");
throw new PaseoToolInputError("workspaceId is required outside an agent-scoped session");
}
const buildCallerAgentScheduleConfigExtras = (
@@ -800,7 +867,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
}
if (!params?.provider?.trim()) {
throw new Error("provider is required when target is new-agent");
throw new PaseoToolInputError("provider is required when target is new-agent");
}
const resolvedProviderModel = resolveScheduleProviderAndModel({
@@ -822,22 +889,41 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
if (!scheduleService) {
throw new Error("Schedule service is not configured");
}
const schedule = await scheduleService.inspect(id);
let schedule: Awaited<ReturnType<ScheduleService["inspect"]>>;
try {
schedule = await scheduleService.inspect(id);
} catch (error) {
if (error instanceof ScheduleNotFoundError) {
throw new PaseoToolInputError(error.message);
}
throw error;
}
if (schedule.target.type !== type) {
throw new Error(
throw new PaseoToolInputError(
type === "agent" ? `Heartbeat not found: ${id}` : `Schedule not found: ${id}`,
);
}
return schedule;
}
async function runScheduleAction<T>(action: () => Promise<T>): Promise<T> {
try {
return await action();
} catch (error) {
if (error instanceof ScheduleStateConflictError) {
throw new PaseoToolInputError(error.message);
}
throw error;
}
}
async function requireCallerHeartbeat(id: string) {
if (!callerAgentId) {
throw new Error("Heartbeat operations require an agent-scoped session");
throw new PaseoToolInputError("Heartbeat operations require an agent-scoped session");
}
const schedule = await requireScheduleTarget(id, "agent");
if (schedule.target.type !== "agent" || schedule.target.agentId !== callerAgentId) {
throw new Error(`Heartbeat ${id} does not belong to caller ${callerAgentId}`);
throw new PaseoToolInputError(`Heartbeat ${id} does not belong to caller ${callerAgentId}`);
}
return schedule;
}
@@ -847,6 +933,9 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
})
.refine(
(value) => {
if (!value.includes("/")) {
return true;
}
try {
resolveRequiredProviderModel(value);
return true;
@@ -1184,11 +1273,13 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
},
async (args, context) => {
if (!callerAgentId) {
throw new Error("speak is only available to agent-scoped tool sessions");
throw new PaseoToolInputError("speak is only available to agent-scoped tool sessions");
}
const handler = resolveSpeakHandler?.(callerAgentId) ?? null;
if (!handler) {
throw new Error(`No speak handler registered for your session '${callerAgentId}'`);
throw new PaseoToolInputError(
`No speak handler registered for your session '${callerAgentId}'`,
);
}
await handler({
text: args.text,
@@ -1289,7 +1380,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
if (isolation === "local") {
const cwd = resolveScopedCwd(path, { required: true });
if (!(await isExistingDirectory(cwd))) {
throw new Error(`Directory not found: ${cwd}`);
throw new PaseoToolInputError(`Directory not found: ${cwd}`);
}
assertOptionsAbsent(
[
@@ -1534,7 +1625,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
if (hasLegacyCreateAgentPlacement(args)) {
// COMPAT(nestedCreateAgentPlacement): accept the old relationship/workspace shape without
// advertising it to models. Added in v0.2.0; remove after 2027-01-17.
const parsed = legacyAgentToAgentCreateAgentArgsSchema.parse(args);
const parsed = parseToolArguments(legacyAgentToAgentCreateAgentArgsSchema, args);
const { cwd, workspaceId, worktree } = await resolveCreateAgentWorkspace(parsed.workspace, {
prompt: parsed.initialPrompt,
});
@@ -1547,7 +1638,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
worktree,
};
}
const parsed = agentToAgentCreateAgentArgsSchema.parse(args);
const parsed = parseToolArguments(agentToAgentCreateAgentArgsSchema, args);
const { cwd, workspaceId } = await resolveCanonicalCreateAgentWorkspace(parsed.workspaceId, {
prompt: parsed.initialPrompt,
});
@@ -1563,13 +1654,15 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
if (hasLegacyCreateAgentPlacement(args)) {
// COMPAT(nestedCreateAgentPlacement): see the agent-scoped branch above.
const parsedArgs = normalizeTopLevelCreateAgentArgs(
legacyTopLevelCreateAgentArgsSchema.parse(args),
parseToolArguments(legacyTopLevelCreateAgentArgsSchema, args),
);
if (parsedArgs.relationship?.kind === "subagent") {
throw new Error("relationship subagent requires an agent-scoped tool session");
throw new PaseoToolInputError(
"relationship subagent requires an agent-scoped tool session",
);
}
if (!parsedArgs.workspace) {
throw new Error("Legacy create_agent placement could not be resolved");
throw new PaseoToolInputError("Legacy create_agent placement could not be resolved");
}
const { cwd, workspaceId, worktree } = await resolveCreateAgentWorkspace(
parsedArgs.workspace,
@@ -1584,7 +1677,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
worktree,
};
}
const parsedArgs = canonicalTopLevelCreateAgentArgsSchema.parse(args);
const parsedArgs = parseToolArguments(canonicalTopLevelCreateAgentArgsSchema, args);
const { cwd, workspaceId } = await resolveCanonicalCreateAgentWorkspace(
parsedArgs.workspaceId,
{ prompt: parsedArgs.initialPrompt },
@@ -1644,7 +1737,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
}
const caller = resolveCallerAgent();
if (!caller?.workspaceId) {
throw new Error(`Caller agent ${callerAgentId} has no current workspace`);
throw new PaseoToolInputError(`Caller agent ${callerAgentId} has no current workspace`);
}
return { cwd: undefined, workspaceId: caller.workspaceId };
}
@@ -1672,18 +1765,18 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
};
if (canonicalCandidate.relationship && canonicalCandidate.workspace) {
return legacyTopLevelCreateAgentArgsSchema.parse({
return parseToolArguments(legacyTopLevelCreateAgentArgsSchema, {
...canonicalCandidate,
...(Object.keys(settings).length > 0 ? { settings } : {}),
});
}
if (canonicalCandidate.relationship || canonicalCandidate.workspace) {
throw new Error("relationship and workspace must be provided together");
throw new PaseoToolInputError("relationship and workspace must be provided together");
}
if (!cwd?.trim()) {
throw new Error("cwd is required for legacy top-level create_agent calls");
throw new PaseoToolInputError("cwd is required for legacy top-level create_agent calls");
}
const legacyWorktreeTarget = resolveLegacyCreateAgentWorktreeTarget({
@@ -1710,7 +1803,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
},
};
return legacyTopLevelCreateAgentArgsSchema.parse({
return parseToolArguments(legacyTopLevelCreateAgentArgsSchema, {
...canonicalCandidate,
relationship: { kind: "detached" },
workspace,
@@ -1763,11 +1856,11 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
}> {
if (workspace.kind === "current") {
if (!callerAgentId) {
throw new Error("workspace current requires an agent-scoped tool session");
throw new PaseoToolInputError("workspace current requires an agent-scoped tool session");
}
const callerAgent = resolveCallerAgent();
if (!callerAgent?.workspaceId) {
throw new Error(`Caller agent ${callerAgentId} has no current workspace`);
throw new PaseoToolInputError(`Caller agent ${callerAgentId} has no current workspace`);
}
return {
cwd: workspace.cwd,
@@ -1784,14 +1877,16 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
(candidate) => candidate.workspaceId === workspace.workspaceId,
);
if (!existingWorkspace) {
throw new Error(`Workspace ${workspace.workspaceId} not found`);
throw new PaseoToolInputError(`Workspace ${workspace.workspaceId} not found`);
}
const cwd = workspace.cwd
? resolveScopedCwd(workspace.cwd, { required: true })
: existingWorkspace.cwd;
const lockedCwd = callerContext?.lockedCwd?.trim();
if (lockedCwd && !isSameOrDescendantPath(expandUserPath(lockedCwd), cwd)) {
throw new Error(`Workspace ${workspace.workspaceId} is outside the allowed cwd`);
throw new PaseoToolInputError(
`Workspace ${workspace.workspaceId} is outside the allowed cwd`,
);
}
return {
cwd,
@@ -1893,6 +1988,9 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
logger: childLogger,
}).catch((error) => {
stopNotification?.();
if (error instanceof AgentNotFoundError) {
throw new PaseoToolInputError(error.message);
}
throw error;
});
@@ -1956,7 +2054,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
const record = await agentStorage.get(agentId);
if (!record || record.internal) {
throw new Error(`Agent ${agentId} not found`);
throw new PaseoToolInputError(`Agent ${agentId} not found`);
}
const structuredSnapshot = buildStoredAgentPayload(
@@ -2194,10 +2292,10 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
const workspaceId = resolveWorkspaceIdForRename(requestedWorkspaceId);
const existing = await options.workspaceRegistry.get(workspaceId);
if (!existing) {
throw new Error(`Workspace ${workspaceId} not found`);
throw new PaseoToolInputError(`Workspace ${workspaceId} not found`);
}
if (existing.archivedAt) {
throw new Error(`Workspace ${workspaceId} is archived`);
throw new PaseoToolInputError(`Workspace ${workspaceId} is archived`);
}
await options.workspaceRegistry.upsert({
@@ -2401,7 +2499,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
const terminal = terminalManager.getTerminal(terminalId);
if (!terminal) {
throw new Error(`Terminal ${terminalId} not found`);
throw new PaseoToolInputError(`Terminal ${terminalId} not found`);
}
terminal.kill();
@@ -2437,7 +2535,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
}
if (!terminalManager.getTerminal(terminalId)) {
throw new Error(`Terminal ${terminalId} not found`);
throw new PaseoToolInputError(`Terminal ${terminalId} not found`);
}
const capture = await terminalManager.captureTerminal(terminalId, {
@@ -2478,7 +2576,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
const terminal = terminalManager.getTerminal(terminalId);
if (!terminal) {
throw new Error(`Terminal ${terminalId} not found`);
throw new PaseoToolInputError(`Terminal ${terminalId} not found`);
}
terminal.send({
@@ -2498,24 +2596,26 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
{
title: "Create schedule",
description: "Create a recurring schedule that starts a new agent on a cron cadence.",
inputSchema: {
prompt: z.string().trim().min(1, "prompt is required"),
cron: z.string().trim().min(1, "cron is required"),
timezone: z
.string()
.trim()
.min(1)
.optional()
.describe("IANA time zone for the cron cadence. For example: America/New_York."),
name: z.string().optional(),
provider: (callerAgentId ? AgentProviderEnum.optional() : AgentProviderEnum).describe(
"Provider, or provider/model (for example: codex or codex/gpt-5.4). Defaults to the caller's provider in an agent-scoped session.",
),
cwd: z.string().optional(),
isolation: z.enum(["local", "worktree"]).optional(),
maxRuns: z.number().int().positive().optional(),
expiresIn: z.string().optional(),
},
inputSchema: z
.object({
prompt: z.string().trim().min(1, "prompt is required"),
cron: z.string().trim().min(1, "cron is required"),
timezone: z
.string()
.trim()
.min(1)
.optional()
.describe("IANA time zone for the cron cadence. For example: America/New_York."),
name: z.string().optional(),
provider: (callerAgentId ? AgentProviderEnum.optional() : AgentProviderEnum).describe(
"Provider, or provider/model (for example: codex or codex/gpt-5.4). Defaults to the caller's provider in an agent-scoped session.",
),
cwd: z.string().optional(),
isolation: z.enum(["local", "worktree"]).optional(),
maxRuns: z.number().int().positive().optional(),
expiresIn: scheduleDurationInputSchema.optional(),
})
.superRefine(addScheduleCadenceIssue),
outputSchema: ScheduleSummarySchema.shape,
},
async ({ prompt, cron, timezone, name, provider, cwd, isolation, maxRuns, expiresIn }) => {
@@ -2548,19 +2648,21 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
{
title: "Create heartbeat",
description: "Create a recurring heartbeat that sends you a prompt on a cron cadence.",
inputSchema: {
prompt: z.string().trim().min(1, "prompt is required"),
cron: z.string().trim().min(1, "cron is required"),
timezone: z
.string()
.trim()
.min(1)
.optional()
.describe("IANA time zone for the cron cadence. For example: America/New_York."),
name: z.string().optional(),
maxRuns: z.number().int().positive().optional(),
expiresIn: z.string().optional(),
},
inputSchema: z
.object({
prompt: z.string().trim().min(1, "prompt is required"),
cron: z.string().trim().min(1, "cron is required"),
timezone: z
.string()
.trim()
.min(1)
.optional()
.describe("IANA time zone for the cron cadence. For example: America/New_York."),
name: z.string().optional(),
maxRuns: z.number().int().positive().optional(),
expiresIn: scheduleDurationInputSchema.optional(),
})
.superRefine(addScheduleCadenceIssue),
outputSchema: ScheduleSummarySchema.shape,
},
async ({ prompt, cron, timezone, name, maxRuns, expiresIn }) => {
@@ -2568,7 +2670,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
throw new Error("Schedule service is not configured");
}
if (!callerAgentId) {
throw new Error("create_heartbeat requires an agent-scoped session");
throw new PaseoToolInputError("create_heartbeat requires an agent-scoped session");
}
resolveCallerAgent();
@@ -2679,7 +2781,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
}
await requireScheduleTarget(id, "new-agent");
await scheduleService.pause(id);
await runScheduleAction(() => scheduleService.pause(id));
return {
content: [],
structuredContent: ensureValidJson({ success: true }),
@@ -2705,7 +2807,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
}
await requireScheduleTarget(id, "new-agent");
await scheduleService.resume(id);
await runScheduleAction(() => scheduleService.resume(id));
return {
content: [],
structuredContent: ensureValidJson({ success: true }),
@@ -2787,13 +2889,13 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
.optional()
.describe("New mode for new-agent target (null to clear)."),
cwd: z.string().trim().min(1).optional().describe("New cwd for new-agent target."),
expiresIn: z
.string()
expiresIn: scheduleDurationInputSchema
.optional()
.describe("New relative expiry duration (for example: 1h, 2d)."),
clearExpires: z.boolean().optional().describe("Clear any schedule expiry."),
})
.passthrough(),
.passthrough()
.superRefine(addScheduleCadenceIssue),
outputSchema: StoredScheduleSchema.shape,
},
async (input) => {
@@ -2850,7 +2952,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
throw new Error("Schedule service is not configured");
}
await requireScheduleTarget(id, "new-agent");
const schedule = await scheduleService.runOnce(id);
const schedule = await runScheduleAction(() => scheduleService.runOnce(id));
return {
content: [],
structuredContent: ensureValidJson(schedule),
@@ -2963,10 +3065,10 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
});
const summary = toProviderSummary(entry);
if (!entry.enabled) {
throw new Error(`Provider '${providerId}' is disabled`);
throw new PaseoToolInputError(`Provider '${providerId}' is disabled`);
}
if (entry.status !== "ready") {
throw new Error(entry.error ?? `Provider '${providerId}' is unavailable`);
throw new PaseoToolInputError(entry.error ?? `Provider '${providerId}' is unavailable`);
}
const selectedModel = settings?.model ?? resolvedProviderModel.model;
const features = await agentManager.listDraftFeatures({
@@ -1,6 +1,18 @@
import type { z } from "zod";
import type { ProviderPaseoToolsPolicy } from "@getpaseo/protocol/provider-config";
export class PaseoToolInputError extends Error {
readonly status = 400;
constructor(
message: string,
readonly issues?: z.ZodError["issues"],
) {
super(message);
this.name = "PaseoToolInputError";
}
}
export interface PaseoToolExecutionContext {
signal?: AbortSignal;
sendUpdate?: (update: PaseoToolResult) => void;
@@ -0,0 +1,60 @@
import { expect, test } from "vitest";
import { mkdtemp, rm } from "node:fs/promises";
import { existsSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import pino from "pino";
import { ClaudeAgentClient } from "../agent/providers/claude/agent.js";
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
import { DaemonClient } from "../test-utils/daemon-client.js";
test("changing thinking then mode preserves a foreground Claude command", async () => {
const logger = pino({ level: "silent" });
const cwd = await mkdtemp(path.join(tmpdir(), "claude-settings-mid-turn-"));
const daemon = await createTestPaseoDaemon({
logger,
agentClients: { claude: new ClaudeAgentClient({ logger }) },
});
const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
try {
await client.connect();
await client.fetchAgents({ subscribe: {} });
const agent = await client.createAgent({
provider: "claude",
model: "claude-haiku-5-5",
modeId: "auto",
thinkingOptionId: "low",
cwd,
});
await client.subscribeAgentTimeline(agent.id, () => {}).ready;
await client.sendMessage(
agent.id,
"Run exactly these three Bash commands in the foreground, one per tool call, in order: python3 -c \"from pathlib import Path; import time; Path('started').write_text('started'); time.sleep(20); print(1)\", then the same with print(2), then print(3). Then reply DONE.",
);
// A tool-call row is emitted before Claude's permission check finishes.
// Wait for the command itself, so changing to acceptEdits cannot gate its start.
await expect.poll(() => existsSync(path.join(cwd, "started")), { timeout: 90_000 }).toBe(true);
const notice = await client.setAgentThinkingOption(agent.id, "medium");
expect(notice).toEqual({ type: "warning", message: "Thinking level applies next turn" });
await client.setAgentMode(agent.id, "acceptEdits");
await expect
.poll(
async () => {
const timeline = await client.fetchAgentTimeline(agent.id, { limit: 100 });
return timeline.entries.some(
({ item }) =>
item.type === "tool_call" &&
item.detail.type === "shell" &&
item.detail.command.includes("print(1)") &&
item.status === "completed",
);
},
{ timeout: 45_000, interval: 100 },
)
.toBe(true);
} finally {
await client.close();
await daemon.close();
await rm(cwd, { recursive: true, force: true });
}
}, 180_000);
@@ -5,6 +5,8 @@ import path from "node:path";
import { promisify } from "node:util";
import pino from "pino";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { afterEach, beforeAll, describe, expect, test } from "vitest";
import type { AgentTimelineItem } from "../agent/agent-sdk-types.js";
@@ -131,6 +133,14 @@ async function timeline(client: DaemonClient, agentId: string): Promise<AgentTim
return result.entries.map((entry) => entry.item);
}
async function readAssistantText(harness: Harness, agentId: string): Promise<string> {
const items = await timeline(harness.client, agentId);
return items
.filter((item) => item.type === "assistant_message")
.map((item) => item.text)
.join("");
}
function compactionFiller(records = 1200): string {
const teas = ["assam", "darjeeling", "sencha", "oolong"];
return Array.from(
@@ -257,8 +267,80 @@ async function contextWindowUsedTokens(client: DaemonClient, agentId: string) {
describe("daemon E2E (real OMP)", () => {
test(
"a prompt OMP rejects before its agent runs fails the turn with OMP's error",
"an idle out-of-band follow-up starts a turn",
async () => {
const harness = await createHarness();
try {
const agent = await createAgent(harness, "idle-follow-up");
await harness.client.sendMessage(agent.id, "/follow-up Reply exactly IDLE_FOLLOWUP_OK");
await expect
.poll(() => readAssistantText(harness, agent.id), { timeout: 30_000 })
.toContain("IDLE_FOLLOWUP_OK");
const finish = await harness.client.waitForFinish(agent.id, TIMEOUT_MS);
expect(finish.status).toBe("idle");
} finally {
await closeHarness(harness);
}
},
TIMEOUT_MS,
);
test(
"an agent-sent idle follow-up starts a turn and notifies its parent",
async () => {
const harness = await createHarness();
const mcp = new Client({ name: "idle-follow-up-test", version: "1.0.0" });
try {
const parent = await createAgent(harness, "follow-up-parent");
const child = await createAgent(harness, "follow-up-child");
await mcp.connect(
new StreamableHTTPClientTransport(
new URL(
`http://127.0.0.1:${harness.daemon.port}/mcp/agents?callerAgentId=${parent.id}`,
),
),
);
const response = await mcp.callTool({
name: "send_agent_prompt",
arguments: { agentId: child.id, prompt: "/follow-up Reply exactly MCP_IDLE_FOLLOWUP_OK" },
});
expect(response.isError).not.toBe(true);
expect(response.structuredContent).toMatchObject({
success: true,
disposition: "out_of_band",
});
await expect
.poll(() => readAssistantText(harness, child.id), { timeout: 30_000 })
.toContain("MCP_IDLE_FOLLOWUP_OK");
await expect
.poll(
async () => {
const items = await timeline(harness.client, parent.id);
let notifications = 0;
for (const item of items) {
if (
item.type === "tool_call" &&
item.agentMessage?.sender.id === child.id &&
item.agentMessage.event === "finished"
)
notifications++;
}
return notifications;
},
{ timeout: 15_000 },
)
.toBe(1);
} finally {
await mcp.close();
await closeHarness(harness);
}
},
TIMEOUT_MS,
);
test.each(["Reply with ok.", "/follow-up Reply with ok."])(
"a rejected %s fails the turn with OMP's error",
async (prompt) => {
const harness = await createHarness({ ompHome: createOfflineOmpHome() });
try {
const agent = await harness.client.createAgent({
@@ -268,7 +350,7 @@ describe("daemon E2E (real OMP)", () => {
model: "anthropic/claude-haiku-4-5",
modeId: "full",
});
await harness.client.sendMessage(agent.id, "Reply with ok.");
await harness.client.sendMessage(agent.id, prompt);
const finish = await harness.client.waitForFinish(agent.id, 60_000);
expect(finish.status).toBe("error");
expect(finish.error).toContain("No API key found for anthropic");
+20 -9
View File
@@ -38,6 +38,20 @@ export class ScheduleTargetGoneError extends Error {
}
}
export class ScheduleNotFoundError extends Error {
constructor(id: string) {
super(`Schedule not found: ${id}`);
this.name = "ScheduleNotFoundError";
}
}
export class ScheduleStateConflictError extends Error {
constructor(message: string) {
super(message);
this.name = "ScheduleStateConflictError";
}
}
function advanceNextRunAtPast(
cadence: StoredSchedule["cadence"],
nextRunAt: Date,
@@ -158,7 +172,7 @@ function shouldCompleteSchedule(schedule: StoredSchedule, now: Date): boolean {
function requireSchedule(schedule: StoredSchedule | null, id: string): StoredSchedule {
if (!schedule) {
throw new Error(`Schedule not found: ${id}`);
throw new ScheduleNotFoundError(id);
}
return schedule;
}
@@ -392,10 +406,7 @@ export class ScheduleService {
async inspect(id: string): Promise<StoredSchedule> {
const schedule = await this.store.get(id);
if (!schedule) {
throw new Error(`Schedule not found: ${id}`);
}
return schedule;
return requireSchedule(schedule, id);
}
async logs(id: string): Promise<ScheduleRun[]> {
@@ -406,7 +417,7 @@ export class ScheduleService {
async pause(id: string): Promise<StoredSchedule> {
const paused = await this.store.update(id, (schedule) => {
if (schedule.status === "completed") {
throw new Error(`Schedule ${id} is already completed`);
throw new ScheduleStateConflictError(`Schedule ${id} is already completed`);
}
if (schedule.status === "paused") {
return schedule;
@@ -426,7 +437,7 @@ export class ScheduleService {
async resume(id: string): Promise<StoredSchedule> {
const resumed = await this.store.update(id, (schedule) => {
if (schedule.status === "completed") {
throw new Error(`Schedule ${id} is already completed`);
throw new ScheduleStateConflictError(`Schedule ${id} is already completed`);
}
if (schedule.status === "active") {
return schedule;
@@ -546,10 +557,10 @@ export class ScheduleService {
async runOnce(id: string): Promise<StoredSchedule> {
const schedule = await this.inspect(id);
if (schedule.status === "completed") {
throw new Error(`Schedule ${id} is already completed`);
throw new ScheduleStateConflictError(`Schedule ${id} is already completed`);
}
if (this.runningScheduleIds.has(id)) {
throw new Error(`Schedule ${id} is already running`);
throw new ScheduleStateConflictError(`Schedule ${id} is already running`);
}
await this.runSchedule(schedule, this.now(), { manual: true });
return this.inspect(id);
@@ -0,0 +1,69 @@
import { randomUUID } from "node:crypto";
import {
MockLoadTestAgentClient,
MockLoadTestAgentSession,
} from "../agent/providers/mock-load-test-agent.js";
import { createTestPaseoDaemon } from "./paseo-daemon.js";
const firstResponse = Promise.withResolvers<void>();
class RetryQuestionSession extends MockLoadTestAgentSession {
private attempts = 0;
override async respondToPermission(
requestId: string,
response: Parameters<MockLoadTestAgentSession["respondToPermission"]>[1],
) {
this.attempts += 1;
process.send?.({ type: "attempt", requestId, response, sessionId: this.id });
if (this.attempts === 1) {
await firstResponse.promise;
throw new Error("Synthetic question response failure");
}
return super.respondToPermission(requestId, response);
}
}
class RetryQuestionClient extends MockLoadTestAgentClient {
override async createSession(config: Parameters<MockLoadTestAgentClient["createSession"]>[0]) {
const session = new RetryQuestionSession({ config, sessionId: randomUUID() });
session.subscribe((event) => {
if (event.type === "permission_resolved") {
process.send?.({
type: "resolved",
requestId: event.requestId,
response: event.resolution,
sessionId: session.id,
pendingRequestIds: session.getPendingPermissions().map((request) => request.id),
});
}
});
return session;
}
}
async function main(): Promise<void> {
const daemon = await createTestPaseoDaemon({
agentClients: { mock: new RetryQuestionClient() },
isDev: true,
mcpEnabled: false,
corsAllowedOrigins: [`http://localhost:${process.env.E2E_METRO_PORT}`],
});
process.on("message", (message: unknown) => {
if (message === "release") firstResponse.resolve();
});
const shutdown = async () => {
firstResponse.resolve();
await daemon.close();
process.exit(0);
};
process.once("SIGINT", () => void shutdown());
process.once("SIGTERM", () => void shutdown());
process.once("disconnect", () => void shutdown());
process.send?.({ type: "ready", port: daemon.port, serverId: daemon.daemon.getServerId() });
}
void main().catch((error: unknown) => {
process.send?.({ type: "error", error: error instanceof Error ? error.message : String(error) });
process.exit(1);
});