Add standalone Agent Client Protocol client layer (#24990)

* Add standalone ACP protocol client and session runtime

* Protect ACP transport teardown from late stream errors

* Retire incoming ACP request ids before publishing responses

* Narrow ACP configuration requests and transport message types

* Remove redundant ACP request handler return unions

* Keep ACP waits caller-owned and preserve protocol extensions

* Preserve open ACP decisions through prompt completion

* Generate open ACP enums and check the generated schema offline

A newer or vendor enum value (tool kind, tool status, option kind, stop
reason) no longer fails the whole message: generated enums accept the known
literals plus any other string, typed so callers can still narrow on the
known ones. The generated header now records the pinned input digests, the
generator digest and a body hash, so `verify:acp-protocol` catches a stale or
hand-edited file without network access; it runs in lint and the PR workflow.

* Land the ACP runtime contract the agent adapters use

- Deliver notifications other than session/update through
  onExtensionNotification, in arrival order with session updates.
- Accept _meta on prompt, setMode, setModel, setConfigOption and cancel.
- cancel() always sends session/cancel once the session runs, since the
  agent can be in a turn it began itself; only a successful send is shared,
  so a failed write is retried.
- Cancel aborts each open agent request's signal and lets its handler send
  its own answer; -32800 only when the handler rejects.
- Permission requests validate only the session, tool call id and options;
  unreadable fields are dropped with a diagnostic, and any answer Orca
  cannot send is `cancelled` instead of a JSON-RPC error. Agent-started
  turns may ask; whether to show it is the caller's decision.
- AcpAgentError marks the agent's own errors; AcpInvalidResponseError keeps
  the raw answer and validation issues for answers Orca could not read.
- Lines over the size limit are classified by prefix (shared with the Codex
  reader): the owed request fails, an oversized agent request is answered
  with an error, and an unattributable response closes the connection.

* Answer every agent request after an ACP cancel

A cancel that lands before a permission handler starts now still runs the
permission path, so the agent gets the `cancelled` outcome rather than a
request-cancelled error. A handler that ignores the abort no longer leaves
the agent waiting: once the abort has run through, any request still
unanswered gets request-cancelled. Handlers that answer on abort keep their
own reply.

Also renames a lint-rejected helper parameter, replaces a Reflect.apply in a
test, and stops the permission diagnostic from firing with an empty list.

* Let each ACP request handler own its answer after a cancel

Removes the next-event-loop-turn fallback that answered request-cancelled
for any handler still silent after a cancel. It raced answers that were
still being saved (an approval mid-journal-write reached the agent as an
error) and made the outcome depend on event-loop timing. The handler that
owns an agent request now always sends its answer, or throws for
request-cancelled; a request it never answers ends when the connection
closes. A permission whose handler had not started still answers
`cancelled`.

* Register the ACP schema verify step in the PR preflight phase test

* feat(acp): a steer's cancel asks once and never ends the agent

The runtime had one cancel: send session/cancel, wait at most 10 s for Orca's prompt to settle,
then close the connection, which ends the agent. A steer used it too, so a slow agent lost its
process just because the person added a message. requestSteerCancel() now sends session/cancel
once per prompt, cancels the agent's open requests and answers later permissions cancelled, and
never bounds or closes: the prompt's own reply ends it and the steer's prompt follows. cancel()
stays the Stop: bounded, then close. A Stop after a steer still bounds and closes. Both cancel
paths move into acp-prompt-cancel.ts over one cancel channel.

* fix(acp): a repeated steer shares the cancel in flight; say what the caller owns

Per review: a second steer before the first write lands returns that write instead of resolving
early. The steer's JSDoc says the wait for the prompt's reply is unbounded and that a prompt that
fails instead must not take the steer until the caller rebuilds the session; the Stop's says a
prompt that settles in time leaves the agent for the Stop's owner to end. The steer test now gives
the runtime a handler that would allow: the open permission's signal aborts and the late one never
reaches it.

* test(ratchet): require src/main/acp now that this PR lands it
This commit is contained in:
Brennan Benson
2026-10-05 22:25:40 -07:00
committed by GitHub
parent 635e7f53cb
commit 020cebeff6
30 changed files with 4772 additions and 22 deletions
+3
View File
@@ -88,3 +88,6 @@
/mobile/web-entry/**/*.otf -text
/mobile/web-entry/**/*.woff -text
/mobile/web-entry/**/*.woff2 -text
# Generated ACP schemas are checked byte-for-byte against formatted output.
/src/main/acp/generated/*.generated.ts linguist-generated=true text eol=lf
+4
View File
@@ -349,6 +349,10 @@ jobs:
if: needs.code_paths.outputs.static_analysis == 'true'
run: pnpm run verify:rpc-params-catalog
- name: Verify the generated ACP protocol schema
if: needs.code_paths.outputs.static_analysis == 'true'
run: pnpm run verify:acp-protocol
- name: Verify bundled skill guides
if: needs.code_paths.outputs.static_analysis == 'true'
run: pnpm run verify:bundled-skill-guides
+1
View File
@@ -208,6 +208,7 @@
}
],
"ignorePatterns": [
"src/main/acp/generated/acp-protocol.generated.ts",
"src/shared/rpc-contract/rpc-params-catalog.generated.ts",
"**/node_modules",
"**/dist",
+294
View File
@@ -0,0 +1,294 @@
import { createHash } from 'node:crypto'
import { mkdir, readdir, readFile, unlink, writeFile } from 'node:fs/promises'
import { fileURLToPath } from 'node:url'
import { resolve } from 'node:path'
const release = 'schema-v1.21.0'
const legacyRelease = 'v0.11.6'
const repository = 'https://github.com/agentclientprotocol/agent-client-protocol'
const output = fileURLToPath(new URL('../../../src/main/acp/generated/', import.meta.url))
const outputFile = 'acp-protocol.generated.ts'
// --check is offline (lint/CI): the header records every input digest and the body hash.
// --check-online regenerates from the pinned downloads and compares byte-for-byte.
const checkOnline = process.argv.includes('--check-online')
const check = checkOnline || process.argv.includes('--check')
const definitions = {}
const inputs = {
schema: '7f77702b34e0a0558e77220e9007bf8ee161a976bb8ac5021aba1b7e7b2c5708',
legacy: 'b3cf8687d979c98c009f0fbcf8f0c237645b82ac2d2f0a3ebca683f963c3d581',
license: 'f250d08cee4549b22b3b4aaaf3a743473336fd280316df5d0340717e5127a221'
}
const sha256 = (text) => createHash('sha256').update(text.replace(/\r\n/g, '\n')).digest('hex')
const generatorDigest = sha256(await readFile(import.meta.filename, 'utf8'))
const header = `// Generated by config/scripts/acp/generate-protocol.mjs; do not edit. Regenerate: pnpm run generate:acp-protocol\n// ACP ${release}, legacy model API ${legacyRelease}; SPDX-License-Identifier: Apache-2.0.\n// Inputs sha256: schema ${inputs.schema}, legacy ${inputs.legacy}, license ${inputs.license}, generator ${generatorDigest}\n`
const bodyDigestPrefix = '// Body sha256: '
async function checkOffline() {
const text = (await readFile(resolve(output, outputFile), 'utf8')).replace(/\r\n/g, '\n')
if (!text.startsWith(header)) {
throw new Error(`Stale generated file: ${outputFile} (inputs or generator changed; regenerate)`)
}
const rest = text.slice(header.length)
const newline = rest.indexOf('\n')
if (!rest.startsWith(bodyDigestPrefix) || newline === -1) {
throw new Error(`Stale generated file: ${outputFile} (missing body digest)`)
}
if (rest.slice(bodyDigestPrefix.length, newline) !== sha256(rest.slice(newline + 1))) {
throw new Error(`Generated file was edited by hand: ${outputFile}`)
}
for (const file of await readdir(output)) {
if ((file.endsWith('.gen.ts') || file.endsWith('.generated.ts')) && file !== outputFile) {
throw new Error(`Unexpected generated file: ${file}`)
}
}
console.log(`Checked ${outputFile} against pinned inputs offline`)
}
if (check && !checkOnline) {
await checkOffline()
process.exit(0)
}
async function download(url, digest) {
const response = await fetch(url)
if (!response.ok) {
throw new Error(`Download failed: ${url} (${response.status})`)
}
const text = await response.text()
if (createHash('sha256').update(text).digest('hex') !== digest) {
throw new Error(`Upstream content changed: ${url}`)
}
return text
}
const [schema, legacy, license] = await Promise.all([
download(`${repository}/releases/download/${release}/schema.unstable.json`, inputs.schema),
download(`${repository}/releases/download/${legacyRelease}/schema.unstable.json`, inputs.legacy),
download(
`https://raw.githubusercontent.com/agentclientprotocol/agent-client-protocol/${release}/LICENSE`,
inputs.license
)
])
Object.assign(definitions, JSON.parse(legacy).$defs, JSON.parse(schema).$defs)
const roots = [
'InitializeRequest',
'InitializeResponse',
'AuthenticateRequest',
'AuthenticateResponse',
'NewSessionRequest',
'NewSessionResponse',
'LoadSessionRequest',
'LoadSessionResponse',
'ResumeSessionRequest',
'ResumeSessionResponse',
'PromptRequest',
'PromptResponse',
'CancelNotification',
'SessionNotification',
'RequestPermissionRequest',
'RequestPermissionResponse',
'SetSessionModeRequest',
'SetSessionModeResponse',
'SetSessionModelRequest',
'SetSessionModelResponse',
'SessionModelState',
'SetSessionConfigOptionRequest',
'SetSessionConfigOptionResponse',
'ReadTextFileRequest',
'ReadTextFileResponse',
'WriteTextFileRequest',
'WriteTextFileResponse',
'CreateTerminalRequest',
'CreateTerminalResponse',
'TerminalOutputRequest',
'TerminalOutputResponse',
'ReleaseTerminalRequest',
'ReleaseTerminalResponse',
'WaitForTerminalExitRequest',
'WaitForTerminalExitResponse',
'KillTerminalRequest',
'KillTerminalResponse'
]
function references(value) {
if (!value || typeof value !== 'object') {
return []
}
if (Array.isArray(value)) {
return value.flatMap(references)
}
return [
...(value.$ref ? [value.$ref.split('/').at(-1)] : []),
...Object.values(value).flatMap(references)
]
}
const ordered = []
const visiting = new Set()
const visited = new Set()
function visit(name) {
if (visited.has(name)) {
return
}
if (visiting.has(name)) {
throw new Error(`Recursive schema needs an explicit type: ${name}`)
}
if (!definitions[name]) {
throw new Error(`Missing definition: ${name}`)
}
visiting.add(name)
for (const dependency of references(definitions[name])) {
visit(dependency)
}
visiting.delete(name)
visited.add(name)
ordered.push(name)
}
roots.forEach(visit)
// A named string enum: two or more string constants, optionally with an open `string` member.
function isStringEnum(value) {
const alternatives = value.oneOf ?? value.anyOf
return (
Array.isArray(alternatives) &&
alternatives.filter((alternative) => typeof alternative.const === 'string').length > 1 &&
alternatives.every(
(alternative) =>
typeof alternative.const === 'string' ||
(alternative.type === 'string' &&
Object.keys(alternative).every((key) => ['type', 'title', 'description'].includes(key)))
)
)
}
// Enums stay open so a newer or vendor value reaches the caller instead of failing the message.
function openEnum(value) {
const known = (value.oneOf ?? value.anyOf).filter(
(alternative) => typeof alternative.const === 'string'
)
return `z.union([${known.map((alternative) => `z.literal(${JSON.stringify(alternative.const)})`).join(',')},otherString])`
}
function expression(value) {
if (value === true) {
return 'z.unknown()'
}
if (value === false) {
return 'z.never()'
}
if (value.$ref) {
return `${value.$ref.split('/').at(-1)}Schema`
}
if ('const' in value) {
return `z.literal(${JSON.stringify(value.const)})`
}
const alternatives = value.oneOf ?? value.anyOf
if (alternatives || value.allOf) {
const combined = alternatives
? `z.union([${alternatives.map(expression).join(',')}])`
: value.allOf.map(expression).reduce((left, right) => `z.intersection(${left},${right})`)
const siblings = { ...value }
delete siblings.oneOf
delete siblings.anyOf
delete siblings.allOf
return siblings.type || siblings.properties
? `z.intersection(${expression(siblings)},${combined})`
: combined
}
if (Array.isArray(value.type)) {
return `z.union([${value.type.map((type) => expression({ ...value, type })).join(',')}])`
}
let result
switch (value.type) {
case 'string':
result = 'z.string()'
break
case 'integer':
result = 'z.number().int()'
break
case 'number':
result = 'z.number()'
break
case 'boolean':
result = 'z.boolean()'
break
case 'null':
result = 'z.null()'
break
case 'array':
result = `z.array(${expression(value.items ?? true)})`
break
case 'object': {
const properties = Object.entries(value.properties ?? {}).map(
([key, property]) =>
`${JSON.stringify(key)}:${expression(property)}${value.required?.includes(key) ? '' : '.optional()'}`
)
result = `z.${value.additionalProperties === false ? 'strictObject' : 'looseObject'}({${properties.join(',')}})`
if (typeof value.additionalProperties === 'object') {
result += `.catchall(${expression(value.additionalProperties)})`
}
break
}
default:
if (
Object.keys(value).some(
(key) => !key.startsWith('x-') && !['description', 'title', 'default'].includes(key)
)
) {
throw new Error(`Unsupported schema: ${JSON.stringify(value)}`)
}
result = 'z.unknown()'
}
if (['integer', 'number'].includes(value.type) && typeof value.minimum === 'number') {
result += `.min(${value.minimum})`
}
if (['integer', 'number'].includes(value.type) && typeof value.maximum === 'number') {
result += `.max(${value.maximum})`
}
if (value.not) {
result += `.refine(value=>!${expression(value.not)}.safeParse(value).success)`
}
return result
}
const source = `${header}/*\n${license.trim()}\n*/\nimport { z } from 'zod'\nexport const ACP_SCHEMA_RELEASE = '${release}'\nexport const ACP_LEGACY_MODEL_SCHEMA_RELEASE = '${legacyRelease}'\nexport const ACP_PROTOCOL_VERSION = 1\n// An enum value this schema release does not name; \`string & {}\` keeps the known literals narrowable.\nconst otherString = z.custom<string & {}>((value) => typeof value === 'string')\n${ordered
.map(
(name) =>
`export const ${name}Schema = ${isStringEnum(definitions[name]) ? openEnum(definitions[name]) : expression(definitions[name])}\nexport type ${name} = z.infer<typeof ${name}Schema>\n`
)
.join('\n')}`
// Use the repository formatter without spawning a platform-dependent executable shim.
const { format } = await import('oxfmt')
const formatted = await format(outputFile, source, {
singleQuote: true,
semi: false,
printWidth: 100,
trailingComma: 'none'
})
if (formatted.errors.length || !formatted.code.startsWith(header)) {
throw new Error(`Formatting failed for ${outputFile}`)
}
const body = formatted.code.slice(header.length)
const code = `${header}${bodyDigestPrefix}${sha256(body)}\n${body}`
await mkdir(output, { recursive: true })
const destination = resolve(output, outputFile)
if (check) {
if ((await readFile(destination, 'utf8')) !== code) {
throw new Error(`Stale generated file: ${outputFile}`)
}
} else {
await writeFile(destination, code)
}
for (const file of await readdir(output)) {
if ((file.endsWith('.gen.ts') || file.endsWith('.generated.ts')) && file !== outputFile) {
const generatedHere = (await readFile(resolve(output, file), 'utf8')).startsWith(
'// Generated by config/scripts/acp/generate-protocol.mjs'
)
if (check || !generatedHere) {
throw new Error(`Unexpected generated file: ${file}`)
}
await unlink(resolve(output, file))
}
}
console.log(`${check ? 'Checked' : 'Generated'} ${ordered.length} ACP definitions`)
@@ -49,17 +49,13 @@ export const STRUCTURED_CHAT_LANES = [
{ directory: ['src', 'shared'] },
{ directory: ['src', 'main', 'runtime'], basename: /^(?:structured-|agent-session-)/ },
{ directory: ['src', 'main', 'provider-process'] },
// Allowed absent until it lands; every other lane throws if missing, so a rename can't empty it.
{ directory: ['src', 'main', 'acp'], mayBeAbsent: true }
{ directory: ['src', 'main', 'acp'] }
]
export function collectStructuredChatEntryPoints(root = ROOT) {
return STRUCTURED_CHAT_LANES.flatMap((lane) => {
const directory = path.join(root, ...lane.directory)
if (!existsSync(directory)) {
if (lane.mayBeAbsent) {
return []
}
throw new Error(
`[runtime-electron-ratchet] ${lane.directory.join('/')} is missing. If it moved, update STRUCTURED_CHAT_LANES; otherwise the gate would silently check nothing there.`
)
@@ -1,5 +1,5 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import path from 'node:path'
import process from 'node:process'
@@ -9,8 +9,7 @@ import {
defaultEntryPoints,
diffAgainstBaseline,
main,
readBaseline,
STRUCTURED_CHAT_LANES
readBaseline
} from './check-runtime-electron-ratchet.mjs'
describe('structured chat coverage', () => {
@@ -32,13 +31,14 @@ describe('structured chat coverage', () => {
return root
}
// Every lane that must exist; acp/ may be absent until it lands.
// Every lane that must exist.
const requiredLanes = {
'src/main/native-chat/reader.ts': 'export {}',
'src/main/claude/claude-session.ts': 'export {}',
'src/main/codex/codex-session.ts': 'export {}',
'src/main/runtime/structured-agent-session-host.ts': 'export {}',
'src/main/provider-process/provider-process-teardown.ts': 'export {}',
'src/main/acp/acp-structured-session-adapter.ts': 'export {}',
'src/shared/agent-session-record.ts': 'export {}'
}
@@ -90,7 +90,7 @@ describe('structured chat coverage', () => {
Object.entries(requiredLanes).filter(([file]) => !file.startsWith(`${lane}/`))
)
expect(() => collectStructuredChatEntryPoints(fixture(without))).toThrow(`${lane} is missing`)
expect(collectStructuredChatEntryPoints(fixture(requiredLanes))).toHaveLength(6)
expect(collectStructuredChatEntryPoints(fixture(requiredLanes))).toHaveLength(7)
}
)
@@ -128,16 +128,6 @@ describe('the default entry points', () => {
expect(entries.some((file) => file.startsWith(lane))).toBe(true)
}
})
// Retires the temporary flag: the PR that adds acp/ must make it required.
it('lets only directories that have not landed yet be absent', () => {
for (const lane of STRUCTURED_CHAT_LANES.filter((candidate) => candidate.mayBeAbsent)) {
expect(
existsSync(path.join(process.cwd(), ...lane.directory)),
lane.directory.join('/')
).toBe(false)
}
})
})
// Why `main`: it is what `pnpm lint` and CI run, so these fail if its entry list drops the lanes.
@@ -200,6 +200,7 @@ it('pins every foreground and background step to its selected phase', () => {
['Check Node runtime pin', staticPhase],
['Boot orcad and round-trip a terminal', staticPhase],
['Verify the generated RPC params catalog', staticPhase],
['Verify the generated ACP protocol schema', staticPhase],
['Verify bundled skill guides', staticPhase],
['Verify skill freshness manifest', staticPhase],
['Verify localization coverage', staticPhase],
+3 -1
View File
@@ -14,7 +14,7 @@
"audit:perf": "oxlint --config config/oxlint-performance-audit.json --format json src",
"test:perf:contracts": "vitest run --config config/vitest.performance.config.ts",
"format": "oxfmt --write .",
"lint": "oxlint && pnpm run audit:anti-slop && pnpm run audit:code-quality:native && pnpm run audit:code-quality:type-aware && pnpm run check:reliability-gates && pnpm run check:dead-classes && pnpm run check:max-lines-ratchet && pnpm run check:ts-nocheck-ratchet && pnpm run check:runtime-electron-ratchet && pnpm run check:readme-local-links && pnpm run check:node-runtime-pin && pnpm run verify:rpc-params-catalog && pnpm run verify:bundled-skill-guides && pnpm run verify:skill-bundle-manifest && pnpm run verify:localization-catalogs && pnpm run verify:localization-extraction && pnpm run verify:localization-coverage",
"lint": "oxlint && pnpm run audit:anti-slop && pnpm run audit:code-quality:native && pnpm run audit:code-quality:type-aware && pnpm run check:reliability-gates && pnpm run check:dead-classes && pnpm run check:max-lines-ratchet && pnpm run check:ts-nocheck-ratchet && pnpm run check:runtime-electron-ratchet && pnpm run check:readme-local-links && pnpm run check:node-runtime-pin && pnpm run verify:rpc-params-catalog && pnpm run verify:acp-protocol && pnpm run verify:bundled-skill-guides && pnpm run verify:skill-bundle-manifest && pnpm run verify:localization-catalogs && pnpm run verify:localization-extraction && pnpm run verify:localization-coverage",
"audit:code-quality": "pnpm run audit:code-quality:native && pnpm run audit:code-quality:type-aware && pnpm run audit:react-doctor",
"audit:code-quality:native": "oxlint --config config/oxlint-code-quality-native-plugins.json src config tests mobile --deny-warnings",
"audit:code-quality:type-aware": "oxlint --type-aware --config config/oxlint-code-quality-type-aware.json src config tests --deny-warnings",
@@ -50,6 +50,8 @@
"check:feature-wall-assets": "node config/scripts/check-feature-wall-assets.mjs",
"generate:rpc-params-catalog": "node config/scripts/generate-rpc-params-catalog.mjs",
"verify:rpc-params-catalog": "node config/scripts/generate-rpc-params-catalog.mjs --check",
"generate:acp-protocol": "node config/scripts/acp/generate-protocol.mjs",
"verify:acp-protocol": "node config/scripts/acp/generate-protocol.mjs --check",
"generate:bundled-skill-guides": "node config/scripts/generate-bundled-skill-guides.mjs --write",
"verify:bundled-skill-guides": "node config/scripts/generate-bundled-skill-guides.mjs --check",
"generate:skill-bundle-manifest": "node config/scripts/generate-skill-bundle-manifest.mjs --write",
+72
View File
@@ -0,0 +1,72 @@
import type { AuthMethod } from './generated/acp-protocol.generated'
export class AcpRpcError extends Error {
constructor(
readonly code: number,
message: string,
readonly data?: unknown
) {
super(message)
this.name = 'AcpRpcError'
}
}
/** The agent's own error answer to a request: it read the request and refused it. Errors Orca
* raises about a request (a timeout, an unreadable answer, a closed connection) are not this. */
export class AcpAgentError extends AcpRpcError {
constructor(code: number, message: string, data?: unknown) {
super(code, message, data)
this.name = 'AcpAgentError'
}
}
export class AcpAuthRequiredError extends AcpAgentError {
constructor(
message: string,
data?: unknown,
readonly authMethods: AuthMethod[] = []
) {
super(-32000, message, data)
this.name = 'AcpAuthRequiredError'
}
}
/** Orca could not read the agent's answer; `data` keeps the raw answer, `issues` why it failed. */
export class AcpInvalidResponseError extends AcpRpcError {
constructor(
message: string,
raw: unknown,
readonly issues?: unknown
) {
super(-32603, message, raw)
this.name = 'AcpInvalidResponseError'
}
}
/** A line from the agent exceeded the framing limit, so the message it carried was never read. */
export class AcpFrameTooLargeError extends Error {
constructor(
readonly method: string | null,
readonly observedBytes: number,
readonly maxBytes: number
) {
super(
`ACP${method ? ` ${method}` : ''} message exceeds ${maxBytes} byte limit (${observedBytes} bytes received)`
)
this.name = 'AcpFrameTooLargeError'
}
}
export class AcpConnectionClosedError extends Error {
constructor(message = 'ACP connection closed') {
super(message)
this.name = 'AcpConnectionClosedError'
}
}
export class AcpRequestTimeoutError extends Error {
constructor(readonly method: string) {
super(`ACP request timed out: ${method}`)
this.name = 'AcpRequestTimeoutError'
}
}
+119
View File
@@ -0,0 +1,119 @@
import { AcpRpcError } from './acp-errors'
import type { AcpJsonRpcMessage, AcpPeerHandlers } from './acp-json-rpc-peer'
type OpenRequest = {
controller: AbortController
abandon: () => void
closed: boolean
cancelled: boolean
}
export class AcpIncomingRequests {
private readonly open = new Map<string | number | null, OpenRequest>()
constructor(
private readonly handler: AcpPeerHandlers['onRequest'],
private readonly send: (message: AcpJsonRpcMessage) => Promise<void>,
private readonly onFailure: (error: Error) => void,
private readonly capacity: number,
private readonly diagnose: (message: string) => void
) {}
close(error: Error): void {
for (const request of this.open.values()) {
request.closed = true
request.controller.abort(error)
request.abandon()
}
this.open.clear()
}
// Each handler answers its own request (a permission answers `cancelled`); -32800 only if it
// throws. The runtime never answers for a live handler: an answer still being saved must win.
cancel(): void {
for (const request of this.open.values()) {
if (!request.cancelled) {
request.cancelled = true
request.controller.abort(new AcpRpcError(-32800, 'Request cancelled'))
}
}
}
handle(id: string | number | null, method: string, params: unknown): void {
if (this.open.has(id)) {
this.diagnose('Ignored duplicate ACP incoming request id')
return
}
if (this.open.size >= this.capacity) {
this.refuse(id, new AcpRpcError(-32603, 'ACP incoming request capacity exceeded'))
return
}
const controller = new AbortController()
let abandon = (): void => {}
const abandoned = new Promise<never>((_resolve, reject) => {
abandon = () => reject(controller.signal.reason)
})
const request: OpenRequest = { controller, abandon, closed: false, cancelled: false }
this.open.set(id, request)
const retire = (): void => {
if (this.open.get(id) === request) {
this.open.delete(id)
}
}
void Promise.race([
abandoned,
Promise.resolve().then(() => {
// A cancelled request still reaches its handler, so a permission can answer `cancelled`.
if (request.closed) {
throw controller.signal.reason
}
if (!this.handler) {
throw new AcpRpcError(-32601, `Unknown ACP client method: ${method}`)
}
return this.handler(method, params, { id, signal: controller.signal })
})
])
.then(async (result) => {
if (request.closed) {
return
}
// The agent may reuse the id as soon as it reads the response.
retire()
await this.send({ jsonrpc: '2.0', id, result: result ?? null })
})
.catch(async (error) => {
if (request.closed) {
return
}
retire()
await this.sendError(
id,
request.cancelled
? new AcpRpcError(-32800, 'Request cancelled')
: error instanceof AcpRpcError
? error
: new AcpRpcError(-32603, error instanceof Error ? error.message : String(error))
)
})
.finally(() => {
retire()
controller.abort()
})
}
refuse(id: string | number | null, error: AcpRpcError): void {
void this.sendError(id, error)
}
private async sendError(id: string | number | null, error: AcpRpcError): Promise<void> {
try {
await this.send({
jsonrpc: '2.0',
id,
error: { code: error.code, message: error.message, data: error.data }
})
} catch (failure) {
this.onFailure(failure instanceof Error ? failure : new Error(String(failure)))
}
}
}
+318
View File
@@ -0,0 +1,318 @@
import { PassThrough, Writable } from 'node:stream'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { AcpJsonRpcPeer, type AcpPeerHandlers, type AcpPeerOptions } from './acp-json-rpc-peer'
import {
AcpConnectionClosedError,
AcpFrameTooLargeError,
AcpRequestTimeoutError,
AcpRpcError
} from './acp-errors'
import { AcpScriptedAgent, deferred, tick } from './acp-scripted-agent.test-support'
const peers: AcpJsonRpcPeer[] = []
const agents: AcpScriptedAgent[] = []
function fixture(handlers: AcpPeerHandlers = {}, options: AcpPeerOptions = {}) {
const agent = new AcpScriptedAgent()
const peer = new AcpJsonRpcPeer(agent.stdout, agent.stdin, handlers, options)
agents.push(agent)
peers.push(peer)
return { peer, agent }
}
afterEach(() => {
peers.splice(0).forEach((peer) => peer.close())
agents.splice(0).forEach((agent) => agent.close())
vi.useRealTimers()
})
describe('ACP JSON-RPC peer', () => {
it('routes interleaved requests in both directions without conflating id types', async () => {
const { peer, agent } = fixture({ onRequest: (method, params) => ({ method, params }) })
const first = peer.request('first', {})
const second = peer.request('second', {})
await tick()
expect(agent.frames.map((frame) => frame.id)).toEqual([1, 2])
agent.send({ jsonrpc: '2.0', id: '1', result: 'not the numeric id' })
const fromAgent = agent.request(1, '_question', { text: 'hello' })
agent.reply(agent.frames[1], 'second result')
agent.reply(agent.frames[0], 'first result')
expect(await second).toBe('second result')
expect(await first).toBe('first result')
expect(await fromAgent).toMatchObject({
id: 1,
result: { method: '_question', params: { text: 'hello' } }
})
expect(await agent.request(null, '_null_id', {})).toMatchObject({
id: null,
result: { method: '_null_id' }
})
})
it('returns method-not-found and preserves explicit handler error objects', async () => {
const { agent } = fixture({
onRequest: (method) => {
if (method === '_denied') {
throw new AcpRpcError(-32005, 'Denied', { reason: 'policy' })
}
if (method === '_crash') {
throw new Error('Handler failed')
}
throw new AcpRpcError(-32601, 'Unknown method')
}
})
expect(await agent.request('missing', '_missing', {})).toMatchObject({
error: { code: -32601 }
})
expect(await agent.request('denied', '_denied', {})).toMatchObject({
error: { code: -32005, message: 'Denied', data: { reason: 'policy' } }
})
expect(await agent.request('crash', '_crash', {})).toMatchObject({
error: { code: -32603, message: 'Handler failed' }
})
})
it('allows the agent to reuse a request id after receiving its response', async () => {
const { agent } = fixture({ onRequest: () => ({ answer: true }) })
expect(await agent.request('reused', '_question', {})).toMatchObject({
result: { answer: true }
})
expect(await agent.request('reused', '_question', {})).toMatchObject({
result: { answer: true }
})
})
it('ignores malformed and oversized lines then resumes at the next newline', async () => {
const diagnostics: string[] = []
const notified: unknown[] = []
const { peer, agent } = fixture(
{
onDiagnostic: (message) => diagnostics.push(message),
onNotification: (_method, params) => notified.push(params)
},
{ maxLineBytes: 100 }
)
agent.stdout.write('not json\n[]\n{"jsonrpc":"1.0","method":"bad"}\n')
for (let i = 0; i < 20; i++) {
agent.stdout.write('x'.repeat(40))
}
agent.stdout.write('\n')
const data = Buffer.from('{"jsonrpc":"2.0","method":"notice","params":"✓"}\r\n')
const split = data.indexOf(Buffer.from('✓')) + 1
agent.stdout.write(data.subarray(0, split))
agent.stdout.write(data.subarray(split))
agent.on('valid', (frame) => agent.reply(frame, 'ok'))
expect(await peer.request('valid', {})).toBe('ok')
expect(notified).toEqual(['✓'])
expect(diagnostics).toContain('Ignored ACP line: invalid-json')
expect(diagnostics).toContain('Ignored ACP line: line-too-long (unknown)')
expect(diagnostics).toContain('Ignored invalid ACP JSON-RPC envelope')
})
it('settles whoever was owed a message that exceeded the line limit', async () => {
const diagnostics: string[] = []
const { peer, agent } = fixture(
{ onDiagnostic: (message) => diagnostics.push(message), onRequest: () => 'unused' },
{ maxLineBytes: 200 }
)
const filler = 'x'.repeat(300)
const big = peer.request('session/load', {})
agent.stdout.write(`{"jsonrpc":"2.0","id":1,"result":{"history":"${filler}"}}\n`)
const lost = await big.catch((error: unknown) => error)
expect(lost).toBeInstanceOf(AcpFrameTooLargeError)
expect(lost).toMatchObject({ method: 'session/load', maxBytes: 200 })
const answer = new Promise<unknown>((resolve) => {
agent.stdin.on('data', (chunk: string) => resolve(JSON.parse(chunk)))
})
agent.stdout.write(
`{"jsonrpc":"2.0","id":"q","method":"session/request_permission","params":{"diff":"${filler}"}}\n`
)
expect(await answer).toMatchObject({ id: 'q', error: { code: -32600 } })
agent.stdout.write(`{"jsonrpc":"2.0","method":"session/update","params":{"x":"${filler}"}}\n`)
agent.stdout.write(`${filler}\n`)
await tick()
expect(peer.closed).toBe(false)
expect(diagnostics).toEqual([
'Ignored ACP line: line-too-long (response)',
'Ignored ACP line: line-too-long (server-request)',
'Ignored ACP line: line-too-long (notification)',
'Ignored ACP line: line-too-long (unknown)'
])
const stranded = peer.request('session/prompt', {})
agent.stdout.write(`{"jsonrpc":"2.0","id":2,"vendor":"${filler}"}\n`)
await expect(stranded).rejects.toBeInstanceOf(AcpFrameTooLargeError)
expect(peer.closed).toBe(true)
})
it('rejects malformed matching responses instead of leaving calls pending', async () => {
const { peer, agent } = fixture()
const rejected = expect(peer.request('wait', {})).rejects.toMatchObject({ code: -32603 })
agent.stdout.write('{"jsonrpc":"2.0","id":1}\n')
await rejected
const contradictory = expect(peer.request('wait', {})).rejects.toMatchObject({ code: -32603 })
agent.stdout.write('{"jsonrpc":"2.0","id":2,"result":"x","error":{"code":1,"message":"y"}}\n')
await contradictory
})
it('rejects all pending calls on exit and stops accepting requests', async () => {
const onClose = vi.fn()
const { peer, agent } = fixture({ onClose })
const first = expect(peer.request('first', {})).rejects.toBeInstanceOf(AcpConnectionClosedError)
const second = expect(peer.request('second', {})).rejects.toBeInstanceOf(
AcpConnectionClosedError
)
agent.stdout.end()
await Promise.all([first, second])
peer.close()
expect(onClose).toHaveBeenCalledTimes(1)
expect(agent.stdout.listenerCount('data')).toBe(0)
await expect(peer.request('late', {})).rejects.toBeInstanceOf(AcpConnectionClosedError)
})
it('bounds pending calls and frees capacity when a request times out', async () => {
vi.useFakeTimers()
const { peer, agent } = fixture({}, { maxPendingRequests: 1, requestTimeoutMs: 20 })
const timedOut = expect(peer.request('wait', {})).rejects.toBeInstanceOf(AcpRequestTimeoutError)
await expect(peer.request('overflow', {})).rejects.toThrow('capacity exceeded')
await vi.advanceTimersByTimeAsync(20)
await timedOut
agent.on('next', (frame) => agent.reply(frame, 'ok'))
expect(await peer.request('next', {})).toBe('ok')
expect(agent.frames.map((frame) => frame.method)).toEqual(['wait', 'next'])
})
it('bounds incoming requests without expiring user decisions', async () => {
vi.useFakeTimers()
const signal = deferred<AbortSignal>()
const answer = deferred<unknown>()
const { agent } = fixture(
{
onRequest: (_method, _params, context) => {
signal.resolve(context.signal)
return answer.promise
}
},
{ maxIncomingRequests: 1, requestTimeoutMs: 20 }
)
const first = agent.request('first', '_question', {})
const context = await signal.promise
expect(await agent.request('overflow', '_question', {})).toMatchObject({
error: { code: -32603 }
})
await vi.advanceTimersByTimeAsync(120_001)
expect(context.aborted).toBe(false)
answer.resolve({ answer: true })
expect(await first).toMatchObject({ result: { answer: true } })
expect(await agent.request('next', '_question', {})).toMatchObject({ result: { answer: true } })
})
it('aborts an in-flight incoming hook on close and never writes its late response', async () => {
const entered = deferred<AbortSignal>()
const result = deferred<unknown>()
const { peer, agent } = fixture({
onRequest: (_method, _params, context) => {
entered.resolve(context.signal)
return result.promise
}
})
void agent.request('open', '_question', {})
const signal = await entered.promise
peer.close()
expect(signal.aborted).toBe(true)
result.resolve({ late: true })
await tick()
expect(agent.frames).toEqual([])
})
it('serializes writes through backpressure and bounds queued output', async () => {
const writes: string[] = []
const callbacks: ((error?: Error | null) => void)[] = []
const output = new Writable({
highWaterMark: 1,
write(chunk, _encoding, callback) {
writes.push(chunk.toString())
callbacks.push(callback)
}
})
const input = new PassThrough()
const peer = new AcpJsonRpcPeer(input, output, {}, { maxQueuedWriteBytes: 170 })
peers.push(peer)
const first = peer.notify('one', {})
const second = peer.notify('two', {})
const third = peer.notify('three', {})
await expect(peer.notify('overflow', {})).rejects.toThrow('capacity exceeded')
expect(writes).toHaveLength(1)
callbacks[0]()
await first
expect(writes).toHaveLength(2)
callbacks[1]()
await second
callbacks[2]()
await third
expect(writes.map((line) => JSON.parse(line).method)).toEqual(['one', 'two', 'three'])
peer.close()
input.destroy()
output.destroy()
})
it('rejects active and queued writes when output closes', async () => {
const input = new PassThrough()
const output = new Writable({ write(_chunk, _encoding, _callback) {} })
const peer = new AcpJsonRpcPeer(input, output)
peers.push(peer)
const first = expect(peer.notify('one', {})).rejects.toBeInstanceOf(AcpConnectionClosedError)
const second = expect(peer.request('two', {})).rejects.toBeInstanceOf(AcpConnectionClosedError)
output.destroy()
await Promise.all([first, second])
input.destroy()
})
it('removes a timed-out queued request before it can reach the agent', async () => {
vi.useFakeTimers()
const writes: string[] = []
const callbacks: ((error?: Error | null) => void)[] = []
const input = new PassThrough()
const output = new Writable({
highWaterMark: 1,
write(chunk, _encoding, callback) {
writes.push(chunk.toString())
callbacks.push(callback)
}
})
const peer = new AcpJsonRpcPeer(input, output, {}, { requestTimeoutMs: 20 })
peers.push(peer)
const first = peer.notify('hold', {})
const timedOut = expect(peer.request('must-not-arrive', {})).rejects.toBeInstanceOf(
AcpRequestTimeoutError
)
await vi.advanceTimersByTimeAsync(20)
await timedOut
callbacks[0]()
await first
expect(writes.map((line) => JSON.parse(line).method)).toEqual(['hold'])
peer.close()
input.destroy()
output.destroy()
})
it('fails pending requests on an output stream error', async () => {
const { peer, agent } = fixture()
const rejected = expect(peer.request('wait', {})).rejects.toThrow('Broken pipe')
agent.stdin.emit('error', new Error('Broken pipe'))
await rejected
expect(peer.closed).toBe(true)
})
it('handles asynchronous write callback errors without an unhandled stream error', async () => {
const input = new PassThrough()
const output = new Writable({
write(_chunk, _encoding, callback) {
setImmediate(() => callback(new Error('Broken pipe from write')))
}
})
const peer = new AcpJsonRpcPeer(input, output)
peers.push(peer)
await expect(peer.request('wait', {})).rejects.toThrow('Broken pipe from write')
await tick()
expect(peer.closed).toBe(true)
input.destroy()
})
})
+304
View File
@@ -0,0 +1,304 @@
import type { Readable, Writable } from 'node:stream'
import { z } from 'zod'
import {
createIncrementalNdjsonFramer,
encodeNdjson
} from '../../shared/main-process-ndjson-framer'
import {
AcpAgentError,
AcpConnectionClosedError,
AcpInvalidResponseError,
AcpRequestTimeoutError
} from './acp-errors'
import { AcpIncomingRequests } from './acp-incoming-requests'
import { bounded, requestTimeout } from './acp-peer-limits'
import { settleOversizedAcpLine } from './acp-oversized-lines'
import { AcpWriteQueue } from './acp-write-queue'
import { detachAcpStreamErrorHandler } from './acp-stdio-error-boundary'
const idSchema = z.union([z.string(), z.number(), z.null()])
const errorSchema = z.object({
code: z.number().int(),
message: z.string(),
data: z.unknown().optional()
})
const envelopeSchema = z.looseObject({
jsonrpc: z.literal('2.0'),
id: idSchema.optional(),
method: z.string().optional(),
params: z.unknown().optional(),
result: z.unknown().optional(),
error: z.unknown().optional()
})
export type AcpJsonRpcMessage = z.infer<typeof envelopeSchema>
export type AcpRequestContext = { id: string | number | null; signal: AbortSignal }
export type AcpPeerHandlers = {
// Void means handled; unsupported methods must throw AcpRpcError(-32601). The handler owns its
// request: once the signal aborts it still answers, or throws (-32800); unanswered ends at close().
onRequest?: (method: string, params: unknown, context: AcpRequestContext) => unknown
onNotification?: (method: string, params: unknown) => void
onDiagnostic?: (message: string) => void
onClose?: (error: Error) => void
}
export type AcpPeerOptions = {
maxLineBytes?: number
maxQueuedWriteBytes?: number
maxPendingRequests?: number
maxIncomingRequests?: number
// Caller-selected outgoing deadlines only; omitted or null means unlimited.
requestTimeoutMs?: number | null
}
type Pending = {
method: string
resolve: (value: unknown) => void
reject: (error: Error) => void
timer?: ReturnType<typeof setTimeout>
}
export class AcpJsonRpcPeer {
private readonly pending = new Map<number, Pending>()
private readonly incoming: AcpIncomingRequests
private readonly writer: AcpWriteQueue
private readonly framer: ReturnType<typeof createIncrementalNdjsonFramer>
private nextId = 1
private terminalError?: Error
private readonly maxLineBytes: number
private readonly maxPending: number
private readonly maxIncoming: number
private readonly timeoutMs: number | null
constructor(
private readonly input: Readable,
private readonly output: Writable,
private readonly handlers: AcpPeerHandlers = {},
options: AcpPeerOptions = {}
) {
this.maxLineBytes = bounded(options.maxLineBytes, 16 * 1024 * 1024)
this.maxPending = bounded(options.maxPendingRequests, 128)
this.maxIncoming = bounded(options.maxIncomingRequests, 128)
this.timeoutMs = requestTimeout(options.requestTimeoutMs)
this.writer = new AcpWriteQueue(
output,
bounded(options.maxQueuedWriteBytes, 32 * 1024 * 1024),
(error) => this.close(error)
)
this.incoming = new AcpIncomingRequests(
handlers.onRequest,
(message) => this.send(message),
(error) => this.close(error),
this.maxIncoming,
(message) => this.diagnose(message)
)
this.framer = createIncrementalNdjsonFramer(
(record) => this.dispatch(record),
(rejected) =>
rejected.kind === 'line-too-long'
? settleOversizedAcpLine(rejected, {
rejectPending: (id, error) => this.rejectPending(id, error),
refuse: (id, error) => this.incoming.refuse(id, error),
close: (error) => this.close(error),
diagnose: (message) => this.diagnose(message)
})
: this.diagnose(`Ignored ACP line: ${rejected.kind}`),
{ maxLineBytes: this.maxLineBytes }
)
input.setEncoding('utf8')
input.on('data', this.onData)
input.on('end', this.onEnd)
input.on('close', this.onEnd)
input.on('error', this.onError)
output.on('close', this.onEnd)
output.on('finish', this.onEnd)
output.on('error', this.onError)
if (input.destroyed || input.readableEnded || output.destroyed || !output.writable) {
this.onEnd()
}
}
get closed(): boolean {
return this.terminalError !== undefined
}
request(
method: string,
params: unknown,
options: { timeoutMs?: number | null } = {}
): Promise<unknown> {
if (this.terminalError) {
return Promise.reject(this.terminalError)
}
if (this.pending.size >= this.maxPending) {
return Promise.reject(new Error('ACP pending request capacity exceeded'))
}
let timeoutMs: number | null
try {
timeoutMs = requestTimeout(
options.timeoutMs === undefined ? this.timeoutMs : options.timeoutMs
)
} catch (error) {
return Promise.reject(error)
}
const id = this.nextId++
const controller = new AbortController()
return new Promise((resolve, reject) => {
const timer =
timeoutMs === null
? undefined
: setTimeout(() => {
this.pending.delete(id)
const error = new AcpRequestTimeoutError(method)
controller.abort(error)
reject(error)
}, timeoutMs)
this.pending.set(id, { method, resolve, reject, timer })
void this.send({ jsonrpc: '2.0', id, method, params }, controller.signal).catch((error) => {
const pending = this.pending.get(id)
if (!pending) {
return
}
this.pending.delete(id)
clearTimeout(timer)
pending.reject(error instanceof Error ? error : new Error(String(error)))
})
})
}
notify(method: string, params: unknown): Promise<void> {
return this.send({ jsonrpc: '2.0', method, params })
}
/** Aborts every open agent request's signal; each handler still sends its own answer. */
cancelIncomingRequests(): void {
this.incoming.cancel()
}
close(error: Error = new AcpConnectionClosedError()): void {
if (this.terminalError) {
return
}
this.terminalError = error
this.input.removeListener('data', this.onData)
this.input.removeListener('end', this.onEnd)
this.input.removeListener('close', this.onEnd)
detachAcpStreamErrorHandler(this.input, this.onError)
this.output.removeListener('close', this.onEnd)
this.output.removeListener('finish', this.onEnd)
detachAcpStreamErrorHandler(this.output, this.onError)
this.framer.reset()
this.writer.close(error)
for (const pending of this.pending.values()) {
clearTimeout(pending.timer)
pending.reject(error)
}
this.pending.clear()
this.incoming.close(error)
try {
this.handlers.onClose?.(error)
} catch (failure) {
this.diagnose(String(failure))
}
}
private readonly onData = (chunk: string): void => {
try {
this.framer.feed(chunk)
} catch (error) {
this.close(error instanceof Error ? error : new Error(String(error)))
}
}
private readonly onEnd = (): void => this.close()
private readonly onError = (error: Error): void => this.close(error)
private diagnose(message: string): void {
try {
this.handlers.onDiagnostic?.(message)
} catch {
/* Diagnostics cannot break the transport. */
}
}
private send(message: AcpJsonRpcMessage, signal?: AbortSignal): Promise<void> {
if (this.terminalError) {
return Promise.reject(this.terminalError)
}
try {
return this.writer.write(encodeNdjson(message, this.maxLineBytes), signal)
} catch (error) {
return Promise.reject(error)
}
}
private rejectPending(id: number, error: (method: string) => Error): void {
const pending = this.pending.get(id)
if (pending) {
this.pending.delete(id)
clearTimeout(pending.timer)
pending.reject(error(pending.method))
}
}
private dispatch(record: unknown): void {
if (this.closed) {
return
}
const parsed = envelopeSchema.safeParse(record)
if (!parsed.success) {
this.diagnose('Ignored invalid ACP JSON-RPC envelope')
const response = z
.object({ id: z.number(), method: z.undefined().optional() })
.safeParse(record)
if (response.success) {
this.rejectPending(response.data.id, invalidEnvelope(record))
}
return
}
const frame = parsed.data
if (frame.method !== undefined) {
if ('result' in frame || 'error' in frame) {
this.diagnose('Ignored invalid ACP request')
return
}
if (frame.id !== undefined) {
this.incoming.handle(frame.id, frame.method, frame.params)
return
}
try {
this.handlers.onNotification?.(frame.method, frame.params)
} catch (error) {
this.diagnose(`ACP notification handler failed: ${String(error)}`)
}
return
}
if ('result' in frame === 'error' in frame) {
this.diagnose('Ignored invalid ACP response')
if (typeof frame.id === 'number') {
this.rejectPending(frame.id, invalidEnvelope(record))
}
return
}
if (typeof frame.id !== 'number') {
return
}
const pending = this.pending.get(frame.id)
if (!pending) {
return
}
this.pending.delete(frame.id)
clearTimeout(pending.timer)
if ('error' in frame) {
const parsedError = errorSchema.safeParse(frame.error)
if (!parsedError.success) {
this.diagnose('Invalid ACP error response')
pending.reject(new AcpInvalidResponseError('Invalid ACP error response', frame.error))
} else {
const error = parsedError.data
pending.reject(new AcpAgentError(error.code, error.message, error.data))
}
} else {
pending.resolve(frame.result)
}
}
}
function invalidEnvelope(raw: unknown): () => Error {
return () => new AcpInvalidResponseError('Invalid ACP response envelope', raw)
}
+38
View File
@@ -0,0 +1,38 @@
import { classifyJsonRpcPrefix } from '../../shared/json-rpc-record-prefix'
import type { NdjsonRejectedRecord } from '../../shared/main-process-ndjson-framer'
import { AcpFrameTooLargeError, AcpRpcError } from './acp-errors'
export type AcpOversizedLineTarget = {
rejectPending: (id: number, error: (method: string) => Error) => void
refuse: (id: string | number, error: AcpRpcError) => void
close: (error: Error) => void
diagnose: (message: string) => void
}
/** The Codex reader's prefix classification: settle whoever was owed the message that was lost. */
export function settleOversizedAcpLine(
rejected: NdjsonRejectedRecord & { kind: 'line-too-long' },
target: AcpOversizedLineTarget
): void {
const { observedBytes, maxLineBytes } = rejected
const record = classifyJsonRpcPrefix(rejected.prefix)
target.diagnose(`Ignored ACP line: line-too-long (${record.kind})`)
if (record.kind === 'server-request') {
target.refuse(
record.id,
new AcpRpcError(-32600, `ACP request exceeds ${maxLineBytes} byte limit`, {
method: record.method,
observedBytes
})
)
} else if (record.kind === 'response') {
target.rejectPending(
record.id,
(method) => new AcpFrameTooLargeError(method, observedBytes, maxLineBytes)
)
} else if (record.kind === 'response-unknown') {
// A numeric id with no readable result: any pending call could be the one that never settles.
target.close(new AcpFrameTooLargeError(null, observedBytes, maxLineBytes))
}
// A notification or non-JSON-RPC output (a stray log line) settles nothing; closing would end the session.
}
+21
View File
@@ -0,0 +1,21 @@
import { isSafeTimerDelayMs } from '../../shared/timer-delay'
export function bounded(value: number | undefined, fallback: number): number {
if (value === undefined) {
return fallback
}
if (!Number.isSafeInteger(value) || value <= 0) {
throw new Error('ACP limits must be positive finite integers')
}
return value
}
export function requestTimeout(value: number | null | undefined): number | null {
if (value == null) {
return null
}
if (!isSafeTimerDelayMs(value) || value <= 0) {
throw new Error('ACP timeouts must be positive finite timer durations')
}
return value
}
@@ -0,0 +1,160 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { AcpSessionRuntime, type AcpSessionRuntimeOptions } from './acp-session-runtime'
import { AcpScriptedAgent, deferred, tick } from './acp-scripted-agent.test-support'
import type { AcpPermissionHandler } from './acp-permission-requests'
import type { RequestPermissionRequest } from './generated/acp-protocol.generated'
const opened: { close: () => void }[] = []
const startOptions = { cwd: '/runtime/project', mcpServers: [] }
const allow = { optionId: 'allow', name: 'Allow', kind: 'allow_once' }
const toolCall = { toolCallId: 'tool-1', title: 'Edit file' }
function fixture(options: AcpSessionRuntimeOptions = {}) {
const agent = new AcpScriptedAgent()
const diagnostics: string[] = []
const runtime = new AcpSessionRuntime(agent.stdout, agent.stdin, {
onDiagnostic: (message) => diagnostics.push(message),
...options
})
opened.push(agent, runtime)
agent.on('initialize', (frame) => agent.reply(frame, { protocolVersion: 1 }))
agent.on('session/new', (frame) => agent.reply(frame, { sessionId: 'session-1' }))
agent.on('session/prompt', () => {})
return { agent, runtime, diagnostics }
}
afterEach(() =>
opened
.splice(0)
.toReversed()
.forEach((resource) => resource.close())
)
describe('ACP permission requests', () => {
it.each([
['a newer tool kind', { ...toolCall, kind: 'web_search' }, allow],
['a newer tool status', { ...toolCall, status: 'cancelled' }, allow],
['a newer option kind', toolCall, { ...allow, kind: 'allow_for_session' }]
])('delivers a permission with %s to the caller unchanged', async (_label, call, option) => {
const asked = vi.fn((_request: RequestPermissionRequest) => ({
outcome: { outcome: 'selected' as const, optionId: option.optionId }
}))
const { agent, runtime, diagnostics } = fixture({ onPermission: asked })
await runtime.start(startOptions)
void runtime.prompt([{ type: 'text', text: 'hi' }]).catch(() => {})
const params = { sessionId: 'session-1', toolCall: call, options: [option] }
expect(await agent.request('p', 'session/request_permission', params)).toMatchObject({
result: { outcome: { outcome: 'selected', optionId: option.optionId } }
})
expect(asked.mock.calls[0][0]).toEqual(params)
expect(diagnostics).toEqual([])
})
it('delivers a permission whose other fields are unreadable, dropping only those fields', async () => {
const asked = vi.fn((_request: RequestPermissionRequest) => ({
outcome: { outcome: 'cancelled' as const }
}))
const { agent, runtime, diagnostics } = fixture({ onPermission: asked })
await runtime.start(startOptions)
void runtime.prompt([{ type: 'text', text: 'hi' }]).catch(() => {})
const content = [{ type: 'content', content: { type: 'reasoning_summary', text: 'x' } }]
await agent.request('p', 'session/request_permission', {
sessionId: 'session-1',
toolCall: { ...toolCall, content, vendor: 'kept' },
options: [{ optionId: 'allow', kind: 'allow_once' }, { name: 'No id' }]
})
expect(asked.mock.calls[0][0]).toEqual({
sessionId: 'session-1',
toolCall: { ...toolCall, vendor: 'kept' },
options: [{ optionId: 'allow', name: 'allow', kind: 'allow_once' }]
})
expect(diagnostics).toEqual([
'Delivered ACP permission request without unreadable fields: options.0.name, options.1, toolCall.content'
])
})
it.each<[string, AcpPermissionHandler, string]>([
[
'the handler throws',
() => {
throw new Error('renderer gone')
},
'handler failed: Error: renderer gone'
],
// JSON.parse yields an untyped value, so a deliberately malformed reply needs no assertion.
[
'the handler answers nonsense',
() => JSON.parse('{"outcome":"yes"}'),
'invalid handler response'
]
])('answers cancelled with a diagnostic when %s', async (_label, onPermission, problem) => {
const { agent, runtime, diagnostics } = fixture({ onPermission })
await runtime.start(startOptions)
void runtime.prompt([{ type: 'text', text: 'hi' }]).catch(() => {})
expect(
await agent.request('p', 'session/request_permission', {
sessionId: 'session-1',
toolCall,
options: [allow]
})
).toMatchObject({ result: { outcome: { outcome: 'cancelled' } } })
expect(diagnostics).toEqual([`Answered ACP permission request cancelled: ${problem}`])
})
it('answers cancelled when the cancel lands before the permission handler starts', async () => {
const asked = vi.fn((_request: RequestPermissionRequest) => new Promise<never>(() => {}))
const { agent, runtime } = fixture({ onPermission: asked })
await runtime.start(startOptions)
runtime.subscribe(() => void runtime.cancel())
const answer = new Promise<unknown>((resolve) => {
agent.stdin.on('data', (chunk: string) => {
if (chunk.includes('"id":5')) {
resolve(JSON.parse(chunk))
}
})
})
// One chunk: the permission, then an update whose listener cancels before the handler runs.
agent.stdout.write(
[
{
jsonrpc: '2.0',
id: 5,
method: 'session/request_permission',
params: { sessionId: 'session-1', toolCall, options: [allow] }
},
{
jsonrpc: '2.0',
method: 'session/update',
params: {
sessionId: 'session-1',
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'x' } }
}
}
]
.map((frame) => `${JSON.stringify(frame)}\n`)
.join('')
)
expect(await answer).toMatchObject({ id: 5, result: { outcome: { outcome: 'cancelled' } } })
expect(asked).not.toHaveBeenCalled()
})
it('lets a turn the agent began itself ask for permission and cancels it with session/cancel', async () => {
const asked = deferred<AbortSignal>()
const { agent, runtime } = fixture({
onPermission: (_request, context) => {
asked.resolve(context.signal)
return new Promise(() => {})
}
})
await runtime.start(startOptions)
const answer = agent.request('p', 'session/request_permission', {
sessionId: 'session-1',
toolCall,
options: [allow]
})
const signal = await asked.promise
await runtime.cancel()
expect(signal.aborted).toBe(true)
expect(await answer).toMatchObject({ result: { outcome: { outcome: 'cancelled' } } })
await tick()
expect(agent.frames.filter((frame) => frame.method === 'session/cancel')).toHaveLength(1)
})
})
+144
View File
@@ -0,0 +1,144 @@
import { z } from 'zod'
import type { AcpRequestContext } from './acp-json-rpc-peer'
import {
PermissionOptionSchema,
RequestPermissionRequestSchema,
RequestPermissionResponseSchema,
ToolCallUpdateSchema,
type RequestPermissionRequest,
type RequestPermissionResponse
} from './generated/acp-protocol.generated'
export type AcpPermissionHandler = (
request: RequestPermissionRequest,
context: AcpRequestContext
) => RequestPermissionResponse | Promise<RequestPermissionResponse>
const cancelled: RequestPermissionResponse = { outcome: { outcome: 'cancelled' } }
const routingSchema = z.looseObject({
sessionId: z.string(),
toolCall: z.looseObject({ toolCallId: z.string() }),
options: z.array(z.unknown())
})
const optionRoutingSchema = z.looseObject({ optionId: z.string() })
// Keeps every field the schema does not know or can read; drops (and names) the unreadable ones.
function readableFields(
knownFields: Record<string, z.ZodType>,
value: Record<string, unknown>,
path: string,
dropped: string[]
): Record<string, unknown> {
const kept: Record<string, unknown> = {}
for (const [key, field] of Object.entries(value)) {
if (!Object.hasOwn(knownFields, key) || knownFields[key].safeParse(field).success) {
kept[key] = field
} else {
dropped.push(`${path}${key}`)
}
}
return kept
}
/** Validates only what answering needs (session, tool call id, options with ids); null if unusable. */
export function readAcpPermissionRequest(
params: unknown,
diagnose: (message: string) => void
): RequestPermissionRequest | null {
const strict = RequestPermissionRequestSchema.safeParse(params)
if (strict.success) {
return strict.data
}
const routing = routingSchema.safeParse(params)
if (!routing.success) {
return null
}
const { toolCall, options: offered, ...rest } = routing.data
const dropped: string[] = []
const options = offered.flatMap((option, index) => {
const ids = optionRoutingSchema.safeParse(option)
if (!ids.success) {
dropped.push(`options.${index}`)
return []
}
if (typeof ids.data.name !== 'string') {
dropped.push(`options.${index}.name`)
}
const named = {
...ids.data,
name: typeof ids.data.name === 'string' ? ids.data.name : ids.data.optionId
}
const parsed = PermissionOptionSchema.safeParse(named)
if (!parsed.success) {
dropped.push(`options.${index}`)
return []
}
return [parsed.data]
})
if (options.length === 0) {
return null
}
const request = RequestPermissionRequestSchema.safeParse({
...readableFields(RequestPermissionRequestSchema.shape, rest, '', dropped),
toolCall: readableFields(ToolCallUpdateSchema.shape, toolCall, 'toolCall.', dropped),
options
})
if (!request.success) {
return null
}
if (dropped.length > 0) {
diagnose(`Delivered ACP permission request without unreadable fields: ${dropped.join(', ')}`)
}
return request.data
}
/** Asks the caller; any answer Orca cannot send (a throw, a bad reply, an unoffered option) is `cancelled`. */
export function answerAcpPermission(
request: RequestPermissionRequest,
context: AcpRequestContext,
handler: AcpPermissionHandler | undefined,
diagnose: (message: string) => void
): Promise<RequestPermissionResponse> {
if (!handler || context.signal.aborted) {
return Promise.resolve(cancelled)
}
return new Promise((resolve) => {
const controller = new AbortController()
let settled = false
const finish = (response: RequestPermissionResponse, problem?: string): void => {
if (settled) {
return
}
settled = true
context.signal.removeEventListener('abort', onAbort)
controller.abort()
if (problem) {
diagnose(`Answered ACP permission request cancelled: ${problem}`)
}
resolve(response)
}
const onAbort = (): void => finish(cancelled)
context.signal.addEventListener('abort', onAbort, { once: true })
void Promise.resolve()
.then(() => handler(request, { id: context.id, signal: controller.signal }))
.then(
(response) => {
const parsed = RequestPermissionResponseSchema.safeParse(response)
if (!parsed.success) {
finish(cancelled, 'invalid handler response')
return
}
const outcome = parsed.data.outcome
if (
outcome.outcome === 'selected' &&
!request.options.some((option) => option.optionId === outcome.optionId)
) {
finish(cancelled, 'handler selected an unavailable option')
return
}
finish(parsed.data)
},
(error) => finish(cancelled, `handler failed: ${String(error)}`)
)
})
}
+94
View File
@@ -0,0 +1,94 @@
import { AcpRequestTimeoutError } from './acp-errors'
import type { PromptResponse } from './generated/acp-protocol.generated'
export type ActivePrompt = {
response: Promise<PromptResponse>
cancelling: boolean
cancelPromise?: Promise<void>
steerCancel?: Promise<void>
}
/** How a cancel reaches the agent: the `session/cancel` write and the connection it rides on. */
export type AcpCancelChannel = {
send: () => Promise<void>
/** Cancels the agent's open requests (permissions, vendor hooks) so each gets its answer. */
cancelIncomingRequests: () => void
close: (error: Error) => void
closed: () => boolean
timeoutMs: number
}
/** A Stop's cancel. Sends `session/cancel` whenever the session runs: the agent may be in a turn it
* began itself. With Orca's prompt running, also waits (bounded) for that prompt to settle, and
* past the bound closes the connection, which ends the agent. Repeated asks share one wait. */
export function cancelAcpPromptForStop(
active: ActivePrompt | undefined,
channel: AcpCancelChannel
): Promise<void> {
channel.cancelIncomingRequests()
if (!active) {
return channel.send()
}
active.cancelling = true
active.cancelPromise ??= confirmAcpPromptCancel(active, channel)
return active.cancelPromise
}
/** Sends `session/cancel` and waits, bounded, for Orca's prompt to settle; past the bound it closes. */
async function confirmAcpPromptCancel(
active: ActivePrompt,
connection: AcpCancelChannel
): Promise<void> {
let timer: ReturnType<typeof setTimeout> | undefined
const unconfirmed = new Promise<never>((_resolve, reject) => {
timer = setTimeout(() => {
const error = new AcpRequestTimeoutError('session/cancel')
connection.close(error)
reject(error)
}, connection.timeoutMs)
})
try {
await Promise.race([connection.send(), unconfirmed]).catch((error) => {
// Only a cancel that reached the agent is shared; a failed write is retried next call.
active.cancelPromise = undefined
active.cancelling = false
throw error
})
await Promise.race([
active.response.catch((error) => {
if (connection.closed()) {
throw error
}
}),
unconfirmed
])
} finally {
clearTimeout(timer)
}
}
/** A steer's cancel: asks the agent once to end Orca's running prompt so the steer can follow it,
* and resolves when the notification is written. Never bounded and never closes: the prompt's own
* reply ends it, however long that takes, and a later Stop still bounds. A prompt that fails
* instead leaves a session the steer must not be sent into until the caller rebuilds it. With no
* prompt of Orca's there is nothing to end; a Stop already cancelling it owns that cancel. */
export function requestAcpSteerCancel(
active: ActivePrompt | undefined,
channel: Pick<AcpCancelChannel, 'send' | 'cancelIncomingRequests'>
): Promise<void> {
if (!active) {
return Promise.resolve()
}
if (active.steerCancel || active.cancelling) {
return active.steerCancel ?? Promise.resolve()
}
active.cancelling = true
channel.cancelIncomingRequests()
active.steerCancel = channel.send().catch((error: unknown) => {
// Only a cancel that reached the agent counts; a failed write may be asked again.
active.cancelling = false
active.steerCancel = undefined
throw error
})
return active.steerCancel
}
@@ -0,0 +1,85 @@
import { PassThrough } from 'node:stream'
import { z } from 'zod'
const frameSchema = z.object({
jsonrpc: z.literal('2.0'),
id: z.union([z.string(), z.number(), z.null()]).optional(),
method: z.string().optional(),
params: z.unknown().optional(),
result: z.unknown().optional(),
error: z
.object({ code: z.number(), message: z.string(), data: z.unknown().optional() })
.optional()
})
export type FakeFrame = z.infer<typeof frameSchema>
type Handler = (frame: FakeFrame) => void
export class AcpScriptedAgent {
readonly stdout = new PassThrough()
readonly stdin = new PassThrough()
readonly frames: FakeFrame[] = []
private readonly methods = new Map<string, Handler>()
private readonly requests = new Map<string | number | null, (response: FakeFrame) => void>()
private suffix = ''
constructor() {
this.stdin.setEncoding('utf8').on('data', (chunk: string) => {
this.suffix += chunk
let newline: number
while ((newline = this.suffix.indexOf('\n')) !== -1) {
const frame = frameSchema.parse(JSON.parse(this.suffix.slice(0, newline)))
this.suffix = this.suffix.slice(newline + 1)
this.frames.push(frame)
if (frame.method !== undefined) {
this.methods.get(frame.method)?.(frame)
} else if (frame.id !== undefined) {
this.requests.get(frame.id)?.(frame)
this.requests.delete(frame.id)
}
}
})
}
on(method: string, handler: Handler): void {
this.methods.set(method, handler)
}
reply(frame: FakeFrame, result: unknown): void {
this.send({ jsonrpc: '2.0', id: frame.id, result })
}
fail(frame: FakeFrame, code: number, message: string, data?: unknown): void {
this.send({ jsonrpc: '2.0', id: frame.id, error: { code, message, data } })
}
notify(method: string, params: unknown): void {
this.send({ jsonrpc: '2.0', method, params })
}
request(id: string | number | null, method: string, params: unknown): Promise<FakeFrame> {
return new Promise((resolve) => {
this.requests.set(id, resolve)
this.send({ jsonrpc: '2.0', id, method, params })
})
}
send(frame: FakeFrame): void {
this.stdout.write(`${JSON.stringify(frame)}\n`)
}
close(): void {
this.stdout.end()
this.stdin.end()
}
}
export function deferred<T>(): { promise: Promise<T>; resolve: (value: T) => void } {
let fulfill: ((value: T) => void) | undefined
const promise = new Promise<T>((resolve) => {
fulfill = resolve
})
return {
promise,
resolve: (value) => {
fulfill?.(value)
}
}
}
export function tick(): Promise<void> {
return new Promise((resolve) => setImmediate(resolve))
}
@@ -0,0 +1,148 @@
import { PassThrough } from 'node:stream'
import { afterEach, describe, expect, it } from 'vitest'
import { AcpAgentError, AcpInvalidResponseError } from './acp-errors'
import { AcpSessionRuntime, type AcpSessionRuntimeOptions } from './acp-session-runtime'
import { AcpScriptedAgent, type FakeFrame, tick } from './acp-scripted-agent.test-support'
const opened: { close: () => void }[] = []
const startOptions = { cwd: '/runtime/project', mcpServers: [] }
const prompt = [{ type: 'text', text: 'hello' }] as const
function fixture(options: AcpSessionRuntimeOptions = {}) {
const agent = new AcpScriptedAgent()
const runtime = new AcpSessionRuntime(agent.stdout, agent.stdin, options)
opened.push(agent, runtime)
agent.on('initialize', (frame) => agent.reply(frame, { protocolVersion: 1 }))
agent.on('session/new', (frame) => agent.reply(frame, { sessionId: 'session-1' }))
return { agent, runtime }
}
afterEach(() =>
opened
.splice(0)
.toReversed()
.forEach((resource) => resource.close())
)
describe('ACP turns the agent begins and the runtime contract around them', () => {
it('delivers extension notifications in arrival order with session updates', async () => {
const order: string[] = []
const { agent, runtime } = fixture({
onExtensionNotification: (method, params) => order.push(`${method} ${JSON.stringify(params)}`)
})
runtime.subscribe((event) => order.push(`update ${event.kind}`))
await runtime.start(startOptions)
const completed = { sessionId: 'session-1', update: { type: 'turn_completed', prompt_id: 'p' } }
agent.stdout.write(
[
{ jsonrpc: '2.0', method: '_x.ai/session_notification', params: { started: true } },
{
jsonrpc: '2.0',
method: 'session/update',
params: {
sessionId: 'session-1',
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'x' } }
}
},
{ jsonrpc: '2.0', method: '_x.ai/session_notification', params: completed }
]
.map((frame) => `${JSON.stringify(frame)}\n`)
.join('')
)
await tick()
expect(order).toEqual([
'_x.ai/session_notification {"started":true}',
'update known',
`_x.ai/session_notification ${JSON.stringify(completed)}`
])
})
it('sends session/cancel with no prompt of Orca running, and _meta on every session call', async () => {
const { agent, runtime } = fixture()
agent.on('session/prompt', (frame) => agent.reply(frame, { stopReason: 'end_turn' }))
agent.on('session/set_mode', (frame) => agent.reply(frame, {}))
agent.on('session/set_model', (frame) => agent.reply(frame, {}))
agent.on('session/set_config_option', (frame) => agent.reply(frame, { configOptions: [] }))
await runtime.start(startOptions)
const meta = { traceId: 't' }
await runtime.prompt([...prompt], meta)
await runtime.setMode('plan', meta)
await runtime.setModel('model-1', meta)
await runtime.setConfigOption('effort', 'high', meta)
await runtime.cancel({ meta })
await tick()
const sent = agent.frames.filter((frame) => frame.method?.startsWith('session/'))
expect(sent.map((frame) => frame.method)).toEqual([
'session/new',
'session/prompt',
'session/set_mode',
'session/set_model',
'session/set_config_option',
'session/cancel'
])
for (const frame of sent.slice(1)) {
expect(frame.params).toMatchObject({ sessionId: 'session-1', _meta: meta })
}
})
it("separates the agent's own errors from answers Orca could not read", async () => {
const { agent, runtime } = fixture()
await runtime.start(startOptions)
agent.on('session/set_mode', (frame) => agent.fail(frame, -32603, 'Agent broke', { why: 1 }))
const refused = await runtime.setMode('plan').catch((error: unknown) => error)
expect(refused).toBeInstanceOf(AcpAgentError)
expect(refused).toMatchObject({ code: -32603, data: { why: 1 } })
agent.on('session/set_mode', (frame) => agent.reply(frame, 'not-an-object'))
const unreadable = await runtime.setMode('plan').catch((error: unknown) => error)
expect(unreadable).toBeInstanceOf(AcpInvalidResponseError)
expect(unreadable).not.toBeInstanceOf(AcpAgentError)
expect(unreadable).toMatchObject({ code: -32603, data: 'not-an-object' })
expect(unreadable instanceof AcpInvalidResponseError && unreadable.issues).toBeTruthy()
})
it('completes a turn whose stop reason is newer than this schema', async () => {
const { agent, runtime } = fixture()
agent.on('session/prompt', (frame) => agent.reply(frame, { stopReason: 'context_exhausted' }))
await runtime.start(startOptions)
expect(await runtime.prompt([...prompt])).toEqual({ stopReason: 'context_exhausted' })
})
it('retries a cancel whose write failed instead of returning the stale failure', async () => {
const agent = new AcpScriptedAgent()
const stdin = new PassThrough({ highWaterMark: 1 })
const runtime = new AcpSessionRuntime(agent.stdout, stdin, {
peer: { maxQueuedWriteBytes: 1000 }
})
opened.push(agent, runtime)
const frames: FakeFrame[] = []
let promptFrame: FakeFrame | undefined
let buffer = ''
stdin.setEncoding('utf8').on('data', (chunk: string) => {
buffer += chunk
let newline: number
while ((newline = buffer.indexOf('\n')) !== -1) {
const frame: FakeFrame = JSON.parse(buffer.slice(0, newline))
buffer = buffer.slice(newline + 1)
frames.push(frame)
if (frame.method === 'initialize') {
agent.reply(frame, { protocolVersion: 1 })
} else if (frame.method === 'session/new') {
agent.reply(frame, { sessionId: 'session-1' })
} else if (frame.method === 'session/prompt') {
promptFrame = frame
stdin.pause()
} else if (frame.method === 'session/cancel' && promptFrame) {
agent.reply(promptFrame, { stopReason: 'cancelled' })
}
}
})
await runtime.start(startOptions)
const turn = runtime.prompt([...prompt])
await tick()
void runtime.setMode('y'.repeat(850)).catch(() => {})
await expect(runtime.cancel()).rejects.toThrow(/capacity/)
stdin.resume()
await tick()
await runtime.cancel()
expect(await turn).toEqual({ stopReason: 'cancelled' })
expect(frames.filter((frame) => frame.method === 'session/cancel')).toHaveLength(1)
})
})
+26
View File
@@ -0,0 +1,26 @@
import { z } from 'zod'
import {
SessionNotificationSchema,
type SessionNotification
} from './generated/acp-protocol.generated'
const updateEnvelopeSchema = z.looseObject({
sessionId: z.string(),
update: z.looseObject({ sessionUpdate: z.string() })
})
export type AcpSessionEvent =
| { kind: 'known'; notification: SessionNotification }
| { kind: 'unrecognized'; sessionId: string; raw: z.infer<typeof updateEnvelopeSchema> }
/** A `session/update`, typed when this build knows its kind; null when it is not one at all. */
export function readAcpSessionEvent(params: unknown): AcpSessionEvent | null {
const envelope = updateEnvelopeSchema.safeParse(params)
if (!envelope.success) {
return null
}
const parsed = SessionNotificationSchema.safeParse(params)
return parsed.success
? { kind: 'known', notification: parsed.data }
: { kind: 'unrecognized', sessionId: envelope.data.sessionId, raw: envelope.data }
}
+373
View File
@@ -0,0 +1,373 @@
import { readFile } from 'node:fs/promises'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { AcpJsonRpcPeer } from './acp-json-rpc-peer'
import { AcpSessionRuntime, type AcpSessionRuntimeOptions } from './acp-session-runtime'
import { AcpScriptedAgent, deferred, tick } from './acp-scripted-agent.test-support'
import { AcpConnectionClosedError, AcpRequestTimeoutError } from './acp-errors'
import type { RequestPermissionResponse } from './generated/acp-protocol.generated'
const opened: { close: () => void }[] = []
const startOptions = { cwd: '/runtime/project', mcpServers: [] }
const prompt = [{ type: 'text', text: 'hello' }] as const
const permission = {
sessionId: 'session-1',
toolCall: { toolCallId: 'tool-1', title: 'Edit file' },
options: [{ optionId: 'allow', name: 'Allow once', kind: 'allow_once' }]
}
function fixture(options: AcpSessionRuntimeOptions = {}) {
const agent = new AcpScriptedAgent()
const runtime = new AcpSessionRuntime(agent.stdout, agent.stdin, options)
opened.push(agent, runtime)
agent.on('initialize', (frame) => agent.reply(frame, { protocolVersion: 1 }))
agent.on('session/new', (frame) => agent.reply(frame, { sessionId: 'session-1' }))
return { agent, runtime }
}
afterEach(() => {
opened
.splice(0)
.toReversed()
.forEach((resource) => resource.close())
vi.useRealTimers()
})
describe('ACP caller-owned waits', () => {
it('keeps permission requests open beyond two minutes and delivers the late user decision', async () => {
vi.useFakeTimers()
const entered = deferred<AbortSignal>()
const decision = deferred<RequestPermissionResponse>()
const { agent, runtime } = fixture({
onPermission: (_request, context) => {
entered.resolve(context.signal)
return decision.promise
}
})
agent.on('session/prompt', () => {})
await runtime.start(startOptions)
const pending = runtime.prompt([...prompt])
const rejected = expect(pending).rejects.toBeInstanceOf(AcpConnectionClosedError)
let answered = false
const response = agent.request('permission', 'session/request_permission', permission)
void response.then(() => {
answered = true
})
const signal = await entered.promise
await vi.advanceTimersByTimeAsync(120_001)
expect(signal.aborted).toBe(false)
expect(answered).toBe(false)
decision.resolve({ outcome: { outcome: 'selected', optionId: 'allow' } })
expect(await response).toMatchObject({
result: { outcome: { outcome: 'selected', optionId: 'allow' } }
})
runtime.close()
await rejected
})
it('lets a streaming turn run beyond thirty minutes and keeps the session usable', async () => {
vi.useFakeTimers()
const onClose = vi.fn()
const event = vi.fn()
const { agent, runtime } = fixture({ onClose })
runtime.subscribe(event)
agent.on('session/prompt', () => {})
agent.on('session/set_mode', (frame) => agent.reply(frame, {}))
await runtime.start(startOptions)
let finished = false
const pending = runtime.prompt([...prompt])
void pending.then(() => {
finished = true
})
await vi.advanceTimersByTimeAsync(30 * 60_000 + 1)
agent.notify('session/update', {
sessionId: 'session-1',
update: {
sessionUpdate: 'agent_message_chunk',
content: { type: 'text', text: 'Still working' }
}
})
expect(event).toHaveBeenCalledTimes(1)
expect(finished).toBe(false)
expect(onClose).not.toHaveBeenCalled()
expect(agent.frames.some((frame) => frame.method === 'session/cancel')).toBe(false)
await expect(runtime.setMode('plan')).resolves.toEqual({})
const frame = agent.frames.find((frame) => frame.method === 'session/prompt')
expect(frame).toBeDefined()
if (frame) {
agent.reply(frame, { stopReason: 'end_turn' })
}
await expect(pending).resolves.toEqual({ stopReason: 'end_turn' })
})
it.each([
{ id: 'key', name: 'API key', type: 'env_var', vars: [{ name: 'X_API_KEY' }] },
{ id: 'browser', name: 'Browser login', type: 'agent' },
{ id: 'default', name: 'Default login' },
{ id: 'terminal', name: 'Terminal login', type: 'terminal' }
])('surfaces $id authentication without choosing it for the caller', async (method) => {
const { agent, runtime } = fixture()
agent.on('initialize', (frame) =>
agent.reply(frame, { protocolVersion: 1, authMethods: [method] })
)
agent.on('session/new', (frame) => agent.fail(frame, -32000, 'Authentication required'))
await expect(runtime.start(startOptions)).rejects.toMatchObject({
name: 'AcpAuthRequiredError',
authMethods: [method]
})
expect(agent.frames.map((frame) => frame.method)).toEqual(['initialize', 'session/new'])
})
it.each(['prompt', 'cancel', 'setMode', 'setModel', 'setConfigOption'] as const)(
'rejects %s before start through its promise',
async (method) => {
const { runtime } = fixture()
let pending: Promise<unknown> | undefined
expect(() => {
switch (method) {
case 'prompt':
pending = runtime.prompt([...prompt])
break
case 'cancel':
pending = runtime.cancel()
break
case 'setMode':
pending = runtime.setMode('plan')
break
case 'setModel':
pending = runtime.setModel('model')
break
case 'setConfigOption':
pending = runtime.setConfigOption('thinking', 'high')
break
}
}).not.toThrow()
await expect(pending).rejects.toThrow('ACP session has not started')
}
)
it('answers a handled void vendor request with null', async () => {
const handler = vi.fn()
const { agent } = fixture({ onRequest: handler })
expect(await agent.request(7, '_vendor/ack', {})).toMatchObject({ id: 7, result: null })
expect(handler).toHaveBeenCalledTimes(1)
})
it('ignores duplicate incoming ids without answering the original request', async () => {
const answer = deferred<unknown>()
const handler = vi.fn(() => answer.promise)
const diagnostics: string[] = []
const { agent } = fixture({
onRequest: handler,
onDiagnostic: (message) => diagnostics.push(message)
})
const original = agent.request(5, '_vendor/question', {})
agent.send({ jsonrpc: '2.0', id: 5, method: '_vendor/question', params: {} })
await tick()
expect(agent.frames).toEqual([])
expect(handler).toHaveBeenCalledTimes(1)
expect(diagnostics).toContain('Ignored duplicate ACP incoming request id')
answer.resolve({ answer: true })
expect(await original).toMatchObject({ id: 5, result: { answer: true } })
expect(agent.frames).toHaveLength(1)
})
it.each([{ code: 'E1', message: 'boom' }, { code: -1, message: null }, null])(
'rejects a malformed error reply immediately with its raw error: %j',
async (error) => {
const { agent, runtime } = fixture()
agent.on('initialize', (frame) => {
agent.stdout.write(`${JSON.stringify({ jsonrpc: '2.0', id: frame.id, error })}\n`)
})
await expect(runtime.initialize()).rejects.toMatchObject({ code: -32603, data: error })
}
)
it('maps authentication-required and retries only the explicitly chosen advertised method once', async () => {
const { agent, runtime } = fixture()
const authMethods = [{ id: 'login', name: 'Login' }]
agent.on('initialize', (frame) => agent.reply(frame, { protocolVersion: 1, authMethods }))
agent.on('session/new', (frame) => agent.fail(frame, -32000, 'Authentication required'))
agent.on('authenticate', (frame) => agent.reply(frame, {}))
await expect(runtime.start({ ...startOptions, authMethodId: 'login' })).rejects.toMatchObject({
name: 'AcpAuthRequiredError',
authMethods
})
expect(agent.frames.map((frame) => frame.method)).toEqual([
'initialize',
'session/new',
'authenticate',
'session/new'
])
})
it('rejects overflowing peer timeouts and supports an explicit unlimited wait', async () => {
vi.useFakeTimers()
const agent = new AcpScriptedAgent()
opened.push(agent)
expect(
() => new AcpJsonRpcPeer(agent.stdout, agent.stdin, {}, { requestTimeoutMs: 3_000_000_000 })
).toThrow('timer durations')
const peer = new AcpJsonRpcPeer(agent.stdout, agent.stdin, {}, { requestTimeoutMs: 20 })
opened.push(peer)
await expect(peer.request('overflow', {}, { timeoutMs: 3_000_000_000 })).rejects.toThrow(
'timer durations'
)
const pending = peer.request('wait', {}, { timeoutMs: null })
const rejected = expect(pending).rejects.toBeInstanceOf(AcpConnectionClosedError)
await vi.advanceTimersByTimeAsync(3_000_000_000)
expect(peer.closed).toBe(false)
expect(agent.frames.map((frame) => frame.method)).toEqual(['wait'])
peer.close()
await rejected
})
it('clamps cancellation confirmation waits to the safe timer range', async () => {
vi.useFakeTimers()
const { agent, runtime } = fixture({ cancelTimeoutMs: 3_000_000_000 })
agent.on('session/prompt', () => {})
await runtime.start(startOptions)
const rejected = expect(runtime.prompt([...prompt])).rejects.toBeInstanceOf(
AcpRequestTimeoutError
)
const cancelled = expect(runtime.cancel()).rejects.toBeInstanceOf(AcpRequestTimeoutError)
await vi.advanceTimersByTimeAsync(1)
expect(agent.frames.at(-1)?.method).toBe('session/cancel')
await vi.advanceTimersByTimeAsync(2_147_483_646)
await Promise.all([rejected, cancelled])
})
it('lets initialize and explicit authentication wait for slow first-run startup and login', async () => {
vi.useFakeTimers()
const { agent, runtime } = fixture({ peer: { requestTimeoutMs: 20 } })
agent.on('initialize', () => {})
const initialized = runtime.initialize()
await vi.advanceTimersByTimeAsync(120_001)
agent.reply(agent.frames[0], { protocolVersion: 1 })
await initialized
const authenticated = runtime.authenticate('caller-chosen')
await vi.advanceTimersByTimeAsync(120_001)
agent.reply(agent.frames[1], {})
await expect(authenticated).resolves.toEqual({})
})
it('lets cancel abort vendor hooks so each sends its own answer, or request-cancelled if it throws', async () => {
const { agent, runtime } = fixture({
onRequest: (method, _params, context) =>
new Promise((resolve, reject) => {
// '_vendor/silent' ignores the abort: the runtime never answers for it.
if (method !== '_vendor/silent') {
context.signal.addEventListener('abort', () =>
method === '_vendor/plan'
? resolve({ outcome: 'abandoned' })
: reject(new Error('stop'))
)
}
})
})
agent.on('session/prompt', (frame) =>
agent.on('session/cancel', () => agent.reply(frame, { stopReason: 'cancelled' }))
)
await runtime.start(startOptions)
const pending = runtime.prompt([...prompt])
const question = agent.request('question', '_vendor/question', {})
const plan = agent.request('plan', '_vendor/plan', {})
void agent.request('silent', '_vendor/silent', {})
await tick()
await runtime.cancel()
expect(await question).toMatchObject({ error: { code: -32800 } })
expect(await plan).toMatchObject({ result: { outcome: 'abandoned' } })
await pending
await tick()
expect(agent.frames.filter((frame) => frame.id === 'plan')).toHaveLength(1)
expect(agent.frames.some((frame) => frame.id === 'silent')).toBe(false)
})
it('keeps a handler answer that finishes saving after the cancel', async () => {
// A real I/O hop, the shape of a journal write the handler commits before replying.
const save = (): Promise<void> => readFile(import.meta.filename).then(() => undefined)
const userAnswer = deferred<void>()
const { agent, runtime } = fixture({
onRequest: (method, _params, context) =>
new Promise((resolve) => {
if (method === '_vendor/plan') {
// The user already approved; the save started before the stop and replies after it.
void userAnswer.promise.then(save).then(() => resolve({ outcome: 'approved' }))
} else {
context.signal.addEventListener(
'abort',
() => void save().then(() => resolve({ outcome: 'abandoned' }))
)
}
})
})
agent.on('session/prompt', (frame) =>
agent.on('session/cancel', () => agent.reply(frame, { stopReason: 'cancelled' }))
)
await runtime.start(startOptions)
const pending = runtime.prompt([...prompt])
const plan = agent.request('plan', '_vendor/plan', {})
const question = agent.request('question', '_vendor/question', {})
await tick()
userAnswer.resolve()
await runtime.cancel()
expect(await plan).toMatchObject({ result: { outcome: 'approved' } })
expect(await question).toMatchObject({ result: { outcome: 'abandoned' } })
await pending
await tick()
expect(agent.frames.filter((frame) => frame.id === 'plan')).toHaveLength(1)
expect(agent.frames.filter((frame) => frame.id === 'question')).toHaveLength(1)
})
it('settles requests when the process owner closes on exit even if stdout stays open', async () => {
const { agent, runtime } = fixture()
agent.on('session/prompt', () => {})
await runtime.start(startOptions)
const rejected = expect(runtime.prompt([...prompt])).rejects.toThrow('Agent exited')
expect(agent.stdout.readableEnded).toBe(false)
runtime.close(new Error('Agent exited'))
await rejected
expect(agent.stdout.readableEnded).toBe(false)
})
it.each(['answer', 'cancel'] as const)(
'keeps a permission pending after prompt completion until caller %s',
async (action) => {
const decision = deferred<RequestPermissionResponse>()
const entered = deferred<AbortSignal>()
const { agent, runtime } = fixture({
onPermission: (_request, context) => {
entered.resolve(context.signal)
return decision.promise
}
})
agent.on('session/prompt', () => {})
await runtime.start(startOptions)
const pending = runtime.prompt([...prompt])
const response = agent.request('permission', 'session/request_permission', permission)
const signal = await entered.promise
const frame = agent.frames.find((candidate) => candidate.method === 'session/prompt')
if (frame) {
agent.reply(frame, { stopReason: 'end_turn' })
}
await pending
expect(signal.aborted).toBe(false)
if (action === 'answer') {
decision.resolve({ outcome: { outcome: 'selected', optionId: 'allow' } })
expect(await response).toMatchObject({
result: { outcome: { outcome: 'selected', optionId: 'allow' } }
})
} else {
await runtime.cancel()
expect(await response).toMatchObject({ result: { outcome: { outcome: 'cancelled' } } })
}
}
)
it('confirms cancellation when the agent settles the prompt with an error', async () => {
const { agent, runtime } = fixture()
agent.on('session/prompt', (frame) => {
agent.on('session/cancel', () => agent.fail(frame, -32800, 'Request cancelled'))
})
await runtime.start(startOptions)
const rejected = expect(runtime.prompt([...prompt])).rejects.toMatchObject({ code: -32800 })
await expect(runtime.cancel()).resolves.toBeUndefined()
await rejected
})
})
@@ -0,0 +1,106 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { AcpSessionRuntime, type AcpSessionEvent } from './acp-session-runtime'
import { AcpScriptedAgent } from './acp-scripted-agent.test-support'
const opened: { close: () => void }[] = []
function fixture() {
const agent = new AcpScriptedAgent()
const diagnostic = vi.fn()
const runtime = new AcpSessionRuntime(agent.stdout, agent.stdin, { onDiagnostic: diagnostic })
opened.push(agent, runtime)
agent.on('initialize', (frame) =>
agent.reply(frame, { protocolVersion: 1, agentCapabilities: { vendor: true } })
)
agent.on('session/new', (frame) => agent.reply(frame, { sessionId: 'session-1', vendor: true }))
return { agent, runtime, diagnostic }
}
afterEach(() =>
opened
.splice(0)
.toReversed()
.forEach((resource) => resource.close())
)
describe('ACP session update compatibility', () => {
it.each([
['known', { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'hi' } }],
// Unfamiliar enum values stay typed: the generated enums are open.
['known', { sessionUpdate: 'tool_call', toolCallId: 't', title: 'Search', kind: 'web_search' }],
['known', { sessionUpdate: 'tool_call_update', toolCallId: 't', status: 'cancelled' }],
[
'known',
{
sessionUpdate: 'plan',
entries: [{ content: 'Ship', priority: 'urgent', status: 'blocked' }]
}
],
['unrecognized', { sessionUpdate: 'vendor_usage', tokens: 1 }],
[
'unrecognized',
{ sessionUpdate: 'agent_message_chunk', content: { type: 'reasoning_summary', text: 'x' } }
]
] as const)(
'delivers %s updates without losing their original fields (%j)',
async (kind, update) => {
const { agent, runtime, diagnostic } = fixture()
const events: AcpSessionEvent[] = []
runtime.subscribe((event) => events.push(event))
await runtime.start({ cwd: '/runtime/project', mcpServers: [] })
const notification = { sessionId: 'session-1', update, vendor: 'retained' }
agent.notify('session/update', notification)
agent.notify('session/update', notification)
expect(events).toHaveLength(2)
expect(events[0]).toEqual(
kind === 'known'
? { kind, notification }
: { kind, sessionId: 'session-1', raw: notification }
)
expect(diagnostic).toHaveBeenCalledTimes(kind === 'known' ? 0 : 1)
}
)
it('rejects only envelopes missing the session id or update discriminator', async () => {
const { agent, runtime } = fixture()
const received = vi.fn()
runtime.subscribe(received)
await runtime.start({ cwd: '/runtime/project', mcpServers: [] })
agent.notify('session/update', { update: { sessionUpdate: 'vendor' } })
agent.notify('session/update', { sessionId: 'session-1', update: {} })
expect(received).not.toHaveBeenCalled()
})
it('preserves extra session fields and exposes legacy model state as typed data', async () => {
const { agent, runtime } = fixture()
const models = {
currentModelId: 'model-1',
availableModels: [{ modelId: 'model-1', name: 'Model one' }]
}
agent.on('session/new', (frame) =>
agent.reply(frame, { sessionId: 'session-1', models, vendor: true })
)
const started = await runtime.start({ cwd: '/runtime/project', mcpServers: [] })
expect(started.response.models?.availableModels[0].modelId).toBe('model-1')
expect(started.response.vendor).toBe(true)
})
it('delivers updates even if diagnostic and earlier event callbacks throw', async () => {
const agent = new AcpScriptedAgent()
const runtime = new AcpSessionRuntime(agent.stdout, agent.stdin, {
onDiagnostic: () => {
throw new Error('Diagnostic failed')
}
})
opened.push(agent, runtime)
runtime.subscribe(() => {
throw new Error('Listener failed')
})
const received = vi.fn()
runtime.subscribe(received)
const notification = { sessionId: 'session-1', update: { sessionUpdate: 'vendor' } }
agent.notify('session/update', notification)
expect(received).toHaveBeenCalledWith({
kind: 'unrecognized',
sessionId: 'session-1',
raw: notification
})
})
})
+496
View File
@@ -0,0 +1,496 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import {
AcpAuthRequiredError,
AcpConnectionClosedError,
AcpRequestTimeoutError,
AcpRpcError
} from './acp-errors'
import {
AcpSessionRuntime,
type AcpSessionRuntimeOptions,
type AcpSessionEvent
} from './acp-session-runtime'
import { AcpScriptedAgent, deferred, tick } from './acp-scripted-agent.test-support'
import type {
AgentCapabilities,
RequestPermissionResponse
} from './generated/acp-protocol.generated'
import { SetSessionConfigOptionRequestSchema } from './generated/acp-protocol.generated'
const opened: { runtime: AcpSessionRuntime; agent: AcpScriptedAgent }[] = []
const startOptions = { cwd: '/runtime/project', mcpServers: [] }
const textPrompt = [{ type: 'text', text: 'hello' }] as const
const permission = {
sessionId: 'session-1',
toolCall: { toolCallId: 'tool-1', title: 'Edit file' },
options: [{ optionId: 'allow', name: 'Allow once', kind: 'allow_once' }]
}
function fixture(capabilities: AgentCapabilities = {}, options: AcpSessionRuntimeOptions = {}) {
const agent = new AcpScriptedAgent()
agent.on('initialize', (frame) =>
agent.reply(frame, { protocolVersion: 1, agentCapabilities: capabilities })
)
agent.on('session/new', (frame) => agent.reply(frame, { sessionId: 'session-1' }))
agent.on('session/load', (frame) => agent.reply(frame, {}))
agent.on('session/resume', (frame) => agent.reply(frame, {}))
agent.on('session/prompt', (frame) => agent.reply(frame, { stopReason: 'end_turn' }))
const runtime = new AcpSessionRuntime(agent.stdout, agent.stdin, options)
opened.push({ runtime, agent })
return { runtime, agent }
}
afterEach(() => {
for (const { runtime, agent } of opened.splice(0)) {
runtime.close()
agent.close()
}
vi.useRealTimers()
})
describe('ACP session runtime', () => {
it('initializes once, starts a session, streams typed updates, and completes the turn', async () => {
const { runtime, agent } = fixture()
const events: AcpSessionEvent[] = []
const unsubscribe = runtime.subscribe((event) => events.push(event))
const updates = [
{ sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'Hello' } },
{ sessionUpdate: 'agent_thought_chunk', content: { type: 'text', text: 'Thinking' } },
{ sessionUpdate: 'user_message_chunk', content: { type: 'text', text: 'hello' } },
{ sessionUpdate: 'tool_call', toolCallId: 'tool-1', title: 'Read file' },
{ sessionUpdate: 'tool_call_update', toolCallId: 'tool-1', status: 'completed' },
{
sessionUpdate: 'plan',
entries: [{ content: 'Read', priority: 'medium', status: 'completed' }]
},
{ sessionUpdate: 'usage_update', used: 12, size: 100 },
{ sessionUpdate: 'available_commands_update', availableCommands: [] },
{ sessionUpdate: 'current_mode_update', currentModeId: 'plan' },
{ sessionUpdate: 'config_option_update', configOptions: [] },
{ sessionUpdate: 'session_info_update', title: 'Test session' }
]
agent.on('session/prompt', (frame) => {
for (const update of updates) {
agent.notify('session/update', { sessionId: 'session-1', update })
}
agent.reply(frame, { stopReason: 'end_turn' })
})
await Promise.all([runtime.initialize(), runtime.initialize()])
expect(await runtime.start(startOptions)).toMatchObject({ kind: 'new', sessionId: 'session-1' })
expect(await runtime.prompt([...textPrompt])).toEqual({ stopReason: 'end_turn' })
expect(
events.map((event) => (event.kind === 'known' ? event.notification.update : event.raw.update))
).toEqual(updates)
expect(agent.frames.filter((frame) => frame.method === 'initialize')).toHaveLength(1)
expect(agent.frames[0].params).toEqual({
protocolVersion: 1,
clientCapabilities: { fs: { readTextFile: false, writeTextFile: false }, terminal: false }
})
expect(agent.frames.find((frame) => frame.method === 'session/new')?.params).toEqual(
startOptions
)
unsubscribe()
agent.notify('session/update', { sessionId: 'session-1', update: updates[0] })
expect(events).toHaveLength(updates.length)
})
it.each([
[{ loadSession: true }, undefined, 'load'],
[{ sessionCapabilities: { resume: {} } }, undefined, 'resume'],
[{ loadSession: true, sessionCapabilities: { resume: {} } }, undefined, 'load'],
[{ loadSession: true, sessionCapabilities: { resume: {} } }, 'resume', 'resume'],
[{ loadSession: true, sessionCapabilities: { resume: null } }, 'resume', 'load']
] as const)(
'selects supported activation (%j, %s)',
async (capabilities, preference, expected) => {
const { runtime, agent } = fixture(capabilities)
const result = await runtime.start({
...startOptions,
sessionId: 'old',
resumePreference: preference
})
expect(result).toMatchObject({ kind: expected, sessionId: 'old' })
expect(agent.frames.at(-1)).toMatchObject({
method: `session/${expected}`,
params: { ...startOptions, sessionId: 'old' }
})
}
)
it('refuses unsupported restoration without silently creating a different session', async () => {
const { runtime, agent } = fixture({ sessionCapabilities: { resume: null } })
await expect(runtime.start({ ...startOptions, sessionId: 'old' })).rejects.toMatchObject({
code: -32601
})
expect(agent.frames.map((frame) => frame.method)).toEqual(['initialize'])
})
it('round-trips permission and vendor requests while prompt is pending', async () => {
const permitted = deferred<unknown>()
const { runtime, agent } = fixture(
{},
{
onPermission: (request) => {
expect(request.toolCall.toolCallId).toBe('tool-1')
return { outcome: { outcome: 'selected', optionId: 'allow' } }
},
onRequest: (method, params) =>
method === '_vendor/question'
? { answer: params }
: (() => {
throw new AcpRpcError(-32601, 'Unknown method')
})()
}
)
agent.on('session/prompt', (frame) => {
void agent.request('agent-id', 'session/request_permission', permission).then((response) => {
permitted.resolve(response.result)
agent.reply(frame, { stopReason: 'end_turn' })
})
})
await runtime.start(startOptions)
const prompt = runtime.prompt([...textPrompt])
expect(await permitted.promise).toEqual({ outcome: { outcome: 'selected', optionId: 'allow' } })
await prompt
expect(await agent.request(1, '_vendor/question', 'yes')).toMatchObject({
id: 1,
result: { answer: 'yes' }
})
expect(await agent.request('unknown', '_vendor/unknown', {})).toMatchObject({
error: { code: -32601 }
})
expect(await agent.request('fs', 'fs/read_text_file', {})).toMatchObject({
error: { code: -32601 }
})
})
it('cancels open and late permissions and waits for the agent cancellation response', async () => {
const requested = deferred<AbortSignal>()
const decision = deferred<RequestPermissionResponse>()
const responses: unknown[] = []
const { runtime, agent } = fixture(
{},
{
onPermission: (_request, context) => {
requested.resolve(context.signal)
return decision.promise
}
}
)
agent.on('session/prompt', (frame) => {
void agent
.request('permission', 'session/request_permission', permission)
.then((response) => responses.push(response.result))
agent.on('session/cancel', (cancel) => {
expect(cancel.id).toBeUndefined()
void agent
.request('late-permission', 'session/request_permission', permission)
.then((response) => {
responses.push(response.result)
agent.reply(frame, { stopReason: 'cancelled' })
})
})
})
await runtime.start(startOptions)
const prompt = runtime.prompt([...textPrompt])
const signal = await requested.promise
const cancel = runtime.cancel()
expect(runtime.cancel()).toBe(cancel)
expect(await prompt).toEqual({ stopReason: 'cancelled' })
await cancel
expect(signal.aborted).toBe(true)
expect(responses).toEqual([
{ outcome: { outcome: 'cancelled' } },
{ outcome: { outcome: 'cancelled' } }
])
decision.resolve({ outcome: { outcome: 'selected', optionId: 'allow' } })
await tick()
expect(agent.frames.filter((frame) => frame.id === 'permission')).toHaveLength(1)
})
it('surfaces authentication-required errors with code and data', async () => {
const { runtime, agent } = fixture()
agent.on('session/new', (frame) =>
agent.fail(frame, -32000, 'Login required', { detail: 'Sign in' })
)
await expect(runtime.start(startOptions)).rejects.toBeInstanceOf(AcpAuthRequiredError)
await expect(runtime.start(startOptions)).rejects.toMatchObject({
code: -32000,
data: { detail: 'Sign in' }
})
})
it('authenticates once with a configured agent method and retries session setup', async () => {
const { runtime, agent } = fixture()
let authenticated = false
agent.on('initialize', (frame) =>
agent.reply(frame, { protocolVersion: 1, authMethods: [{ id: 'login', name: 'Login' }] })
)
agent.on('session/new', (frame) =>
authenticated
? agent.reply(frame, { sessionId: 'session-1' })
: agent.fail(frame, -32000, 'Login required')
)
agent.on('authenticate', (frame) => {
authenticated = true
agent.reply(frame, {})
})
await runtime.start({ ...startOptions, authMethodId: 'login' })
expect(agent.frames.map((frame) => frame.method)).toEqual([
'initialize',
'session/new',
'authenticate',
'session/new'
])
expect(agent.frames[2].params).toEqual({ methodId: 'login' })
})
it('leaves advertised authentication methods for the caller to choose', async () => {
const { runtime, agent } = fixture()
const authMethods = [
{ id: 'terminal', name: 'Interactive login', type: 'terminal' },
{ id: 'agent', name: 'Agent login' }
]
agent.on('initialize', (frame) => agent.reply(frame, { protocolVersion: 1, authMethods }))
agent.on('session/new', (frame) => agent.fail(frame, -32000, 'Login required'))
await expect(runtime.start(startOptions)).rejects.toMatchObject({ authMethods })
expect(agent.frames.map((frame) => frame.method)).toEqual(['initialize', 'session/new'])
})
it('does not retry authentication indefinitely or start an interactive login', async () => {
const { runtime, agent } = fixture()
agent.on('initialize', (frame) =>
agent.reply(frame, {
protocolVersion: 1,
authMethods: [{ id: 'terminal', name: 'Interactive login', type: 'terminal' }]
})
)
agent.on('session/new', (frame) => agent.fail(frame, -32000, 'Login required'))
await expect(runtime.start(startOptions)).rejects.toBeInstanceOf(AcpAuthRequiredError)
expect(agent.frames.map((frame) => frame.method)).toEqual(['initialize', 'session/new'])
const retry = fixture()
retry.agent.on('initialize', (frame) =>
retry.agent.reply(frame, {
protocolVersion: 1,
authMethods: [{ id: 'agent', name: 'Agent login' }]
})
)
retry.agent.on('authenticate', (frame) => retry.agent.reply(frame, {}))
retry.agent.on('session/new', (frame) =>
retry.agent.fail(frame, -32000, 'Still requires login')
)
await expect(
retry.runtime.start({ ...startOptions, authMethodId: 'agent' })
).rejects.toBeInstanceOf(AcpAuthRequiredError)
expect(retry.agent.frames.map((frame) => frame.method)).toEqual([
'initialize',
'session/new',
'authenticate',
'session/new'
])
})
it('sends mode, model, and config changes to the active session', async () => {
const { runtime, agent } = fixture()
agent.on('session/set_mode', (frame) => agent.reply(frame, {}))
agent.on('session/set_model', (frame) => agent.reply(frame, {}))
agent.on('session/set_config_option', (frame) => {
SetSessionConfigOptionRequestSchema.parse(frame.params)
agent.reply(frame, { configOptions: [] })
})
await runtime.start(startOptions)
await runtime.setMode('plan')
await runtime.setModel('model-1')
await runtime.setConfigOption('thinking', 'high')
expect(agent.frames.slice(-3).map((frame) => frame.params)).toEqual([
{ sessionId: 'session-1', modeId: 'plan' },
{ sessionId: 'session-1', modelId: 'model-1' },
{ sessionId: 'session-1', configId: 'thinking', value: 'high' }
])
await runtime.setConfigOption('enabled', true)
expect(agent.frames.at(-1)?.params).toEqual({
sessionId: 'session-1',
configId: 'enabled',
value: true,
type: 'boolean'
})
})
it('rejects an unsupported protocol and malformed session responses', async () => {
const { runtime, agent } = fixture()
agent.on('initialize', (frame) => agent.reply(frame, { protocolVersion: 2 }))
await expect(runtime.start(startOptions)).rejects.toMatchObject({ code: -32602 })
expect(agent.frames.map((frame) => frame.method)).toEqual(['initialize'])
const malformed = fixture()
malformed.agent.on('session/new', (frame) => malformed.agent.reply(frame, {}))
await expect(malformed.runtime.start(startOptions)).rejects.toMatchObject({ code: -32603 })
})
it('rejects unroutable permission requests and answers an unavailable selection cancelled', async () => {
const diagnostics: string[] = []
const { runtime, agent } = fixture(
{},
{
onPermission: () => ({ outcome: { outcome: 'selected', optionId: 'not-offered' } }),
onDiagnostic: (message) => diagnostics.push(message)
}
)
agent.on('session/prompt', () => {})
await runtime.start(startOptions)
void runtime.prompt([...textPrompt]).catch(() => {})
expect(await agent.request('invalid', 'session/request_permission', {})).toMatchObject({
error: { code: -32602 }
})
expect(
await agent.request('no-options', 'session/request_permission', {
...permission,
options: [{ name: 'No id', kind: 'allow_once' }]
})
).toMatchObject({ error: { code: -32602 } })
expect(
await agent.request('bad-selection', 'session/request_permission', permission)
).toMatchObject({ result: { outcome: { outcome: 'cancelled' } } })
expect(diagnostics).toContain(
'Answered ACP permission request cancelled: handler selected an unavailable option'
)
// The session stays usable: a local permission failure is not a protocol failure.
await expect(runtime.prompt([...textPrompt])).rejects.toThrow('already in progress')
})
it('preserves unrecognized updates and isolates event listener failures', async () => {
const diagnostics: string[] = []
const { runtime, agent } = fixture({}, { onDiagnostic: (message) => diagnostics.push(message) })
const updates: AcpSessionEvent[] = []
runtime.subscribe(() => {
throw new Error('Consumer failed')
})
runtime.subscribe((event) => updates.push(event))
await runtime.start(startOptions)
agent.notify('session/update', {
sessionId: 'session-1',
update: { sessionUpdate: 'agent_message_chunk', content: {} }
})
agent.notify('session/update', {
sessionId: 'session-1',
update: {
sessionUpdate: 'agent_message_chunk',
content: { type: 'text', text: 'valid', _meta: { vendor: true } }
}
})
expect(updates).toHaveLength(2)
expect(updates[0]).toMatchObject({ kind: 'unrecognized' })
expect(updates[1]).toMatchObject({
kind: 'known',
notification: { update: { content: { _meta: { vendor: true } } } }
})
expect(diagnostics).toContain('Forwarded unrecognized ACP session update')
expect(diagnostics.some((message) => message.includes('Consumer failed'))).toBe(true)
})
it('rejects pending calls and aborts permission hooks when the agent exits', async () => {
const requested = deferred<AbortSignal>()
const { runtime, agent } = fixture(
{},
{
onPermission: (_request, context) => {
requested.resolve(context.signal)
return new Promise(() => {})
}
}
)
agent.on('session/prompt', () => {
void agent.request('permission', 'session/request_permission', permission)
})
await runtime.start(startOptions)
const pending = runtime.prompt([...textPrompt])
const rejected = expect(pending).rejects.toBeInstanceOf(AcpConnectionClosedError)
const signal = await requested.promise
agent.stdout.end()
await rejected
expect(signal.aborted).toBe(true)
await expect(runtime.setMode('plan')).rejects.toBeInstanceOf(AcpConnectionClosedError)
})
it('closes an unconfirmed cancellation rather than allowing another prompt', async () => {
vi.useFakeTimers()
const { runtime, agent } = fixture({}, { cancelTimeoutMs: 100 })
agent.on('session/prompt', () => {})
await runtime.start(startOptions)
const prompt = runtime.prompt([...textPrompt])
const rejectedPrompt = expect(prompt).rejects.toBeInstanceOf(AcpRequestTimeoutError)
const cancelled = expect(runtime.cancel()).rejects.toBeInstanceOf(AcpRequestTimeoutError)
await vi.advanceTimersByTimeAsync(100)
await Promise.all([rejectedPrompt, cancelled])
await expect(runtime.prompt([...textPrompt])).rejects.toBeInstanceOf(AcpRequestTimeoutError)
})
describe("a steer's cancel", () => {
const cancels = (agent: AcpScriptedAgent) =>
agent.frames.filter((frame) => frame.method === 'session/cancel')
it("asks once, never times out or closes, and the prompt's own reply ends it", async () => {
vi.useFakeTimers()
const signals: AbortSignal[] = []
const { runtime, agent } = fixture(
{},
{
cancelTimeoutMs: 100,
// Would allow every request it is asked; a steer's cancel keeps late ones from it.
onPermission: (_request, context) => {
signals.push(context.signal)
return new Promise<RequestPermissionResponse>(() => {})
}
}
)
const promptFrame = deferred<Parameters<AcpScriptedAgent['reply']>[0]>()
agent.on('session/prompt', (frame) => promptFrame.resolve(frame))
await runtime.start(startOptions)
const prompt = runtime.prompt([...textPrompt])
const frame = await promptFrame.promise
const open = agent.request('open', 'session/request_permission', permission)
await vi.waitFor(() => expect(signals).toHaveLength(1))
const steer = runtime.requestSteerCancel()
expect(runtime.requestSteerCancel()).toBe(steer)
await steer
await vi.advanceTimersByTimeAsync(1_000)
expect(cancels(agent)).toHaveLength(1)
expect(signals[0]?.aborted).toBe(true)
expect(await open).toMatchObject({ result: { outcome: { outcome: 'cancelled' } } })
expect(await agent.request('late', 'session/request_permission', permission)).toMatchObject({
result: { outcome: { outcome: 'cancelled' } }
})
expect(signals).toHaveLength(1)
agent.reply(frame, { stopReason: 'cancelled' })
expect(await prompt).toEqual({ stopReason: 'cancelled' })
// The connection is still open, so the steer's own prompt follows.
agent.on('session/prompt', (next) => agent.reply(next, { stopReason: 'end_turn' }))
expect(await runtime.prompt([...textPrompt])).toEqual({ stopReason: 'end_turn' })
})
it('leaves a later Stop bounded, closing the connection when the prompt never settles', async () => {
vi.useFakeTimers()
const { runtime, agent } = fixture({}, { cancelTimeoutMs: 100 })
agent.on('session/prompt', () => {})
await runtime.start(startOptions)
const prompt = runtime.prompt([...textPrompt])
const rejectedPrompt = expect(prompt).rejects.toBeInstanceOf(AcpRequestTimeoutError)
await runtime.requestSteerCancel()
const stopped = expect(runtime.cancel()).rejects.toBeInstanceOf(AcpRequestTimeoutError)
await vi.advanceTimersByTimeAsync(100)
await Promise.all([rejectedPrompt, stopped])
expect(cancels(agent)).toHaveLength(2)
await expect(runtime.prompt([...textPrompt])).rejects.toBeInstanceOf(AcpRequestTimeoutError)
})
it("writes nothing with no prompt of Orca's running", async () => {
const { runtime, agent } = fixture()
await expect(runtime.requestSteerCancel()).resolves.toBeUndefined()
await runtime.start(startOptions)
await runtime.requestSteerCancel()
await tick()
expect(cancels(agent)).toHaveLength(0)
})
})
})
+330
View File
@@ -0,0 +1,330 @@
import type { Readable, Writable } from 'node:stream'
import type { z } from 'zod'
import { MAX_TIMER_DELAY_MS } from '../../shared/timer-delay'
import {
AcpAgentError,
AcpAuthRequiredError,
AcpInvalidResponseError,
AcpRpcError
} from './acp-errors'
import { AcpJsonRpcPeer, type AcpPeerOptions, type AcpRequestContext } from './acp-json-rpc-peer'
import {
answerAcpPermission,
readAcpPermissionRequest,
type AcpPermissionHandler
} from './acp-permission-requests'
import { readAcpSessionEvent, type AcpSessionEvent } from './acp-session-events'
import {
cancelAcpPromptForStop,
requestAcpSteerCancel,
type AcpCancelChannel,
type ActivePrompt
} from './acp-prompt-cancel'
import {
setupAcpSession,
type AcpSessionStarted,
type AcpSessionStartOptions
} from './acp-session-setup'
export type { AcpSessionStarted, AcpSessionStartOptions } from './acp-session-setup'
export type { AcpSessionEvent } from './acp-session-events'
import {
ACP_PROTOCOL_VERSION,
InitializeResponseSchema,
AuthenticateResponseSchema,
PromptResponseSchema,
SetSessionModeResponseSchema,
SetSessionModelResponseSchema,
SetSessionConfigOptionResponseSchema,
type InitializeRequest,
type InitializeResponse,
type AuthenticateResponse,
type CancelNotification,
type PromptRequest,
type PromptResponse,
type SetSessionConfigOptionRequest,
type SetSessionConfigOptionResponse,
type SetSessionModeRequest,
type SetSessionModeResponse,
type SetSessionModelRequest,
type SetSessionModelResponse
} from './generated/acp-protocol.generated'
type Meta = PromptRequest['_meta']
const withMeta = (meta: Meta): { _meta?: Meta } => (meta ? { _meta: meta } : {})
export type AcpSessionRuntimeOptions = {
clientInfo?: InitializeRequest['clientInfo']
peer?: AcpPeerOptions
cancelTimeoutMs?: number
onPermission?: AcpPermissionHandler
/** Agent requests other than permissions. The handler owns its request: once `context.signal`
* aborts, send the agent's own cancelled reply, finish an answer already in progress, or throw
* (-32800). The runtime never answers for it; a request left unanswered ends at `close()`. */
onRequest?: (method: string, params: unknown, context: AcpRequestContext) => unknown
/** Agent notifications other than `session/update` (protocol extensions), delivered
* synchronously in arrival order with the `subscribe` events. */
onExtensionNotification?: (method: string, params: unknown) => void
onDiagnostic?: (message: string) => void
onClose?: (error: Error) => void
}
export class AcpSessionRuntime {
private readonly peer: AcpJsonRpcPeer
private readonly listeners = new Set<(event: AcpSessionEvent) => void>()
private initialized?: Promise<InitializeResponse>
private starting?: Promise<AcpSessionStarted>
private started?: AcpSessionStarted
private activePrompt?: ActivePrompt
private reportedUpdateAnomaly = false
constructor(
input: Readable,
output: Writable,
private readonly options: AcpSessionRuntimeOptions = {}
) {
const timeout = options.cancelTimeoutMs
if (timeout !== undefined && (!Number.isSafeInteger(timeout) || timeout <= 0)) {
throw new Error('ACP timeouts must be positive finite timer durations')
}
this.peer = new AcpJsonRpcPeer(
input,
output,
{
onRequest: (method, params, context) => this.handleRequest(method, params, context),
onNotification: (method, params) => this.handleNotification(method, params),
onDiagnostic: options.onDiagnostic,
onClose: options.onClose
},
options.peer
)
}
subscribe(listener: (event: AcpSessionEvent) => void): () => void {
this.listeners.add(listener)
return () => {
this.listeners.delete(listener)
}
}
initialize(): Promise<InitializeResponse> {
this.initialized ??= this.call(
'initialize',
{
protocolVersion: ACP_PROTOCOL_VERSION,
clientCapabilities: { fs: { readTextFile: false, writeTextFile: false }, terminal: false },
...(this.options.clientInfo === undefined ? {} : { clientInfo: this.options.clientInfo })
} satisfies InitializeRequest,
InitializeResponseSchema
)
.then((response) => {
if (response.protocolVersion !== ACP_PROTOCOL_VERSION) {
const error = new AcpRpcError(
-32602,
`Unsupported ACP protocol version: ${response.protocolVersion}`
)
this.peer.close(error)
throw error
}
return response
})
.catch((error) => {
this.initialized = undefined
throw error
})
return this.initialized
}
async authenticate(methodId: string): Promise<AuthenticateResponse> {
await this.initialize()
return this.call('authenticate', { methodId }, AuthenticateResponseSchema)
}
start(options: AcpSessionStartOptions): Promise<AcpSessionStarted> {
if (this.started) {
return Promise.resolve(this.started)
}
this.starting ??= this.initialize()
.then((initialized) =>
setupAcpSession(initialized, options, (method, params, schema) =>
this.call(method, params, schema)
)
)
.then((started) => {
this.started = started
return started
})
.catch((error) => {
this.starting = undefined
throw error
})
return this.starting
}
async prompt(prompt: PromptRequest['prompt'], meta?: Meta): Promise<PromptResponse> {
if (this.activePrompt) {
throw new Error('ACP prompt already in progress')
}
const params: PromptRequest = { sessionId: this.sessionId(), prompt, ...withMeta(meta) }
const active: ActivePrompt = {
cancelling: false,
response: this.call('session/prompt', params, PromptResponseSchema)
}
this.activePrompt = active
active.response = active.response.finally(() => {
if (this.activePrompt === active) {
this.activePrompt = undefined
}
})
return active.response
}
/** A Stop's cancel (`cancelAcpPromptForStop`): bounded, and past the bound it closes. A prompt
* that settles in time leaves the agent running; the Stop's owner ends its process. */
cancel(options: { meta?: CancelNotification['_meta'] } = {}): Promise<void> {
const channel = this.cancelChannel(options.meta)
return channel
? cancelAcpPromptForStop(this.activePrompt, channel)
: Promise.reject(new Error('ACP session has not started'))
}
/** A steer's cancel (`requestAcpSteerCancel`): never bounded and never closes. */
requestSteerCancel(options: { meta?: CancelNotification['_meta'] } = {}): Promise<void> {
const channel = this.cancelChannel(options.meta)
return channel ? requestAcpSteerCancel(this.activePrompt, channel) : Promise.resolve()
}
async setMode(modeId: string, meta?: Meta): Promise<SetSessionModeResponse> {
const params: SetSessionModeRequest = { sessionId: this.sessionId(), modeId, ...withMeta(meta) }
return this.call('session/set_mode', params, SetSessionModeResponseSchema)
}
async setModel(modelId: string, meta?: Meta): Promise<SetSessionModelResponse> {
const params: SetSessionModelRequest = {
sessionId: this.sessionId(),
modelId,
...withMeta(meta)
}
return this.call('session/set_model', params, SetSessionModelResponseSchema)
}
async setConfigOption(
configId: SetSessionConfigOptionRequest['configId'],
value: SetSessionConfigOptionRequest['value'],
meta?: Meta
): Promise<SetSessionConfigOptionResponse> {
const sessionId = this.sessionId()
const request =
typeof value === 'boolean'
? ({
configId,
value,
sessionId,
type: 'boolean',
...withMeta(meta)
} satisfies SetSessionConfigOptionRequest)
: ({
configId,
value,
sessionId,
...withMeta(meta)
} satisfies SetSessionConfigOptionRequest)
return this.call('session/set_config_option', request, SetSessionConfigOptionResponseSchema)
}
// The process owner must call close on child exit, even if descendants keep stdio open.
close(error?: Error): void {
this.peer.close(error)
this.listeners.clear()
}
private cancelChannel(meta: CancelNotification['_meta']): AcpCancelChannel | null {
if (!this.started) {
return null
}
const params: CancelNotification = { sessionId: this.started.sessionId, ...withMeta(meta) }
return {
send: () => this.peer.notify('session/cancel', params),
cancelIncomingRequests: () => this.peer.cancelIncomingRequests(),
close: (error) => this.peer.close(error),
closed: () => this.peer.closed,
timeoutMs: Math.min(this.options.cancelTimeoutMs ?? 10_000, MAX_TIMER_DELAY_MS)
}
}
private sessionId(): string {
if (!this.started) {
throw new Error('ACP session has not started')
}
return this.started.sessionId
}
private async call<T>(method: string, params: unknown, schema: z.ZodType<T>): Promise<T> {
const result = await this.peer.request(method, params, { timeoutMs: null }).catch((error) => {
if (error instanceof AcpAgentError && error.code === -32000) {
throw new AcpAuthRequiredError(error.message, error.data)
}
throw error
})
const parsed = schema.safeParse(result)
if (!parsed.success) {
throw new AcpInvalidResponseError(
`Invalid ACP response: ${method}`,
result,
parsed.error.issues
)
}
return parsed.data
}
private handleRequest(method: string, params: unknown, context: AcpRequestContext): unknown {
if (method !== 'session/request_permission') {
if (!this.options.onRequest) {
throw new AcpRpcError(-32601, `Unknown ACP client method: ${method}`)
}
return this.options.onRequest(method, params, context)
}
const diagnose = (message: string): void => this.diagnose(message)
const request = readAcpPermissionRequest(params, diagnose)
if (!request) {
throw new AcpRpcError(-32602, 'Invalid ACP permission request')
}
// Whether a turn the agent began itself may ask is the caller's call; the runtime cannot see it.
if (this.activePrompt?.cancelling || request.sessionId !== this.started?.sessionId) {
return { outcome: { outcome: 'cancelled' } }
}
return answerAcpPermission(request, context, this.options.onPermission, diagnose)
}
private diagnose(message: string): void {
try {
this.options.onDiagnostic?.(message)
} catch {
/* Diagnostics cannot prevent event delivery. */
}
}
private handleNotification(method: string, params: unknown): void {
if (method !== 'session/update') {
try {
this.options.onExtensionNotification?.(method, params)
} catch (error) {
this.diagnose(`ACP extension listener failed: ${String(error)}`)
}
return
}
const event = readAcpSessionEvent(params)
if (!event) {
this.diagnose('Ignored invalid ACP session update envelope')
return
}
if (event.kind === 'unrecognized' && !this.reportedUpdateAnomaly) {
this.reportedUpdateAnomaly = true
this.diagnose('Forwarded unrecognized ACP session update')
}
for (const listener of this.listeners) {
try {
listener(event)
} catch (error) {
this.diagnose(`ACP event listener failed: ${String(error)}`)
}
}
}
}
+85
View File
@@ -0,0 +1,85 @@
import type { z } from 'zod'
import { AcpAuthRequiredError, AcpRpcError } from './acp-errors'
import {
NewSessionResponseSchema,
LoadSessionResponseSchema,
ResumeSessionResponseSchema,
AuthenticateResponseSchema,
SessionModelStateSchema,
type InitializeResponse,
type NewSessionRequest
} from './generated/acp-protocol.generated'
const newSessionSchema = NewSessionResponseSchema.extend({
models: SessionModelStateSchema.optional()
})
const loadSessionSchema = LoadSessionResponseSchema.extend({
models: SessionModelStateSchema.optional()
})
const resumeSessionSchema = ResumeSessionResponseSchema.extend({
models: SessionModelStateSchema.optional()
})
export type AcpSessionStarted =
| { kind: 'new'; sessionId: string; response: z.infer<typeof newSessionSchema> }
| { kind: 'load'; sessionId: string; response: z.infer<typeof loadSessionSchema> }
| { kind: 'resume'; sessionId: string; response: z.infer<typeof resumeSessionSchema> }
export type AcpSessionStartOptions = NewSessionRequest & {
sessionId?: string
resumePreference?: 'load' | 'resume'
authMethodId?: string
}
type Request = <T>(method: string, params: unknown, schema: z.ZodType<T>) => Promise<T>
export async function setupAcpSession(
initialized: InitializeResponse,
options: AcpSessionStartOptions,
request: Request
): Promise<AcpSessionStarted> {
const setup = async (): Promise<AcpSessionStarted> => {
const { sessionId, resumePreference, authMethodId: _auth, ...params } = options
if (sessionId === undefined) {
const response = await request('session/new', params, newSessionSchema)
return { kind: 'new', sessionId: response.sessionId, response }
}
const capabilities = initialized.agentCapabilities
const load = capabilities?.loadSession === true
const resume = capabilities?.sessionCapabilities?.resume != null
if (load && (resumePreference !== 'resume' || !resume)) {
return {
kind: 'load',
sessionId,
response: await request('session/load', { ...params, sessionId }, loadSessionSchema)
}
}
if (resume) {
return {
kind: 'resume',
sessionId,
response: await request('session/resume', { ...params, sessionId }, resumeSessionSchema)
}
}
throw new AcpRpcError(-32601, 'ACP agent cannot load or resume this session')
}
try {
return await setup()
} catch (error) {
if (!(error instanceof AcpAuthRequiredError)) {
throw error
}
const required = new AcpAuthRequiredError(error.message, error.data, initialized.authMethods)
const method = initialized.authMethods?.find((method) => method.id === options.authMethodId)
if (!method) {
throw required
}
try {
await request('authenticate', { methodId: method.id }, AuthenticateResponseSchema)
return await setup()
} catch (failure) {
if (failure instanceof AcpAuthRequiredError) {
throw new AcpAuthRequiredError(failure.message, failure.data, initialized.authMethods)
}
throw failure
}
}
}
+16
View File
@@ -0,0 +1,16 @@
import type { Readable, Writable } from 'node:stream'
function ignoreLateError(): void {}
export function detachAcpStreamErrorHandler(
stream: Readable | Writable,
handler: (error: Error) => void
): void {
stream.removeListener('error', handler)
if (stream.closed) {
return
}
// Node may emit the write error after its callback has already closed the peer.
stream.on('error', ignoreLateError)
stream.once('close', () => stream.removeListener('error', ignoreLateError))
}
+119
View File
@@ -0,0 +1,119 @@
import type { Writable } from 'node:stream'
import { AcpConnectionClosedError } from './acp-errors'
type Write = {
line: string
resolve: () => void
reject: (error: Error) => void
detachAbort?: () => void
}
export class AcpWriteQueue {
private readonly queue: Write[] = []
private bytes = 0
private active?: Write
private terminalError?: Error
private detachDrain?: () => void
constructor(
private readonly output: Writable,
private readonly maxBytes: number,
private readonly onFailure: (error: Error) => void
) {}
write(line: string, signal?: AbortSignal): Promise<void> {
if (this.terminalError) {
return Promise.reject(this.terminalError)
}
if (signal?.aborted) {
return Promise.reject(signal.reason)
}
const bytes = Buffer.byteLength(line)
if (this.bytes + bytes > this.maxBytes) {
return Promise.reject(new Error('ACP write queue capacity exceeded'))
}
this.bytes += bytes
return new Promise((resolve, reject) => {
const write: Write = { line, resolve, reject }
const abort = (): void => {
const index = this.queue.indexOf(write)
if (index === -1) {
return
}
this.queue.splice(index, 1)
this.bytes -= bytes
write.detachAbort?.()
reject(signal?.reason)
}
signal?.addEventListener('abort', abort, { once: true })
write.detachAbort = () => signal?.removeEventListener('abort', abort)
this.queue.push(write)
this.flush()
})
}
close(error: Error): void {
if (this.terminalError) {
return
}
this.terminalError = error
this.detachDrain?.()
this.active?.reject(error)
this.active = undefined
for (const write of this.queue.splice(0)) {
write.detachAbort?.()
write.reject(error)
}
this.bytes = 0
}
private flush(): void {
if (this.active || this.terminalError) {
return
}
const write = this.queue.shift()
if (!write) {
return
}
write.detachAbort?.()
this.active = write
if (this.output.destroyed || !this.output.writable) {
this.onFailure(new AcpConnectionClosedError('ACP output is not writable'))
return
}
let completed = false
let drained = false
let returned = false
const finish = (): void => {
if (!returned || !completed || !drained || this.terminalError) {
return
}
this.detachDrain?.()
this.active = undefined
this.bytes -= Buffer.byteLength(write.line)
write.resolve()
this.flush()
}
const onDrain = (): void => {
drained = true
finish()
}
this.output.once('drain', onDrain)
this.detachDrain = () => this.output.removeListener('drain', onDrain)
try {
const accepted = this.output.write(write.line, (error) => {
if (error) {
this.onFailure(error)
return
}
completed = true
finish()
})
drained ||= accepted
returned = true
finish()
} catch (error) {
this.onFailure(error instanceof Error ? error : new Error(String(error)))
}
}
}
File diff suppressed because it is too large Load Diff
@@ -7,7 +7,7 @@ import {
CodexAppServerUnsupportedError,
isCodexMethodNotFoundError
} from './codex-app-server-session'
import { classifyJsonRpcPrefix } from './codex-app-server-record-prefix'
import { classifyJsonRpcPrefix } from '../../shared/json-rpc-record-prefix'
const OVERSIZED_REQUEST_ERROR_CODE = -32001
const MAX_REMEMBERED_TIMEOUTS = 64