From 6ef1e67b0add36fec2149f4b0352002857c12ffa Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Tue, 13 Oct 2026 23:05:22 +0200 Subject: [PATCH 1/4] feat(ai-acp): serve a harness as an ACP v2 agent --- .changeset/harness-acp-agent.md | 5 + docs/config.json | 5 + docs/harness/acp.md | 87 ++++ packages/ai-acp/package.json | 14 +- packages/ai-acp/src/agent/index.ts | 384 ++++++++++++++++++ packages/ai-acp/tests/agent-handlers.test.ts | 398 +++++++++++++++++++ packages/ai-acp/tests/agent-media.test.ts | 267 +++++++++++++ packages/ai-acp/tests/agent.test.ts | 175 ++++++++ packages/ai-acp/tests/child-updates.test.ts | 62 +++ packages/ai-acp/vite.config.ts | 2 +- pnpm-lock.yaml | 18 +- 11 files changed, 1409 insertions(+), 8 deletions(-) create mode 100644 .changeset/harness-acp-agent.md create mode 100644 docs/harness/acp.md create mode 100644 packages/ai-acp/src/agent/index.ts create mode 100644 packages/ai-acp/tests/agent-handlers.test.ts create mode 100644 packages/ai-acp/tests/agent-media.test.ts create mode 100644 packages/ai-acp/tests/agent.test.ts create mode 100644 packages/ai-acp/tests/child-updates.test.ts diff --git a/.changeset/harness-acp-agent.md b/.changeset/harness-acp-agent.md new file mode 100644 index 0000000000..d4412ed886 --- /dev/null +++ b/.changeset/harness-acp-agent.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-acp': minor +--- + +`@tanstack/ai-acp/agent` adds `createAcpAgent` and `serveAcp`, which serve a harness as an ACP v2 agent (experimental, like the draft protocol). Tool approvals become permission requests, cancel cancels the running turn, and image, audio, and embedded file blocks go to the media store of the session. `@tanstack/ai-acp` now uses `@agentclientprotocol/sdk` 1.5. The existing ACP client adapters keep working. diff --git a/docs/config.json b/docs/config.json index 0d0de346cd..7fc9619ba8 100644 --- a/docs/config.json +++ b/docs/config.json @@ -992,6 +992,11 @@ "label": "Run agents from a harness", "to": "harness/subagents", "addedAt": "2026-10-13" + }, + { + "label": "Use a harness from an editor (ACP)", + "to": "harness/acp", + "addedAt": "2026-10-13" } ] }, diff --git a/docs/harness/acp.md b/docs/harness/acp.md new file mode 100644 index 0000000000..9a1429c73a --- /dev/null +++ b/docs/harness/acp.md @@ -0,0 +1,87 @@ +--- +title: Use a harness from an editor (ACP) +id: harness-acp +order: 6 +description: "Serve a harness as an ACP v2 agent, so an editor such as Zed can start it as a process and talk to it over stdio." +keywords: + - tanstack ai + - harness + - ACP + - agent client protocol + - editor +--- + +Your users want your agent in their editor. Editors such as Zed start an agent as a process and talk the Agent Client Protocol (ACP) over stdin and stdout. `serveAcp` makes your harness an ACP v2 agent. + +## 1. Install the package + + + +react: @tanstack/ai-acp +vue: @tanstack/ai-acp +solid: @tanstack/ai-acp +svelte: @tanstack/ai-acp +preact: @tanstack/ai-acp +angular: @tanstack/ai-acp +octane: @tanstack/ai-acp +vanilla: @tanstack/ai-acp + + + +## 2. Serve the harness + +Make a script that the editor starts: + +```ts group=harness-acp +import { serveAcp } from '@tanstack/ai-acp/agent' +import { createHarnessHost, defineHarness } from '@tanstack/ai-harness' +import { openaiText } from '@tanstack/ai-openai' +import { memoryPersistence } from '@tanstack/ai-persistence' + +export const assistant = defineHarness({ + name: 'acme/assistant', + adapter: openaiText('gpt-6.1-sol'), +}) + +const host = createHarnessHost({ persistence: memoryPersistence() }) + +serveAcp({ host, harness: assistant, version: '1.0.0' }) +``` + +- Each ACP session is one harness session. The ACP session id is the thread id. +- `version` is the version that `initialize` reports. Default: `'0.0.0'`. +- Use a persistent store instead of `memoryPersistence()` if a session must stay after the process stops. + +## 3. Add the agent to the editor + +Give the editor the command that runs the script, for example `node agent.js`. The editor then sends prompts, and the harness streams the answer back. + +## What the editor gets + +- The text and the reasoning of the main agent. Text from child agents stays out of the main message. +- Each tool call, of the main agent and of child agents, as a tool call update. +- Tool approvals as permission requests. The user picks **Allow** or **Reject**, and the turn continues. +- Cancel stops the running turn. A turn that fails ends with the `error` stop reason and the message. +- Image blocks, audio blocks, and embedded files of a prompt go to the media store of the session. A file that the harness refuses fails the prompt with an invalid params error. See [Send and show media](./media). + +## Known limits + +- ACP v2 is still a draft, so this API is experimental. +- The editor shows tool approvals only. A turn that asks the user a question, or that needs a sign-in, stops and waits, because the editor cannot answer it. +- The agent has no `session/list`. An editor cannot list the old sessions. +- A resource link in a prompt is only named. The harness does not read it. The model can open it with its own tools. + +## Use your own stream + +`serveAcp` uses stdin and stdout when you pass no `stream`. To use another transport, pass an ACP `Stream`, or call `createAcpAgent({ host, harness })` and connect the app yourself: + +```ts group=harness-acp +import { createAcpAgent } from '@tanstack/ai-acp/agent' + +export const app = createAcpAgent({ host, harness: assistant }) +``` + +## What you have now + +- Your harness as an agent in any ACP v2 editor. +- Tool approvals, cancel, and media from the editor. diff --git a/packages/ai-acp/package.json b/packages/ai-acp/package.json index 95e8182eef..4893a07c80 100644 --- a/packages/ai-acp/package.json +++ b/packages/ai-acp/package.json @@ -26,6 +26,10 @@ ".": { "types": "./dist/esm/index.d.ts", "import": "./dist/esm/index.js" + }, + "./agent": { + "types": "./dist/esm/agent/index.d.ts", + "import": "./dist/esm/agent/index.js" } }, "files": [ @@ -44,14 +48,22 @@ "test:types": "tsc" }, "dependencies": { - "@agentclientprotocol/sdk": "^0.25.0" + "@agentclientprotocol/sdk": "^1.5.0" }, "peerDependencies": { "@tanstack/ai": "workspace:^", + "@tanstack/ai-harness": "workspace:^", "@tanstack/ai-sandbox": "workspace:^" }, + "peerDependenciesMeta": { + "@tanstack/ai-harness": { + "optional": true + } + }, "devDependencies": { "@tanstack/ai": "workspace:*", + "@tanstack/ai-harness": "workspace:*", + "@tanstack/ai-persistence": "workspace:*", "@tanstack/ai-sandbox": "workspace:*", "@tanstack/ai-sandbox-local-process": "workspace:*", "@vitest/coverage-v8": "4.1.10" diff --git a/packages/ai-acp/src/agent/index.ts b/packages/ai-acp/src/agent/index.ts new file mode 100644 index 0000000000..df78c14969 --- /dev/null +++ b/packages/ai-acp/src/agent/index.ts @@ -0,0 +1,384 @@ +import { EventType } from '@tanstack/ai' +import { + ContentBlock, + PROTOCOL_VERSION, + RequestError, + agent, + ndJsonStream, +} from '@agentclientprotocol/sdk/experimental/v2' +import { MediaError, kindOf, mediaPart } from '@tanstack/ai-harness' +import type { + AgentApp, + AgentContext, + AgentConnection, + SessionUpdate, + Stream, +} from '@agentclientprotocol/sdk/experimental/v2' +import type { + ContentPart, + Interrupt, + RunAgentResumeItem, + StreamChunk, +} from '@tanstack/ai' +import type { + AnyHarness, + HarnessHost, + HarnessSession, + Operation, +} from '@tanstack/ai-harness' + +export interface AcpAgentOptions { + host: HarnessHost + harness: AnyHarness + /** Reported in `initialize`. Default `'0.0.0'`. */ + version?: string +} + +function textPart(content: string) { + const part: ContentPart = { type: 'text', content } + return part +} + +/** A text line that names a resource the prompt does not include. */ +function linkPart(uri: string) { + return textPart(`[resource link: ${uri}]`) +} + +/** The last path segment of a URI, as a file name. */ +function fileName(uri: string) { + return uri.split('/').pop() || uri +} + +/** + * Store the base64 `data` of a block in the media store of the session, and + * return the part that sends it. Throws a `MediaError` when the harness + * refuses the file. + */ +async function storedPart( + session: HarnessSession, + data: string, + info: { mimeType: string; name: string }, +) { + const bytes = Uint8Array.from(atob(data), (char) => char.charCodeAt(0)) + return mediaPart(await session.putMedia(bytes, info)) +} + +/** One ACP prompt block as a content part, or `undefined` to skip it. */ +async function promptPart(session: HarnessSession, block: ContentBlock) { + // The SDK guards check the payload too, so a malformed block is skipped. + if (ContentBlock.isText(block)) { + return block.text === '' ? undefined : textPart(block.text) + } + if (ContentBlock.isImage(block)) { + return storedPart(session, block.data, { + mimeType: block.mimeType, + name: block.uri ? fileName(block.uri) : 'image', + }) + } + if (ContentBlock.isAudio(block)) { + return storedPart(session, block.data, { + mimeType: block.mimeType, + name: 'audio', + }) + } + // ponytail: a resource link is only named. The agent does not read it; the + // model can open it with its own tools. + if (ContentBlock.isResourceLink(block)) return linkPart(block.uri) + if (!ContentBlock.isResource(block)) return undefined + const { resource } = block + if ('text' in resource) { + return textPart(`[resource: ${resource.uri}]\n${resource.text}`) + } + const { mimeType } = resource + const isStored = + typeof mimeType === 'string' && kindOf(mimeType) !== undefined + if (!isStored) return linkPart(resource.uri) + return storedPart(session, resource.blob, { + mimeType, + name: fileName(resource.uri), + }) +} + +/** + * The turn input of an ACP prompt. Image and audio blocks, and embedded blobs + * of a type the harness stores, go to the media store of the session and + * become media parts. Embedded text and resource links become text. A prompt + * of only text is one string, as before. + */ +async function promptInput( + session: HarnessSession, + prompt: ReadonlyArray, +) { + const blocks = await Promise.all( + prompt.map((block) => promptPart(session, block)), + ) + const parts = blocks.filter((part) => part !== undefined) + const texts = parts.flatMap((part) => + part.type === 'text' ? [part.content] : [], + ) + return texts.length === parts.length ? texts.join('\n') : parts +} + +/** One AG-UI chunk as an ACP session update, or `undefined` to skip it. */ +export function toSessionUpdate(chunk: StreamChunk): SessionUpdate | undefined { + // A child agent's text and thoughts stay out of the main message. Its tool + // calls (the edits and commands it runs) show up like the lead's own. + const fromChild = 'subagentRunId' in chunk && Boolean(chunk.subagentRunId) + if (chunk.type === EventType.TEXT_MESSAGE_CONTENT) { + if (fromChild) return undefined + return { + sessionUpdate: 'agent_message_chunk', + messageId: chunk.messageId, + content: { type: 'text', text: chunk.delta }, + } + } + if (chunk.type === EventType.REASONING_MESSAGE_CONTENT) { + if (fromChild) return undefined + return { + sessionUpdate: 'agent_thought_chunk', + messageId: chunk.messageId, + content: { type: 'text', text: chunk.delta }, + } + } + if (chunk.type === EventType.TOOL_CALL_START) { + return { + sessionUpdate: 'tool_call_update', + toolCallId: chunk.toolCallId, + name: chunk.toolCallName, + title: chunk.toolCallName, + status: 'in_progress', + } + } + if (chunk.type === EventType.TOOL_CALL_RESULT) { + return { + sessionUpdate: 'tool_call_update', + toolCallId: chunk.toolCallId, + status: 'completed', + } + } + return undefined +} + +const APPROVAL_OPTIONS = [ + { optionId: 'allow', name: 'Allow', kind: 'allow_once' }, + { optionId: 'reject', name: 'Reject', kind: 'reject_once' }, +] as const + +/** + * An ACP v2 agent backed by a harness. Each ACP session is one harness + * session. Prompts follow the harness `busy` setting, tool approvals become + * `session/request_permission` requests, and cancel cancels the running turn. + * Image, audio, and embedded resource blocks of a prompt go to the media + * store of the session. A file the harness refuses (too big, or a type it + * does not take) fails `session/prompt` with an invalid params error. + * + * ACP v2 is a draft, so this API is experimental too. + */ +export function createAcpAgent(options: AcpAgentOptions): AgentApp { + const { host, harness } = options + const sessions = new Map() + + const sessionFor = async (sessionId: string) => { + let session = sessions.get(sessionId) + if (!session) { + session = await host.open(harness, { threadId: sessionId }) + sessions.set(sessionId, session) + } + return session + } + + const notify = ( + client: AgentContext, + sessionId: string, + update: SessionUpdate, + ) => client.notify('session/update', { sessionId, update }).catch(() => {}) + + /** + * Ask the client about each approval, then continue the turn. The next + * operation is wrapped, because an operation is thenable and an async + * function would unwrap it. + */ + const approve = async ( + client: AgentContext, + sessionId: string, + session: HarnessSession, + interrupts: ReadonlyArray, + ): Promise<{ next: Operation | undefined }> => { + await notify(client, sessionId, { + sessionUpdate: 'state_update', + state: 'requires_action', + }) + const resume: Array = [] + for (const interrupt of interrupts) { + const toolCallId = interrupt.toolCallId ?? interrupt.id + const response = await client.request('session/request_permission', { + sessionId, + title: interrupt.message ?? 'Allow this tool call?', + subject: { type: 'tool_call', toolCall: { toolCallId } }, + options: [...APPROVAL_OPTIONS], + }) + const allowed = + response.outcome.outcome === 'selected' && + 'optionId' in response.outcome && + response.outcome.optionId === 'allow' + resume.push({ + interruptId: interrupt.id, + status: 'resolved', + payload: allowed, + }) + } + const receipt = await session.resolve(resume) + return { + next: receipt.operationId + ? session.operation(receipt.operationId) + : undefined, + } + } + + /** + * Stream one turn to the client, through any approvals, then report idle. + * A failed turn, or a failure here, ends with the `error` stop reason. An + * interrupt that is not a tool approval (a question, a sign-in) stays open, + * because the client cannot answer it, so the turn waits. + */ + const pump = async ( + client: AgentContext, + sessionId: string, + session: HarnessSession, + first: Operation, + ) => { + let operation: Operation | undefined = first + let stopReason: 'end_turn' | 'cancelled' | 'error' = 'end_turn' + let failure: string | undefined + try { + while (operation) { + await notify(client, sessionId, { + sessionUpdate: 'state_update', + state: 'running', + }) + for await (const chunk of operation.stream()) { + const update = toSessionUpdate(chunk) + if (update) await notify(client, sessionId, update) + } + const result: unknown = await operation.then( + (value) => value, + (error: unknown) => { + failure = error instanceof Error ? error.message : String(error) + return undefined + }, + ) + if (operation.status() === 'cancelled') stopReason = 'cancelled' + if (operation.status() === 'failed') stopReason = 'error' + const interrupts = + typeof result === 'object' && + result !== null && + 'interrupts' in result && + Array.isArray(result.interrupts) + ? (result.interrupts as Array) + : [] + const areApprovals = + interrupts.length > 0 && + interrupts.every( + (interrupt) => + interrupt.metadata?.kind === 'approval' || + interrupt.reason === 'approval_required', + ) + operation = areApprovals + ? (await approve(client, sessionId, session, interrupts)).next + : undefined + } + } catch (error) { + stopReason = 'error' + failure = error instanceof Error ? error.message : String(error) + } + await notify( + client, + sessionId, + stopReason === 'error' + ? { + sessionUpdate: 'state_update', + state: 'idle', + stopReason, + error: { code: -32603, message: failure ?? 'The turn failed.' }, + } + : { sessionUpdate: 'state_update', state: 'idle', stopReason }, + ) + } + + return agent() + .onRequest('initialize', () => ({ + protocolVersion: PROTOCOL_VERSION, + info: { name: harness.name, version: options.version ?? '0.0.0' }, + // Text and resource links are the baseline. Images, audio, and embedded + // resources go to the media store of the session. + capabilities: { + session: { prompt: { image: {}, audio: {}, embeddedContext: {} } }, + }, + })) + .onRequest('session/new', async () => { + const sessionId = `acp-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 9)}` + await sessionFor(sessionId) + return { sessionId } + }) + .onRequest('session/resume', async ({ params }) => { + await sessionFor(params.sessionId) + return {} + }) + .onRequest('session/prompt', async ({ params, client }) => { + const session = await sessionFor(params.sessionId) + const input = await promptInput(session, params.prompt).catch( + (error: unknown) => { + // The editor sent a file the harness refuses, so the prompt is + // invalid: answer with the reason, not an internal error. + if (error instanceof MediaError) { + throw RequestError.invalidParams( + { status: error.status }, + error.message, + ) + } + throw error + }, + ) + const operation = session.prompt(input) + // `pump` catches its own errors and always reports idle. + void pump(client, params.sessionId, session, operation) + return { messageId: operation.id } + }) + .onRequest('session/close', async ({ params }) => { + const session = sessions.get(params.sessionId) + sessions.delete(params.sessionId) + await session?.close() + }) + .onNotification('session/cancel', async ({ params }) => { + await sessions.get(params.sessionId)?.cancel() + }) +} + +/** + * Serve a harness as an ACP v2 agent over a stream. With no stream, it uses + * stdin and stdout as newline-delimited JSON (for editors that start the + * agent as a process). + */ +export function serveAcp( + options: AcpAgentOptions & { stream?: Stream }, +): AgentConnection { + return createAcpAgent(options).connect(options.stream ?? stdioStream()) +} + +function stdioStream(): Stream { + const output = new WritableStream({ + write: (chunk) => + new Promise((resolve, reject) => + process.stdout.write(chunk, (error) => + error ? reject(error) : resolve(), + ), + ), + }) + const input = new ReadableStream({ + start(controller) { + process.stdin.on('data', (data: Uint8Array) => controller.enqueue(data)) + process.stdin.on('end', () => controller.close()) + }, + }) + return ndJsonStream(output, input) +} diff --git a/packages/ai-acp/tests/agent-handlers.test.ts b/packages/ai-acp/tests/agent-handlers.test.ts new file mode 100644 index 0000000000..4bc4867714 --- /dev/null +++ b/packages/ai-acp/tests/agent-handlers.test.ts @@ -0,0 +1,398 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { EventType, toolDefinition } from '@tanstack/ai' +import { client, ndJsonStream } from '@agentclientprotocol/sdk/experimental/v2' +import { memoryPersistence } from '@tanstack/ai-persistence' +import { createHarnessHost, defineHarness } from '@tanstack/ai-harness' +import { createAcpAgent, serveAcp, toSessionUpdate } from '../src/agent' +import type { AnyTextAdapter, ModelMessage, StreamChunk } from '@tanstack/ai' +import type { + ClientApp, + SessionUpdate, +} from '@agentclientprotocol/sdk/experimental/v2' + +const now = () => Date.now() + +/** + * A model that answers from the last user message: `wait` waits until the + * turn is cancelled, `remove` calls the remove tool, anything else echoes. + */ +function model(): AnyTextAdapter { + return { + kind: 'text', + name: 'mock', + model: 'test-model', + '~types': { + providerOptions: {} as Record, + inputModalities: ['text'] as readonly ['text'], + messageMetadataByModality: { + text: undefined as unknown, + image: undefined as unknown, + audio: undefined as unknown, + video: undefined as unknown, + document: undefined as unknown, + }, + toolCapabilities: [] as ReadonlyArray, + toolCallMetadata: undefined as unknown, + systemPromptMetadata: undefined as never, + }, + structuredOutput: async () => ({ data: {}, rawText: '{}' }), + chatStream: (options: any) => + (async function* (): AsyncGenerator { + const last: ModelMessage | undefined = options.messages.at(-1) + // A prompt with an image comes as parts: read its text parts. + const said = + typeof last?.content === 'string' + ? last.content + : (last?.content ?? []) + .map((part) => (part.type === 'text' ? part.content : '')) + .join('') + yield { + type: EventType.RUN_STARTED, + runId: 'r', + threadId: 't', + timestamp: now(), + } + if (said === 'wait') { + const signal: AbortSignal | undefined = + options.abortController?.signal ?? options.request?.signal + await new Promise((resolve) => { + if (!signal || signal.aborted) return resolve() + signal.addEventListener('abort', () => resolve(), { once: true }) + }) + return + } + if (said.startsWith('remove')) { + yield { + type: EventType.TOOL_CALL_START, + toolCallId: 'call_1', + toolCallName: 'remove', + timestamp: now(), + } + yield { + type: EventType.TOOL_CALL_ARGS, + toolCallId: 'call_1', + delta: '{"path":"a.txt"}', + timestamp: now(), + } + yield { + type: EventType.TOOL_CALL_END, + toolCallId: 'call_1', + timestamp: now(), + } + yield { + type: EventType.RUN_FINISHED, + runId: 'r', + threadId: 't', + timestamp: now(), + metadata: { tanstack: { finishReason: 'tool_calls' } }, + } + return + } + const reply = last?.role === 'tool' ? 'Kept a.txt.' : `echo: ${said}` + yield { + type: EventType.TEXT_MESSAGE_START, + messageId: 'm', + role: 'assistant', + timestamp: now(), + } + yield { + type: EventType.TEXT_MESSAGE_CONTENT, + messageId: 'm', + delta: reply, + timestamp: now(), + } + yield { + type: EventType.TEXT_MESSAGE_END, + messageId: 'm', + timestamp: now(), + } + yield { + type: EventType.RUN_FINISHED, + runId: 'r', + threadId: 't', + timestamp: now(), + metadata: { tanstack: { finishReason: 'stop' } }, + } + })(), + } +} + +function setup() { + const remove = vi.fn(async () => ({ removed: true })) + const harness = defineHarness({ + name: 'test/acp-handlers', + adapter: model(), + tools: [ + toolDefinition({ + name: 'remove', + description: 'Remove a file', + needsApproval: true, + inputSchema: { + type: 'object', + properties: { path: { type: 'string' } }, + required: ['path'], + }, + }).server(remove), + ], + }) + const host = createHarnessHost({ persistence: memoryPersistence() }) + const updates: Array<{ sessionId: string; update: SessionUpdate }> = [] + const permissions: Array<{ title: string; toolCallId: string }> = [] + const editor: ClientApp = client() + .onNotification('session/update', ({ params }) => { + updates.push({ sessionId: params.sessionId, update: params.update }) + }) + .onRequest('session/request_permission', ({ params }) => { + permissions.push({ + title: params.title, + // The subject names the tool call: `{ type, toolCall: { toolCallId } }`. + toolCallId: JSON.stringify(params.subject ?? null).includes('call_1') + ? 'call_1' + : '', + }) + return { outcome: { outcome: 'selected', optionId: 'reject' } } + }) + const idle = (sessionId: string) => + updates.find( + (entry) => + entry.sessionId === sessionId && + entry.update.sessionUpdate === 'state_update' && + entry.update.state === 'idle', + )?.update + const textOf = (sessionId: string) => + updates + .filter((entry) => entry.sessionId === sessionId) + .map((entry) => { + if (entry.update.sessionUpdate !== 'agent_message_chunk') return '' + const content: unknown = entry.update.content + return typeof content === 'object' && + content !== null && + 'text' in content && + typeof content.text === 'string' + ? content.text + : '' + }) + .join('') + const running = (sessionId: string) => + updates.some( + (entry) => + entry.sessionId === sessionId && + entry.update.sessionUpdate === 'state_update' && + entry.update.state === 'running', + ) + return { + harness, + host, + editor, + updates, + permissions, + remove, + idle, + running, + textOf, + } +} + +describe('toSessionUpdate', () => { + it('maps reasoning and tool results, and skips child and unknown events', () => { + expect( + toSessionUpdate({ + type: EventType.REASONING_MESSAGE_CONTENT, + messageId: 'r1', + delta: 'thinking', + timestamp: 1, + }), + ).toEqual({ + sessionUpdate: 'agent_thought_chunk', + messageId: 'r1', + content: { type: 'text', text: 'thinking' }, + }) + expect( + toSessionUpdate({ + type: EventType.TOOL_CALL_RESULT, + messageId: 'm', + toolCallId: 'call_9', + content: 'done', + timestamp: 1, + }), + ).toEqual({ + sessionUpdate: 'tool_call_update', + toolCallId: 'call_9', + status: 'completed', + }) + expect( + toSessionUpdate({ + type: EventType.TEXT_MESSAGE_CONTENT, + messageId: 'm', + delta: 'from a child', + subagentRunId: 'child-1', + timestamp: 1, + } as StreamChunk), + ).toBeUndefined() + expect( + toSessionUpdate({ + type: EventType.RUN_STARTED, + runId: 'r', + threadId: 't', + timestamp: 1, + }), + ).toBeUndefined() + }) +}) + +describe('ACP agent handlers', () => { + let close: (() => Promise) | undefined + afterEach(async () => { + await close?.() + close = undefined + }) + + it('keeps the tool from running when the editor rejects it', async () => { + const { harness, host, editor, permissions, remove, idle, textOf } = setup() + close = () => host.close() + await editor.connectWith(createAcpAgent({ host, harness }), async (ctx) => { + const init = await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + expect(init.info.version).toBe('0.0.0') + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + await ctx.request('session/prompt', { + sessionId, + prompt: [ + { type: 'image', data: 'AA==', mimeType: 'image/png' }, + { type: 'text', text: 'remove a.txt' }, + ], + }) + await vi.waitFor(() => expect(idle(sessionId)).toBeDefined()) + expect(idle(sessionId)).toMatchObject({ stopReason: 'end_turn' }) + expect(textOf(sessionId)).toBe('Kept a.txt.') + }) + expect(remove).not.toHaveBeenCalled() + expect(permissions).toEqual([ + { title: expect.any(String), toolCallId: 'call_1' }, + ]) + }) + + it('resumes a session by id, cancels a running turn, and closes it', async () => { + const { harness, host, editor, idle, running, textOf } = setup() + close = () => host.close() + await editor.connectWith(createAcpAgent({ host, harness }), async (ctx) => { + await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + await ctx.request('session/resume', { + sessionId: 'kept-session', + cwd: '/tmp', + }) + await ctx.request('session/prompt', { + sessionId: 'kept-session', + prompt: [{ type: 'text', text: 'hello' }], + }) + await vi.waitFor(() => expect(idle('kept-session')).toBeDefined()) + expect(textOf('kept-session')).toBe('echo: hello') + + await ctx.request('session/prompt', { + sessionId: 'slow-session', + prompt: [{ type: 'text', text: 'wait' }], + }) + await vi.waitFor(() => expect(running('slow-session')).toBe(true)) + await ctx.notify('session/cancel', { sessionId: 'slow-session' }) + await vi.waitFor(() => expect(idle('slow-session')).toBeDefined()) + expect(idle('slow-session')).toMatchObject({ stopReason: 'cancelled' }) + + await ctx.request('session/close', { sessionId: 'kept-session' }) + // Closing an unknown session and cancelling one are both quiet. + await ctx.request('session/close', { sessionId: 'never-opened' }) + await ctx.notify('session/cancel', { sessionId: 'never-opened' }) + }) + }) +}) + +describe('serveAcp', () => { + it('serves the agent over a stream', async () => { + const { harness, host, editor, idle, textOf } = setup() + const toClient = new TransformStream() + const toAgent = new TransformStream() + const connection = serveAcp({ + host, + harness, + version: '2.0.0', + stream: ndJsonStream(toClient.writable, toAgent.readable), + }) + await editor.connectWith( + ndJsonStream(toAgent.writable, toClient.readable), + async (ctx) => { + const init = await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + expect(init.info).toEqual({ + name: 'test/acp-handlers', + version: '2.0.0', + }) + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'over a stream' }], + }) + await vi.waitFor(() => expect(idle(sessionId)).toBeDefined()) + expect(textOf(sessionId)).toBe('echo: over a stream') + }, + ) + connection.close() + await host.close() + }) + + it('uses stdin and stdout when no stream is given', async () => { + const { harness, host } = setup() + const listeners = new Map void>() + const on = vi + .spyOn(process.stdin, 'on') + .mockImplementation((event: string | symbol, listener: any) => { + listeners.set(String(event), listener) + return process.stdin + }) + const written: Array = [] + const write = vi + .spyOn(process.stdout, 'write') + .mockImplementation((chunk: any, callback?: any) => { + written.push(Buffer.from(chunk).toString()) + if (typeof callback === 'function') callback() + return true + }) + try { + const connection = serveAcp({ host, harness }) + await vi.waitFor(() => expect(listeners.has('data')).toBe(true)) + listeners.get('data')?.( + new TextEncoder().encode( + `${JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }, + })}\n`, + ), + ) + await vi.waitFor(() => + expect(written.join('')).toContain('test/acp-handlers'), + ) + listeners.get('end')?.() + connection.close() + } finally { + on.mockRestore() + write.mockRestore() + await host.close() + } + }) +}) diff --git a/packages/ai-acp/tests/agent-media.test.ts b/packages/ai-acp/tests/agent-media.test.ts new file mode 100644 index 0000000000..5dbee7a619 --- /dev/null +++ b/packages/ai-acp/tests/agent-media.test.ts @@ -0,0 +1,267 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { EventType } from '@tanstack/ai' +import { client } from '@agentclientprotocol/sdk/experimental/v2' +import { memoryPersistence } from '@tanstack/ai-persistence' +import { + createHarnessHost, + defineHarness, + mediaOfMessage, +} from '@tanstack/ai-harness' +import { createAcpAgent } from '../src/agent' +import type { AnyTextAdapter, StreamChunk } from '@tanstack/ai' +import type { + ClientContext, + ContentBlock, + InitializeResponse, +} from '@agentclientprotocol/sdk/experimental/v2' +import type { MediaOptions, MediaRecord } from '@tanstack/ai-harness' + +const now = () => Date.now() + +/** Base64 of the four bytes 1, 2, 3, 4. */ +const FOUR_BYTES = 'AQIDBA==' + +// ponytail: a cut-down copy of the fake model in agent.test.ts. Move the +// copies to one helpers file when a fourth test file needs one. +/** A model that answers "ok" to every call. */ +function model(): AnyTextAdapter { + return { + kind: 'text', + name: 'mock', + model: 'test-model', + '~types': { + providerOptions: {} as Record, + inputModalities: ['text'] as readonly ['text'], + messageMetadataByModality: { + text: undefined as unknown, + image: undefined as unknown, + audio: undefined as unknown, + video: undefined as unknown, + document: undefined as unknown, + }, + toolCapabilities: [] as ReadonlyArray, + toolCallMetadata: undefined as unknown, + systemPromptMetadata: undefined as never, + }, + structuredOutput: async () => ({ data: {}, rawText: '{}' }), + chatStream: () => + (async function* (): AsyncGenerator { + yield { + type: EventType.RUN_STARTED, + runId: 'r', + threadId: 't', + timestamp: now(), + } + yield { + type: EventType.TEXT_MESSAGE_START, + messageId: 'm', + role: 'assistant', + timestamp: now(), + } + yield { + type: EventType.TEXT_MESSAGE_CONTENT, + messageId: 'm', + delta: 'ok', + timestamp: now(), + } + yield { + type: EventType.TEXT_MESSAGE_END, + messageId: 'm', + timestamp: now(), + } + yield { + type: EventType.RUN_FINISHED, + runId: 'r', + threadId: 't', + timestamp: now(), + metadata: { tanstack: { finishReason: 'stop' } }, + } + })(), + } +} + +let close: (() => Promise) | undefined +afterEach(async () => { + await close?.() + close = undefined +}) + +/** + * A harness with `media` options behind an ACP agent. `connect` runs `run` + * with an editor that has sent `initialize`, and gives it that answer. + */ +function setup(media?: MediaOptions) { + const harness = defineHarness({ + name: 'test/acp-media', + adapter: model(), + media, + }) + const host = createHarnessHost({ persistence: memoryPersistence() }) + close = () => host.close() + const connect = ( + run: (ctx: ClientContext, init: InitializeResponse) => Promise, + ) => + client().connectWith(createAcpAgent({ host, harness }), async (ctx) => + run( + ctx, + await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }), + ), + ) + return { harness, host, connect } +} + +/** + * Send `prompt` to a new ACP session and wait for the turn. Returns the + * harness session and the user message the turn saved. + */ +async function savedPrompt(prompt: Array) { + const { harness, host, connect } = setup() + const sent = await connect(async (ctx) => { + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + const { messageId } = await ctx.request('session/prompt', { + sessionId, + prompt, + }) + return { sessionId, messageId } + }) + const session = await host.open(harness, { threadId: sent.sessionId }) + await session.operation(sent.messageId) + const [user] = await session.transcript() + if (user?.role !== 'user') throw new Error('The turn saved no user message.') + return { session, user } +} + +describe('ACP agent media', () => { + it('advertises image, audio, and embedded context prompts in initialize', async () => { + const { connect } = setup() + + const init = await connect(async (_ctx, answer) => answer) + + expect(init.info).toEqual({ name: 'test/acp-media', version: '0.0.0' }) + expect(init.capabilities).toEqual({ + session: { prompt: { image: {}, audio: {}, embeddedContext: {} } }, + }) + }) + + it.each([ + { + label: 'an image block', + block: { + type: 'image', + data: FOUR_BYTES, + mimeType: 'image/png', + uri: 'file:///shots/cat.png', + }, + stored: { kind: 'image', mimeType: 'image/png', name: 'cat.png' }, + }, + { + label: 'an audio block', + block: { type: 'audio', data: FOUR_BYTES, mimeType: 'audio/wav' }, + stored: { kind: 'audio', mimeType: 'audio/wav', name: 'audio' }, + }, + { + label: 'an embedded blob resource', + block: { + type: 'resource', + resource: { + uri: 'file:///docs/spec.pdf', + blob: FOUR_BYTES, + mimeType: 'application/pdf', + }, + }, + stored: { + kind: 'document', + mimeType: 'application/pdf', + name: 'spec.pdf', + }, + }, + ] satisfies Array<{ + label: string + block: ContentBlock + stored: Pick + }>)( + 'stores $label and sends it as a media part', + async ({ block, stored }) => { + const { session, user } = await savedPrompt([ + { type: 'text', text: 'What is this?' }, + block, + ]) + + const [record] = mediaOfMessage(user) + if (!record) throw new Error('The user message has no media record.') + expect(record).toMatchObject({ ...stored, size: 4, source: 'user' }) + expect(user.content).toEqual([ + { type: 'text', content: 'What is this?' }, + { + type: stored.kind, + source: { + type: 'url', + value: `harness-media:${record.id}`, + mimeType: stored.mimeType, + }, + }, + ]) + expect(await session.loadMedia(record.id)).toEqual( + new Uint8Array([1, 2, 3, 4]), + ) + }, + ) + + it('sends text resources, resource links, and blobs it cannot store as text', async () => { + const { user } = await savedPrompt([ + { type: 'text', text: 'Review these.' }, + { + type: 'resource', + resource: { + uri: 'file:///src/a.ts', + text: 'export const a = 1', + mimeType: 'text/typescript', + }, + }, + { type: 'resource_link', name: 'b.ts', uri: 'file:///src/b.ts' }, + { + type: 'resource', + resource: { + uri: 'file:///build.zip', + blob: FOUR_BYTES, + mimeType: 'application/zip', + }, + }, + ]) + + expect(user.content).toBe( + 'Review these.\n[resource: file:///src/a.ts]\nexport const a = 1\n[resource link: file:///src/b.ts]\n[resource link: file:///build.zip]', + ) + }) + + it('answers the prompt with an error, and runs no turn, when a file is over the limit', async () => { + const { harness, host, connect } = setup({ maxBytes: 2 }) + + const threadId = await connect(async (ctx) => { + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + await expect( + ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'image', data: FOUR_BYTES, mimeType: 'image/png' }], + }), + ).rejects.toMatchObject({ + code: -32602, + message: + 'Invalid params: The file is bigger than the limit of 2 bytes.', + }) + return sessionId + }) + + const session = await host.open(harness, { threadId }) + expect(await session.transcript()).toEqual([]) + }) +}) diff --git a/packages/ai-acp/tests/agent.test.ts b/packages/ai-acp/tests/agent.test.ts new file mode 100644 index 0000000000..51a7db4590 --- /dev/null +++ b/packages/ai-acp/tests/agent.test.ts @@ -0,0 +1,175 @@ +import { describe, expect, it, vi } from 'vitest' +import { EventType, toolDefinition } from '@tanstack/ai' +import { client } from '@agentclientprotocol/sdk/experimental/v2' +import { memoryPersistence } from '@tanstack/ai-persistence' +import { createHarnessHost, defineHarness } from '@tanstack/ai-harness' +import { createAcpAgent } from '../src/agent' +import type { AnyTextAdapter, StreamChunk } from '@tanstack/ai' +import type { SessionUpdate } from '@agentclientprotocol/sdk/experimental/v2' + +const now = () => Date.now() + +function scripted(turns: Array>) { + let call = 0 + const adapter: AnyTextAdapter = { + kind: 'text', + name: 'mock', + model: 'test-model', + '~types': { + providerOptions: {} as Record, + inputModalities: ['text'] as readonly ['text'], + messageMetadataByModality: { + text: undefined as unknown, + image: undefined as unknown, + audio: undefined as unknown, + video: undefined as unknown, + document: undefined as unknown, + }, + toolCapabilities: [] as ReadonlyArray, + toolCallMetadata: undefined as unknown, + systemPromptMetadata: undefined as never, + }, + chatStream: () => { + const chunks = turns[call] ?? [] + call += 1 + return (async function* () { + yield* chunks + })() + }, + structuredOutput: async () => ({ data: {}, rawText: '{}' }), + } + return adapter +} + +const textTurn = (text: string): Array => [ + { type: EventType.RUN_STARTED, runId: 'r', threadId: 't', timestamp: now() }, + { + type: EventType.TEXT_MESSAGE_START, + messageId: 'm', + role: 'assistant', + timestamp: now(), + }, + { + type: EventType.TEXT_MESSAGE_CONTENT, + messageId: 'm', + delta: text, + timestamp: now(), + }, + { type: EventType.TEXT_MESSAGE_END, messageId: 'm', timestamp: now() }, + { + type: EventType.RUN_FINISHED, + runId: 'r', + threadId: 't', + timestamp: now(), + metadata: { tanstack: { finishReason: 'stop' } }, + }, +] + +const toolTurn = (name: string, args: string): Array => [ + { type: EventType.RUN_STARTED, runId: 'r', threadId: 't', timestamp: now() }, + { + type: EventType.TOOL_CALL_START, + toolCallId: 'call_1', + toolCallName: name, + timestamp: now(), + }, + { + type: EventType.TOOL_CALL_ARGS, + toolCallId: 'call_1', + delta: args, + timestamp: now(), + }, + { type: EventType.TOOL_CALL_END, toolCallId: 'call_1', timestamp: now() }, + { + type: EventType.RUN_FINISHED, + runId: 'r', + threadId: 't', + timestamp: now(), + metadata: { tanstack: { finishReason: 'tool_calls' } }, + }, +] + +describe('ACP v2 agent', () => { + it('streams a prompt, asks for approval, and reports idle', async () => { + const remove = vi.fn(async () => ({ removed: true })) + const harness = defineHarness({ + name: 'test/acp', + adapter: scripted([ + toolTurn('remove', '{"path":"a.txt"}'), + textTurn('Removed a.txt.'), + ]), + tools: [ + toolDefinition({ + name: 'remove', + description: 'Remove a file', + needsApproval: true, + inputSchema: { + type: 'object', + properties: { path: { type: 'string' } }, + required: ['path'], + }, + }).server(remove), + ], + }) + const host = createHarnessHost({ persistence: memoryPersistence() }) + const agentApp = createAcpAgent({ host, harness, version: '1.0.0' }) + + const updates: Array = [] + const permissions: Array = [] + const editor = client() + .onNotification('session/update', ({ params }) => { + updates.push(params.update) + }) + .onRequest('session/request_permission', ({ params }) => { + permissions.push(params.title) + return { outcome: { outcome: 'selected', optionId: 'allow' } } + }) + + await editor.connectWith(agentApp, async (ctx) => { + const init = await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'test-editor', version: '1.0.0' }, + }) + expect(init.info.name).toBe('test/acp') + + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + const prompted = await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'remove a.txt' }], + }) + expect(prompted.messageId).toMatch(/^op-chat-/) + + await vi.waitFor(() => + expect( + updates.some( + (update) => + update.sessionUpdate === 'state_update' && + update.state === 'idle', + ), + ).toBe(true), + ) + }) + + expect(permissions).toHaveLength(1) + expect(remove).toHaveBeenCalledTimes(1) + const states = updates + .filter((update) => update.sessionUpdate === 'state_update') + .map((update) => ('state' in update ? update.state : '')) + expect(states).toEqual(['running', 'requires_action', 'running', 'idle']) + const text = updates + .map((update) => + update.sessionUpdate === 'agent_message_chunk' + ? JSON.stringify(update) + : '', + ) + .join('') + expect(text).toContain('Removed a.txt.') + expect( + updates.some((update) => update.sessionUpdate === 'tool_call_update'), + ).toBe(true) + await host.close() + }) +}) diff --git a/packages/ai-acp/tests/child-updates.test.ts b/packages/ai-acp/tests/child-updates.test.ts new file mode 100644 index 0000000000..fd67ee7526 --- /dev/null +++ b/packages/ai-acp/tests/child-updates.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, it } from 'vitest' +import { EventType } from '@tanstack/ai' +import { toSessionUpdate } from '../src/agent' +import type { StreamChunk } from '@tanstack/ai' + +const child = (event: Record): StreamChunk => + ({ subagentRunId: 'child-1', timestamp: 1, ...event }) as StreamChunk + +describe('toSessionUpdate for child agents', () => { + it('shows child tool calls like the lead tool calls', () => { + expect( + toSessionUpdate( + child({ + type: EventType.TOOL_CALL_START, + toolCallId: 'edit-1', + toolCallName: 'Edit', + }), + ), + ).toEqual({ + sessionUpdate: 'tool_call_update', + toolCallId: 'edit-1', + name: 'Edit', + title: 'Edit', + status: 'in_progress', + }) + expect( + toSessionUpdate( + child({ + type: EventType.TOOL_CALL_RESULT, + messageId: 'm', + toolCallId: 'edit-1', + content: 'ok', + }), + ), + ).toEqual({ + sessionUpdate: 'tool_call_update', + toolCallId: 'edit-1', + status: 'completed', + }) + }) + + it('keeps child text and thoughts out of the main message', () => { + expect( + toSessionUpdate( + child({ + type: EventType.TEXT_MESSAGE_CONTENT, + messageId: 'm', + delta: 'hi', + }), + ), + ).toBeUndefined() + expect( + toSessionUpdate( + child({ + type: EventType.REASONING_MESSAGE_CONTENT, + messageId: 'r', + delta: 'hmm', + }), + ), + ).toBeUndefined() + }) +}) diff --git a/packages/ai-acp/vite.config.ts b/packages/ai-acp/vite.config.ts index 588ff924d2..ab0a9e68fc 100644 --- a/packages/ai-acp/vite.config.ts +++ b/packages/ai-acp/vite.config.ts @@ -33,7 +33,7 @@ const config = defineConfig({ // Put package test settings last so they win over the shared build config. export default mergeConfig( tanstackViteConfig({ - entry: ['./src/index.ts'], + entry: ['./src/index.ts', './src/agent/index.ts'], srcDir: './src', cjs: false, }), diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8ca632dd08..8bb68566b9 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2043,12 +2043,18 @@ importers: packages/ai-acp: dependencies: '@agentclientprotocol/sdk': - specifier: ^0.25.0 - version: 0.25.1(zod@4.3.6) + specifier: ^1.5.0 + version: 1.8.0(zod@4.6.5) devDependencies: '@tanstack/ai': specifier: workspace:* version: link:../ai + '@tanstack/ai-harness': + specifier: workspace:* + version: link:../ai-harness + '@tanstack/ai-persistence': + specifier: workspace:* + version: link:../ai-persistence '@tanstack/ai-sandbox': specifier: workspace:* version: link:../ai-sandbox @@ -4019,8 +4025,8 @@ packages: zod: optional: true - '@agentclientprotocol/sdk@0.25.1': - resolution: {integrity: sha512-jx2rF3bdpGwZ75Q/meyEDLLbYmbtxk82Uh9hDCdxDvcEedBnNSF5hZAnL/kJR5VNz56JqwOmqnAqasC84MwwkQ==} + '@agentclientprotocol/sdk@1.8.0': + resolution: {integrity: sha512-fbP7qHZsFyTJVT0/RPrcWJ8An26qLXHEuGCHHEKDyxD1xpXV+Ejyh2LAE6vBAFePkpVjZeeohnbgvuEhZ7wLMg==} peerDependencies: zod: ^3.25.0 || ^4.0.0 @@ -19021,9 +19027,9 @@ snapshots: optionalDependencies: zod: 4.2.1 - '@agentclientprotocol/sdk@0.25.1(zod@4.3.6)': + '@agentclientprotocol/sdk@1.8.0(zod@4.6.5)': dependencies: - zod: 4.3.6 + zod: 4.6.5 '@ampproject/remapping@2.3.0': dependencies: From 3fd6a0ceb1d1700783d135531c6f7a59cb6b7b28 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Tue, 13 Oct 2026 23:30:52 +0200 Subject: [PATCH 2/4] fix(ai-acp): give each idle stop reason its own update type --- packages/ai-acp/src/agent/index.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/ai-acp/src/agent/index.ts b/packages/ai-acp/src/agent/index.ts index df78c14969..c0b3ebec9e 100644 --- a/packages/ai-acp/src/agent/index.ts +++ b/packages/ai-acp/src/agent/index.ts @@ -301,7 +301,9 @@ export function createAcpAgent(options: AcpAgentOptions): AgentApp { stopReason, error: { code: -32603, message: failure ?? 'The turn failed.' }, } - : { sessionUpdate: 'state_update', state: 'idle', stopReason }, + : stopReason === 'cancelled' + ? { sessionUpdate: 'state_update', state: 'idle', stopReason } + : { sessionUpdate: 'state_update', state: 'idle', stopReason }, ) } From 201b53ec7eaef1e5f7c5c61d78315d243984684d Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Wed, 14 Oct 2026 17:07:05 +0200 Subject: [PATCH 3/4] feat(ai-acp): list the sessions of the harness with session/list ACP v2 agents with sessions must answer session/list. The agent reads the top-level sessions of the harness from the host's session index, 50 per page. Each session reports the agent's cwd (new cwd option, default process.cwd()), because the harness keeps no cwd per thread. --- .changeset/harness-acp-agent.md | 2 +- docs/harness/acp.md | 13 +++++- packages/ai-acp/src/agent/index.ts | 35 +++++++++++++- packages/ai-acp/tests/agent-handlers.test.ts | 49 ++++++++++++++++++++ 4 files changed, 94 insertions(+), 5 deletions(-) diff --git a/.changeset/harness-acp-agent.md b/.changeset/harness-acp-agent.md index d4412ed886..f860ff5c0a 100644 --- a/.changeset/harness-acp-agent.md +++ b/.changeset/harness-acp-agent.md @@ -2,4 +2,4 @@ '@tanstack/ai-acp': minor --- -`@tanstack/ai-acp/agent` adds `createAcpAgent` and `serveAcp`, which serve a harness as an ACP v2 agent (experimental, like the draft protocol). Tool approvals become permission requests, cancel cancels the running turn, and image, audio, and embedded file blocks go to the media store of the session. `@tanstack/ai-acp` now uses `@agentclientprotocol/sdk` 1.5. The existing ACP client adapters keep working. +`@tanstack/ai-acp/agent` adds `createAcpAgent` and `serveAcp`, which serve a harness as an ACP v2 agent (experimental, like the draft protocol). Tool approvals become permission requests, cancel cancels the running turn, and image, audio, and embedded file blocks go to the media store of the session. `session/list` lists the sessions of the harness from the session index of the host. `@tanstack/ai-acp` now uses `@agentclientprotocol/sdk` 1.5. The existing ACP client adapters keep working. diff --git a/docs/harness/acp.md b/docs/harness/acp.md index 9a1429c73a..8eb4a69640 100644 --- a/docs/harness/acp.md +++ b/docs/harness/acp.md @@ -50,6 +50,7 @@ serveAcp({ host, harness: assistant, version: '1.0.0' }) - Each ACP session is one harness session. The ACP session id is the thread id. - `version` is the version that `initialize` reports. Default: `'0.0.0'`. +- `cwd` is the working directory that the session list shows for each session. Default: `process.cwd()`. - Use a persistent store instead of `memoryPersistence()` if a session must stay after the process stops. ## 3. Add the agent to the editor @@ -64,11 +65,19 @@ Give the editor the command that runs the script, for example `node agent.js`. T - Cancel stops the running turn. A turn that fails ends with the `error` stop reason and the message. - Image blocks, audio blocks, and embedded files of a prompt go to the media store of the session. A file that the harness refuses fails the prompt with an invalid params error. See [Send and show media](./media). +## Open an old session + +The editor lists the sessions of the harness, newest first. The user picks one, and the next prompt continues that thread. + +- The list shows the top-level sessions of the harness. Sessions of child agents stay out. +- The list comes from the session index of the host. Use a persistent store, so that the list keeps the sessions after the process stops. + ## Known limits - ACP v2 is still a draft, so this API is experimental. - The editor shows tool approvals only. A turn that asks the user a question, or that needs a sign-in, stops and waits, because the editor cannot answer it. -- The agent has no `session/list`. An editor cannot list the old sessions. +- Each session in the list shows the same working directory: the `cwd` of the agent. If the editor asks for the sessions of another directory, the list is empty. +- An old session opens without its earlier messages. The editor shows only the new turns. The model still gets the full thread. - A resource link in a prompt is only named. The harness does not read it. The model can open it with its own tools. ## Use your own stream @@ -84,4 +93,4 @@ export const app = createAcpAgent({ host, harness: assistant }) ## What you have now - Your harness as an agent in any ACP v2 editor. -- Tool approvals, cancel, and media from the editor. +- Tool approvals, cancel, media, and old sessions from the editor. diff --git a/packages/ai-acp/src/agent/index.ts b/packages/ai-acp/src/agent/index.ts index c0b3ebec9e..eefc3115b0 100644 --- a/packages/ai-acp/src/agent/index.ts +++ b/packages/ai-acp/src/agent/index.ts @@ -32,8 +32,16 @@ export interface AcpAgentOptions { harness: AnyHarness /** Reported in `initialize`. Default `'0.0.0'`. */ version?: string + /** + * The working directory that `session/list` reports for each session. + * Default `process.cwd()`. + */ + cwd?: string } +/** The most sessions in one `session/list` page. */ +const LIST_PAGE_SIZE = 50 + function textPart(content: string) { const part: ContentPart = { type: 'text', content } return part @@ -166,8 +174,10 @@ const APPROVAL_OPTIONS = [ /** * An ACP v2 agent backed by a harness. Each ACP session is one harness - * session. Prompts follow the harness `busy` setting, tool approvals become - * `session/request_permission` requests, and cancel cancels the running turn. + * session, and `session/list` reads the top-level sessions of the harness from + * the session index of the host. Prompts follow the harness `busy` setting, + * tool approvals become `session/request_permission` requests, and cancel + * cancels the running turn. * Image, audio, and embedded resource blocks of a prompt go to the media * store of the session. A file the harness refuses (too big, or a type it * does not take) fails `session/prompt` with an invalid params error. @@ -322,6 +332,27 @@ export function createAcpAgent(options: AcpAgentOptions): AgentApp { await sessionFor(sessionId) return { sessionId } }) + .onRequest('session/list', async ({ params }) => { + // ponytail: the harness keeps no cwd per thread, so every session + // reports the cwd of the agent. Store one per thread when one agent + // serves more than one project. + const cwd = options.cwd ?? process.cwd() + if (params.cwd && params.cwd !== cwd) return { sessions: [] } + const page = await host.sessions.list({ + harness: harness.name, + limit: LIST_PAGE_SIZE, + ...(params.cursor ? { cursor: params.cursor } : {}), + }) + return { + sessions: page.entries.map((entry) => ({ + sessionId: entry.threadId, + cwd, + title: entry.title ?? null, + updatedAt: new Date(entry.updatedAt).toISOString(), + })), + nextCursor: page.truncated ? page.cursor : null, + } + }) .onRequest('session/resume', async ({ params }) => { await sessionFor(params.sessionId) return {} diff --git a/packages/ai-acp/tests/agent-handlers.test.ts b/packages/ai-acp/tests/agent-handlers.test.ts index 4bc4867714..e75ca9ae87 100644 --- a/packages/ai-acp/tests/agent-handlers.test.ts +++ b/packages/ai-acp/tests/agent-handlers.test.ts @@ -277,6 +277,55 @@ describe('ACP agent handlers', () => { ]) }) + it('lists the sessions of the harness, a page at a time', async () => { + const { harness, host, editor } = setup() + close = () => host.close() + const other = defineHarness({ name: 'test/other', adapter: model() }) + await host.open(other, { threadId: 'other-thread' }) + for (let index = 0; index < 50; index++) { + await host.open(harness, { threadId: `old-${index}` }) + } + await editor.connectWith( + createAcpAgent({ host, harness, cwd: '/work' }), + async (ctx) => { + await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + const { sessionId } = await ctx.request('session/new', { + cwd: '/work', + mcpServers: [], + }) + await host.sessions.rename(sessionId, 'Fix the build') + + const first = await ctx.request('session/list', {}) + expect(first.sessions).toHaveLength(50) + expect(first.nextCursor).toEqual(expect.any(String)) + const second = await ctx.request('session/list', { + cursor: first.nextCursor, + }) + expect(second.nextCursor ?? undefined).toBeUndefined() + const sessions = [...first.sessions, ...second.sessions] + const ids = sessions.map((session) => session.sessionId) + expect(ids).toHaveLength(51) + expect(ids).not.toContain('other-thread') + expect( + sessions.find((session) => session.sessionId === sessionId), + ).toEqual({ + sessionId, + cwd: '/work', + title: 'Fix the build', + updatedAt: expect.stringMatching(/^\d{4}-\d{2}-\d{2}T/), + }) + + const elsewhere = await ctx.request('session/list', { + cwd: '/somewhere/else', + }) + expect(elsewhere.sessions).toEqual([]) + }, + ) + }) + it('resumes a session by id, cancels a running turn, and closes it', async () => { const { harness, host, editor, idle, running, textOf } = setup() close = () => host.close() From 130250edf3b9f2c8b1ce5a01229080688e5a6389 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Wed, 14 Oct 2026 18:05:17 +0200 Subject: [PATCH 4/4] fix(ai-acp): make approvals, cancel, and queued prompts work in the agent - Cancel during a permission request ends the turn as cancelled. The next prompt cancels the open interrupts first, because the thread takes no new input while they are open. - Interrupts the editor cannot answer (sign-in, input form, client tool) are cancelled, and the model continues. - A rejected resolve receipt ends the turn with an error. - The prompts of a session run one after another, so a later prompt cannot overtake a permission request or report out of order. - session/prompt echoes a user_message with the returned messageId. - Data that is not base64 fails with invalid params. - Unknown session ids fail with resource not found. - The agent keeps no session objects; host.open returns the live one. --- docs/harness/acp.md | 6 +- packages/ai-acp/src/agent/index.ts | 212 ++++++++---- packages/ai-acp/tests/agent-handlers.test.ts | 336 ++++++++++++++++++- packages/ai-acp/tests/agent-media.test.ts | 18 +- packages/ai-acp/tests/agent.test.ts | 2 +- 5 files changed, 495 insertions(+), 79 deletions(-) diff --git a/docs/harness/acp.md b/docs/harness/acp.md index 8eb4a69640..e87958c313 100644 --- a/docs/harness/acp.md +++ b/docs/harness/acp.md @@ -49,6 +49,7 @@ serveAcp({ host, harness: assistant, version: '1.0.0' }) ``` - Each ACP session is one harness session. The ACP session id is the thread id. +- The prompts of a session run one after another. The harness `busy` setting does not apply. - `version` is the version that `initialize` reports. Default: `'0.0.0'`. - `cwd` is the working directory that the session list shows for each session. Default: `process.cwd()`. - Use a persistent store instead of `memoryPersistence()` if a session must stay after the process stops. @@ -62,7 +63,7 @@ Give the editor the command that runs the script, for example `node agent.js`. T - The text and the reasoning of the main agent. Text from child agents stays out of the main message. - Each tool call, of the main agent and of child agents, as a tool call update. - Tool approvals as permission requests. The user picks **Allow** or **Reject**, and the turn continues. -- Cancel stops the running turn. A turn that fails ends with the `error` stop reason and the message. +- Cancel stops the running turn, also while the editor asks for an approval. A turn that fails ends with the `error` stop reason and the message. - Image blocks, audio blocks, and embedded files of a prompt go to the media store of the session. A file that the harness refuses fails the prompt with an invalid params error. See [Send and show media](./media). ## Open an old session @@ -71,11 +72,12 @@ The editor lists the sessions of the harness, newest first. The user picks one, - The list shows the top-level sessions of the harness. Sessions of child agents stay out. - The list comes from the session index of the host. Use a persistent store, so that the list keeps the sessions after the process stops. +- A session id that the session index does not know fails with a not found error. ## Known limits - ACP v2 is still a draft, so this API is experimental. -- The editor shows tool approvals only. A turn that asks the user a question, or that needs a sign-in, stops and waits, because the editor cannot answer it. +- The editor shows tool approvals only. The agent cancels a sign-in, an input form, or a client tool call, because the editor cannot answer it. The model then continues without the answer. A question from `session.ask` waits until the user cancels the turn. - Each session in the list shows the same working directory: the `cwd` of the agent. If the editor asks for the sessions of another directory, the list is empty. - An old session opens without its earlier messages. The editor shows only the new turns. The model still gets the full thread. - A resource link in a prompt is only named. The harness does not read it. The model can open it with its own tools. diff --git a/packages/ai-acp/src/agent/index.ts b/packages/ai-acp/src/agent/index.ts index eefc3115b0..3a3659d581 100644 --- a/packages/ai-acp/src/agent/index.ts +++ b/packages/ai-acp/src/agent/index.ts @@ -6,7 +6,12 @@ import { agent, ndJsonStream, } from '@agentclientprotocol/sdk/experimental/v2' -import { MediaError, kindOf, mediaPart } from '@tanstack/ai-harness' +import { + InputRejectedError, + MediaError, + kindOf, + mediaPart, +} from '@tanstack/ai-harness' import type { AgentApp, AgentContext, @@ -59,16 +64,22 @@ function fileName(uri: string) { /** * Store the base64 `data` of a block in the media store of the session, and - * return the part that sends it. Throws a `MediaError` when the harness - * refuses the file. + * return the part that sends it. Throws an invalid params error for data that + * is not base64, and a `MediaError` when the harness refuses the file. */ async function storedPart( session: HarnessSession, data: string, info: { mimeType: string; name: string }, ) { - const bytes = Uint8Array.from(atob(data), (char) => char.charCodeAt(0)) - return mediaPart(await session.putMedia(bytes, info)) + // `Buffer.from` skips the characters that are not base64, so check first. + if (!/^[A-Za-z0-9+/]*={0,2}$/.test(data)) { + throw RequestError.invalidParams( + undefined, + `The data of ${info.name} is not base64.`, + ) + } + return mediaPart(await session.putMedia(Buffer.from(data, 'base64'), info)) } /** One ACP prompt block as a content part, or `undefined` to skip it. */ @@ -167,6 +178,14 @@ export function toSessionUpdate(chunk: StreamChunk): SessionUpdate | undefined { return undefined } +/** True for an interrupt that asks to approve a tool call. */ +function isApproval(interrupt: Interrupt) { + return ( + interrupt.metadata?.kind === 'approval' || + interrupt.reason === 'approval_required' + ) +} + const APPROVAL_OPTIONS = [ { optionId: 'allow', name: 'Allow', kind: 'allow_once' }, { optionId: 'reject', name: 'Reject', kind: 'reject_once' }, @@ -175,10 +194,10 @@ const APPROVAL_OPTIONS = [ /** * An ACP v2 agent backed by a harness. Each ACP session is one harness * session, and `session/list` reads the top-level sessions of the harness from - * the session index of the host. Prompts follow the harness `busy` setting, - * tool approvals become `session/request_permission` requests, and cancel - * cancels the running turn. - * Image, audio, and embedded resource blocks of a prompt go to the media + * the session index of the host. The prompts of a session run one after + * another, whatever the harness `busy` setting says. Tool approvals become + * `session/request_permission` requests, and cancel cancels the running + * turn. Image, audio, and embedded resource blocks of a prompt go to the media * store of the session. A file the harness refuses (too big, or a type it * does not take) fails `session/prompt` with an invalid params error. * @@ -186,15 +205,23 @@ const APPROVAL_OPTIONS = [ */ export function createAcpAgent(options: AcpAgentOptions): AgentApp { const { host, harness } = options - const sessions = new Map() + /** The ids of the sessions that this agent opened. */ + const opened = new Set() + /** The last pump of each session. The pumps of one session run in order. */ + const tails = new Map>() - const sessionFor = async (sessionId: string) => { - let session = sessions.get(sessionId) - if (!session) { - session = await host.open(harness, { threadId: sessionId }) - sessions.set(sessionId, session) + // The host gives back the live session of a thread, or opens it again after + // it closed, so the agent keeps no session objects. + const live = (sessionId: string) => + host.open(harness, { threadId: sessionId }) + + /** The session of an id that this agent or the session index knows. */ + const known = async (sessionId: string) => { + if (!opened.has(sessionId) && !(await host.sessions.get(sessionId))) { + throw RequestError.resourceNotFound(sessionId) } - return session + opened.add(sessionId) + return live(sessionId) } const notify = ( @@ -204,40 +231,49 @@ export function createAcpAgent(options: AcpAgentOptions): AgentApp { ) => client.notify('session/update', { sessionId, update }).catch(() => {}) /** - * Ask the client about each approval, then continue the turn. The next - * operation is wrapped, because an operation is thenable and an async - * function would unwrap it. + * Answer the interrupts of a turn, then continue it. The client gets a + * permission request for each tool approval. The other interrupts (a + * sign-in, an input form, a client tool) are cancelled, because the client + * cannot answer them. A permission answer other than a selected option (the + * client cancelled the prompt) stops the turn, and the interrupts stay open + * until the next prompt. The next operation is wrapped, because an + * operation is thenable and an async function would unwrap it. */ - const approve = async ( + const answer = async ( client: AgentContext, sessionId: string, session: HarnessSession, interrupts: ReadonlyArray, - ): Promise<{ next: Operation | undefined }> => { - await notify(client, sessionId, { - sessionUpdate: 'state_update', - state: 'requires_action', - }) + ): Promise<{ next?: Operation; cancelled?: boolean }> => { + if (interrupts.some(isApproval)) { + await notify(client, sessionId, { + sessionUpdate: 'state_update', + state: 'requires_action', + }) + } const resume: Array = [] for (const interrupt of interrupts) { + if (!isApproval(interrupt)) { + resume.push({ interruptId: interrupt.id, status: 'cancelled' }) + continue + } const toolCallId = interrupt.toolCallId ?? interrupt.id - const response = await client.request('session/request_permission', { + const { outcome } = await client.request('session/request_permission', { sessionId, title: interrupt.message ?? 'Allow this tool call?', subject: { type: 'tool_call', toolCall: { toolCallId } }, options: [...APPROVAL_OPTIONS], }) - const allowed = - response.outcome.outcome === 'selected' && - 'optionId' in response.outcome && - response.outcome.optionId === 'allow' + if (outcome.outcome !== 'selected') return { cancelled: true } resume.push({ interruptId: interrupt.id, status: 'resolved', - payload: allowed, + payload: 'optionId' in outcome && outcome.optionId === 'allow', }) } const receipt = await session.resolve(resume) + // For example `busy`: a later prompt runs already. + if (receipt.status === 'rejected') throw new InputRejectedError(receipt) return { next: receipt.operationId ? session.operation(receipt.operationId) @@ -246,26 +282,66 @@ export function createAcpAgent(options: AcpAgentOptions): AgentApp { } /** - * Stream one turn to the client, through any approvals, then report idle. - * A failed turn, or a failure here, ends with the `error` stop reason. An - * interrupt that is not a tool approval (a question, a sign-in) stays open, - * because the client cannot answer it, so the turn waits. + * Cancel the open interrupts of a session (from a cancelled approval, or + * from another client), and wait for the turn that takes the answers. The + * thread takes no new prompt while an interrupt is open. Resolves `false` + * when the client cancelled that turn: its interrupts stay open. + */ + const closeInterrupts = async (session: HarnessSession) => { + const open = session.snapshot().pendingInterrupts + if (open.length === 0) return true + const receipt = await session.resolve( + open.map((interrupt) => ({ + interruptId: interrupt.id, + status: 'cancelled' as const, + })), + ) + if (receipt.status === 'rejected') throw new InputRejectedError(receipt) + // ponytail: the reply of the model to the cancel stays out of the client. + const turn = receipt.operationId + ? session.operation(receipt.operationId) + : undefined + let failure: unknown + await turn?.then( + () => {}, + (error: unknown) => { + failure = error + }, + ) + if (turn?.status() === 'cancelled') return false + if (turn?.status() === 'failed') throw failure + if (session.snapshot().pendingInterrupts.length > 0) { + throw new Error( + 'The model asked again for input that the client cannot give.', + ) + } + return true + } + + /** + * Close the open interrupts, run one prompt, and stream its turn to the + * client through its interrupts, then report idle. A failed turn, or a + * failure here, ends with the `error` stop reason. */ const pump = async ( client: AgentContext, sessionId: string, session: HarnessSession, - first: Operation, + input: Parameters[0], ) => { - let operation: Operation | undefined = first + let operation: Operation | undefined let stopReason: 'end_turn' | 'cancelled' | 'error' = 'end_turn' let failure: string | undefined + const running = () => + notify(client, sessionId, { + sessionUpdate: 'state_update', + state: 'running', + }) try { + await running() + if (await closeInterrupts(session)) operation = session.prompt(input) + else stopReason = 'cancelled' while (operation) { - await notify(client, sessionId, { - sessionUpdate: 'state_update', - state: 'running', - }) for await (const chunk of operation.stream()) { const update = toSessionUpdate(chunk) if (update) await notify(client, sessionId, update) @@ -286,16 +362,11 @@ export function createAcpAgent(options: AcpAgentOptions): AgentApp { Array.isArray(result.interrupts) ? (result.interrupts as Array) : [] - const areApprovals = - interrupts.length > 0 && - interrupts.every( - (interrupt) => - interrupt.metadata?.kind === 'approval' || - interrupt.reason === 'approval_required', - ) - operation = areApprovals - ? (await approve(client, sessionId, session, interrupts)).next - : undefined + if (interrupts.length === 0) break + const answered = await answer(client, sessionId, session, interrupts) + if (answered.cancelled) stopReason = 'cancelled' + operation = answered.next + if (operation) await running() } } catch (error) { stopReason = 'error' @@ -329,7 +400,8 @@ export function createAcpAgent(options: AcpAgentOptions): AgentApp { })) .onRequest('session/new', async () => { const sessionId = `acp-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 9)}` - await sessionFor(sessionId) + opened.add(sessionId) + await live(sessionId) return { sessionId } }) .onRequest('session/list', async ({ params }) => { @@ -354,11 +426,12 @@ export function createAcpAgent(options: AcpAgentOptions): AgentApp { } }) .onRequest('session/resume', async ({ params }) => { - await sessionFor(params.sessionId) + await known(params.sessionId) return {} }) .onRequest('session/prompt', async ({ params, client }) => { - const session = await sessionFor(params.sessionId) + const { sessionId } = params + const session = await known(sessionId) const input = await promptInput(session, params.prompt).catch( (error: unknown) => { // The editor sent a file the harness refuses, so the prompt is @@ -372,18 +445,33 @@ export function createAcpAgent(options: AcpAgentOptions): AgentApp { throw error }, ) - const operation = session.prompt(input) - // `pump` catches its own errors and always reports idle. - void pump(client, params.sessionId, session, operation) - return { messageId: operation.id } + const messageId = `user-${crypto.randomUUID()}` + await notify(client, sessionId, { + sessionUpdate: 'user_message', + messageId, + content: params.prompt, + }) + // `pump` catches its own errors and always reports idle. The prompts of + // one session go to the harness in order, each after the turn before it + // ended, so a later prompt cannot overtake a permission request. + const tail = (tails.get(sessionId) ?? Promise.resolve()).then(() => + pump(client, sessionId, session, input), + ) + tails.set(sessionId, tail) + void tail.then(() => { + if (tails.get(sessionId) === tail) tails.delete(sessionId) + }) + return { messageId } }) .onRequest('session/close', async ({ params }) => { - const session = sessions.get(params.sessionId) - sessions.delete(params.sessionId) - await session?.close() + if (!opened.delete(params.sessionId)) return + await (await live(params.sessionId)).close() }) .onNotification('session/cancel', async ({ params }) => { - await sessions.get(params.sessionId)?.cancel() + // While the client asks for an approval, no turn runs: the client + // answers that request with `cancelled`, and the turn stops there. + if (!opened.has(params.sessionId)) return + await (await live(params.sessionId)).cancel() }) } diff --git a/packages/ai-acp/tests/agent-handlers.test.ts b/packages/ai-acp/tests/agent-handlers.test.ts index e75ca9ae87..299ab3d274 100644 --- a/packages/ai-acp/tests/agent-handlers.test.ts +++ b/packages/ai-acp/tests/agent-handlers.test.ts @@ -7,6 +7,7 @@ import { createAcpAgent, serveAcp, toSessionUpdate } from '../src/agent' import type { AnyTextAdapter, ModelMessage, StreamChunk } from '@tanstack/ai' import type { ClientApp, + RequestPermissionOutcome, SessionUpdate, } from '@agentclientprotocol/sdk/experimental/v2' @@ -14,7 +15,9 @@ const now = () => Date.now() /** * A model that answers from the last user message: `wait` waits until the - * turn is cancelled, `remove` calls the remove tool, anything else echoes. + * turn is cancelled, `remove` calls the remove tool, `pick` calls the pick + * client tool, anything else echoes. After the tool of `remove slowly`, it + * waits until the turn is cancelled too. */ function model(): AnyTextAdapter { return { @@ -52,7 +55,12 @@ function model(): AnyTextAdapter { threadId: 't', timestamp: now(), } - if (said === 'wait') { + const asked: ModelMessage | undefined = options.messages.findLast( + (message: ModelMessage) => message.role === 'user', + ) + const isSlow = + last?.role === 'tool' && asked?.content === 'remove a.txt slowly' + if (said === 'wait' || isSlow) { const signal: AbortSignal | undefined = options.abortController?.signal ?? options.request?.signal await new Promise((resolve) => { @@ -61,11 +69,16 @@ function model(): AnyTextAdapter { }) return } - if (said.startsWith('remove')) { + const tool = said.startsWith('remove') + ? 'remove' + : said === 'pick' + ? 'pick' + : undefined + if (tool && last?.role === 'user') { yield { type: EventType.TOOL_CALL_START, toolCallId: 'call_1', - toolCallName: 'remove', + toolCallName: tool, timestamp: now(), } yield { @@ -133,11 +146,21 @@ function setup() { required: ['path'], }, }).server(remove), + // No server function: the client runs it, and an editor cannot. + toolDefinition({ + name: 'pick', + description: 'Pick a file in the editor', + inputSchema: { type: 'object', properties: {} }, + }), ], }) const host = createHarnessHost({ persistence: memoryPersistence() }) const updates: Array<{ sessionId: string; update: SessionUpdate }> = [] const permissions: Array<{ title: string; toolCallId: string }> = [] + // The answer of the editor to each permission request. Default: Reject. + const editorState: { answer: () => Promise } = { + answer: async () => ({ outcome: 'selected', optionId: 'reject' }), + } const editor: ClientApp = client() .onNotification('session/update', ({ params }) => { updates.push({ sessionId: params.sessionId, update: params.update }) @@ -150,7 +173,7 @@ function setup() { ? 'call_1' : '', }) - return { outcome: { outcome: 'selected', optionId: 'reject' } } + return editorState.answer().then((outcome) => ({ outcome })) }) const idle = (sessionId: string) => updates.find( @@ -173,6 +196,17 @@ function setup() { : '' }) .join('') + /** The states of a session, in order, with the stop reason of an idle. */ + const states = (sessionId: string) => + updates.flatMap(({ sessionId: id, update }) => + id === sessionId && update.sessionUpdate === 'state_update' + ? [ + update.state === 'idle' && 'stopReason' in update + ? `idle:${String(update.stopReason)}` + : update.state, + ] + : [], + ) const running = (sessionId: string) => updates.some( (entry) => @@ -186,8 +220,10 @@ function setup() { editor, updates, permissions, + editorState, remove, idle, + states, running, textOf, } @@ -326,9 +362,299 @@ describe('ACP agent handlers', () => { ) }) + it('stops the turn when the editor cancels an approval', async () => { + const { harness, host, editor, editorState, remove, states, textOf } = + setup() + close = () => host.close() + editorState.answer = async () => ({ outcome: 'cancelled' }) + await editor.connectWith(createAcpAgent({ host, harness }), async (ctx) => { + await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'remove a.txt' }], + }) + await vi.waitFor(() => + expect(states(sessionId)).toContain('idle:cancelled'), + ) + expect(states(sessionId)).toEqual([ + 'running', + 'requires_action', + 'idle:cancelled', + ]) + // The next prompt cancels the open approval first, then runs. + await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'hello' }], + }) + await vi.waitFor(() => + expect(states(sessionId)).toContain('idle:end_turn'), + ) + expect(textOf(sessionId)).toBe('echo: hello') + }) + expect(remove).not.toHaveBeenCalled() + }) + + it('cancels an interrupt that the editor cannot answer, and goes on', async () => { + const { harness, host, editor, idle, permissions, textOf } = setup() + close = () => host.close() + await editor.connectWith(createAcpAgent({ host, harness }), async (ctx) => { + await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'pick' }], + }) + await vi.waitFor(() => expect(idle(sessionId)).toBeDefined()) + expect(idle(sessionId)).toMatchObject({ stopReason: 'end_turn' }) + expect(textOf(sessionId)).toBe('Kept a.txt.') + }) + expect(permissions).toEqual([]) + }) + + it('echoes the user message with the id of the prompt response', async () => { + const { harness, host, editor, updates } = setup() + close = () => host.close() + await editor.connectWith(createAcpAgent({ host, harness }), async (ctx) => { + await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + const prompt = [{ type: 'text' as const, text: 'hello' }] + const { messageId } = await ctx.request('session/prompt', { + sessionId, + prompt, + }) + expect(updates).toContainEqual({ + sessionId, + update: { sessionUpdate: 'user_message', messageId, content: prompt }, + }) + }) + }) + + it('reports a queued prompt after the turn before it', async () => { + const { harness, host, editor, running, states, textOf } = setup() + close = () => host.close() + await editor.connectWith(createAcpAgent({ host, harness }), async (ctx) => { + await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'wait' }], + }) + await vi.waitFor(() => expect(running(sessionId)).toBe(true)) + await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'hello' }], + }) + await ctx.notify('session/cancel', { sessionId }) + await vi.waitFor(() => + expect(states(sessionId)).toContain('idle:end_turn'), + ) + expect(states(sessionId)).toEqual([ + 'running', + 'idle:cancelled', + 'running', + 'idle:end_turn', + ]) + expect(textOf(sessionId)).toBe('echo: hello') + }) + }) + + it('runs a prompt sent during an approval after that approval', async () => { + const { + harness, + host, + editor, + editorState, + permissions, + remove, + states, + textOf, + } = setup() + close = () => host.close() + let allow = () => {} + editorState.answer = () => + new Promise((resolve) => { + allow = () => resolve({ outcome: 'selected', optionId: 'allow' }) + }) + await editor.connectWith(createAcpAgent({ host, harness }), async (ctx) => { + await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'remove a.txt' }], + }) + await vi.waitFor(() => expect(permissions).toHaveLength(1)) + // The editor still asks when the next prompt comes. + await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'hello' }], + }) + allow() + await vi.waitFor(() => + expect( + states(sessionId).filter((state) => state === 'idle:end_turn'), + ).toHaveLength(2), + ) + expect(states(sessionId)).toEqual([ + 'running', + 'requires_action', + 'running', + 'idle:end_turn', + 'running', + 'idle:end_turn', + ]) + expect(textOf(sessionId)).toBe('Kept a.txt.echo: hello') + }) + expect(remove).toHaveBeenCalledTimes(1) + }) + + it('stops a prompt when the editor cancels the turn that closes an approval', async () => { + const { harness, host, editor, editorState, states, textOf } = setup() + close = () => host.close() + editorState.answer = async () => ({ outcome: 'cancelled' }) + await editor.connectWith(createAcpAgent({ host, harness }), async (ctx) => { + await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'remove a.txt slowly' }], + }) + await vi.waitFor(() => + expect(states(sessionId)).toContain('idle:cancelled'), + ) + // This prompt first closes the approval. That turn waits, and the + // editor cancels it. + await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'hello' }], + }) + const session = await host.open(harness, { threadId: sessionId }) + await vi.waitFor(() => expect(session.snapshot().status).toBe('running')) + await ctx.notify('session/cancel', { sessionId }) + await vi.waitFor(() => + expect( + states(sessionId).filter((state) => state === 'idle:cancelled'), + ).toHaveLength(2), + ) + expect(states(sessionId)).toEqual([ + 'running', + 'requires_action', + 'idle:cancelled', + 'running', + 'idle:cancelled', + ]) + expect(textOf(sessionId)).toBe('') + }) + }) + + it('fails the turn when the session refuses the approval answers', async () => { + const { harness, host, editor, editorState, idle } = setup() + close = () => host.close() + editorState.answer = async () => ({ + outcome: 'selected', + optionId: 'allow', + }) + await editor.connectWith(createAcpAgent({ host, harness }), async (ctx) => { + await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + const session = await host.open(harness, { threadId: sessionId }) + vi.spyOn(session, 'resolve').mockResolvedValue({ + inputId: 'input-1', + status: 'rejected', + reason: 'busy', + }) + await ctx.request('session/prompt', { + sessionId, + prompt: [{ type: 'text', text: 'remove a.txt' }], + }) + await vi.waitFor(() => expect(idle(sessionId)).toBeDefined()) + expect(idle(sessionId)).toMatchObject({ + stopReason: 'error', + error: { message: 'A chat turn is already running.' }, + }) + }) + }) + + it('refuses an unknown session and data that is not base64', async () => { + const { harness, host, editor } = setup() + close = () => host.close() + await editor.connectWith(createAcpAgent({ host, harness }), async (ctx) => { + await ctx.request('initialize', { + protocolVersion: 2, + info: { name: 'editor', version: '1.0.0' }, + }) + await expect( + ctx.request('session/resume', { sessionId: 'never-made', cwd: '/tmp' }), + ).rejects.toMatchObject({ code: -32002 }) + await expect( + ctx.request('session/prompt', { + sessionId: 'never-made', + prompt: [{ type: 'text', text: 'hello' }], + }), + ).rejects.toMatchObject({ code: -32002 }) + const { sessionId } = await ctx.request('session/new', { + cwd: '/tmp', + mcpServers: [], + }) + await expect( + ctx.request('session/prompt', { + sessionId, + prompt: [ + { type: 'image', data: 'not base64!', mimeType: 'image/png' }, + ], + }), + ).rejects.toMatchObject({ code: -32602 }) + }) + }) + it('resumes a session by id, cancels a running turn, and closes it', async () => { const { harness, host, editor, idle, running, textOf } = setup() close = () => host.close() + // Sessions from before: the session index knows them. + await host.open(harness, { threadId: 'kept-session' }) + await host.open(harness, { threadId: 'slow-session' }) await editor.connectWith(createAcpAgent({ host, harness }), async (ctx) => { await ctx.request('initialize', { protocolVersion: 2, diff --git a/packages/ai-acp/tests/agent-media.test.ts b/packages/ai-acp/tests/agent-media.test.ts index 5dbee7a619..fe9df9b4ba 100644 --- a/packages/ai-acp/tests/agent-media.test.ts +++ b/packages/ai-acp/tests/agent-media.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it } from 'vitest' +import { afterEach, describe, expect, it, vi } from 'vitest' import { EventType } from '@tanstack/ai' import { client } from '@agentclientprotocol/sdk/experimental/v2' import { memoryPersistence } from '@tanstack/ai-persistence' @@ -124,16 +124,16 @@ async function savedPrompt(prompt: Array) { cwd: '/tmp', mcpServers: [], }) - const { messageId } = await ctx.request('session/prompt', { - sessionId, - prompt, - }) - return { sessionId, messageId } + await ctx.request('session/prompt', { sessionId, prompt }) + return { sessionId } }) const session = await host.open(harness, { threadId: sent.sessionId }) - await session.operation(sent.messageId) - const [user] = await session.transcript() - if (user?.role !== 'user') throw new Error('The turn saved no user message.') + // The prompt goes to the harness after the response, so wait for it. + const user = await vi.waitFor(async () => { + const [first] = await session.transcript() + if (first?.role !== 'user') throw new Error('No user message yet.') + return first + }) return { session, user } } diff --git a/packages/ai-acp/tests/agent.test.ts b/packages/ai-acp/tests/agent.test.ts index 51a7db4590..097478a367 100644 --- a/packages/ai-acp/tests/agent.test.ts +++ b/packages/ai-acp/tests/agent.test.ts @@ -140,7 +140,7 @@ describe('ACP v2 agent', () => { sessionId, prompt: [{ type: 'text', text: 'remove a.txt' }], }) - expect(prompted.messageId).toMatch(/^op-chat-/) + expect(prompted.messageId).toMatch(/^user-/) await vi.waitFor(() => expect(