From a0f2ddec50ca55b6c266c4bbf95925152f77e3d4 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Tue, 27 Oct 2026 18:10:16 -0700 Subject: [PATCH 1/3] feat(seats): ACP path for the supervisor Squash of the 31 commits of #2857, rebased onto main. main's failure kinds (not_authenticated, quota, provider_unsupported, capacity) sit beside this change's refusal_retry_loop and provider_unavailable. The tests' fake keys are assembled at runtime, so no key-shaped literal is committed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MDs9DRK3nqxjeAV7xL8uhg --- cli/package.json | 2 +- packages/seats/src/acp/client.js | 161 ++++ packages/seats/src/adapters/acp-launch.js | 211 +++++ packages/seats/src/adapters/acp.js | 613 ++++++++++++++ packages/seats/src/failures.js | 2 + packages/seats/src/index.js | 9 + packages/seats/src/provider-watch.js | 131 +++ packages/seats/src/trace/events.js | 3 +- .../seats/test-support/fake-acp-agent.mjs | 256 ++++++ packages/seats/test-support/record-trace.js | 29 + packages/seats/test/acp-real.test.js | 267 ++++++ packages/seats/test/acp.test.js | 794 ++++++++++++++++++ packages/seats/test/provider-watch.test.js | 135 +++ 13 files changed, 2611 insertions(+), 2 deletions(-) create mode 100644 packages/seats/src/acp/client.js create mode 100644 packages/seats/src/adapters/acp-launch.js create mode 100644 packages/seats/src/adapters/acp.js create mode 100644 packages/seats/src/provider-watch.js create mode 100644 packages/seats/test-support/fake-acp-agent.mjs create mode 100644 packages/seats/test-support/record-trace.js create mode 100644 packages/seats/test/acp-real.test.js create mode 100644 packages/seats/test/acp.test.js create mode 100644 packages/seats/test/provider-watch.test.js diff --git a/cli/package.json b/cli/package.json index fc5a0e3441..10435c0f9c 100644 --- a/cli/package.json +++ b/cli/package.json @@ -2,7 +2,7 @@ "name": "@commonlyai/cli", "version": "0.1.168", "license": "Apache-2.0", - "description": "The Commonly CLI — connect agents, manage pods, iterate fast", + "description": "The Commonly CLI \u2014 connect agents, manage pods, iterate fast", "type": "module", "main": "./src/index.js", "bin": { diff --git a/packages/seats/src/acp/client.js b/packages/seats/src/acp/client.js new file mode 100644 index 0000000000..57053160f8 --- /dev/null +++ b/packages/seats/src/acp/client.js @@ -0,0 +1,161 @@ +/** + * Newline-delimited JSON-RPC 2.0 over a child's stdio, as ACP's stdio + * transport defines it: one message per line, no embedded newlines. + * + * The client owns framing and id bookkeeping only. What a notification or an + * agent→client request MEANS is the adapter's business, so both arrive through + * handlers. A request with no handler answer is -32601, which is how this + * client declines fs/* and terminal/*: it advertised neither in initialize, so + * an agent that asks anyway gets the standard "method not found". + */ +import { StringDecoder } from 'node:string_decoder'; +import { runFailure } from '../failures.js'; + +/** ACP lines carry tool output and diffs, so the cap is generous, not tight. */ +export const DEFAULT_MAX_LINE_CHARS = 8 * 1024 * 1024; + +export const METHOD_NOT_FOUND = -32601; +const INTERNAL_ERROR = -32603; + +/** A JSON-RPC error response, kept whole so the adapter can classify it. */ +export class AcpRpcError extends Error { + constructor(method, error) { + super(`${method} failed (${error?.code}): ${error?.message || 'no message'}`); + this.name = 'AcpRpcError'; + this.method = method; + this.code = error?.code; + this.rpcMessage = typeof error?.message === 'string' ? error.message : ''; + this.data = error?.data; + } +} + +/** Throw this from an onRequest handler to answer with a specific error. */ +export const rpcError = (code, message) => Object.assign(new Error(message), { rpcCode: code }); + +const isObject = (value) => value !== null && typeof value === 'object' && !Array.isArray(value); +const hasId = (message) => Object.prototype.hasOwnProperty.call(message, 'id') && message.id !== null; + +/** + * @param {object} opts + * @param {import('node:stream').Readable} opts.input the child's stdout + * @param {import('node:stream').Writable} opts.output the child's stdin + * @param {(method: string, params: any) => void} [opts.onNotification] + * @param {(method: string, params: any) => Promise | any} [opts.onRequest] + * resolve with the result; throw `rpcError(code, message)` to answer an error; + * resolve `undefined` for "not mine" and the client answers -32601. + * @param {(failure: import('../failures.js').RunFailure) => void} [opts.onProtocolError] + * @param {number} [opts.maxLineChars] + */ +export const createAcpClient = ({ + input, output, onNotification = () => {}, onRequest = () => undefined, + onProtocolError = () => {}, maxLineChars = DEFAULT_MAX_LINE_CHARS, +}) => { + const pending = new Map(); + const decoder = new StringDecoder('utf8'); + let buffer = ''; + let nextId = 1; + let broken = null; + + const write = (message) => { + if (broken || output.destroyed || output.writableEnded) return false; + output.write(`${JSON.stringify({ jsonrpc: '2.0', ...message })}\n`); + return true; + }; + + // Every in-flight request learns why the connection ended, so no caller + // waits on a response that cannot come. + const fail = (failure) => { + if (broken) return; + broken = failure; + for (const { reject } of pending.values()) reject(failure); + pending.clear(); + }; + + const protocolError = (detail) => { + const failure = runFailure('protocol', detail); + fail(failure); + onProtocolError(failure); + }; + + const answerRequest = async (message) => { + const { id, method, params } = message; + try { + const result = await onRequest(method, params); + if (result === undefined) { + write({ id, error: { code: METHOD_NOT_FOUND, message: `method not supported by this client: ${method}` } }); + } else { + write({ id, result }); + } + } catch (error) { + write({ id, error: { code: error?.rpcCode ?? INTERNAL_ERROR, message: error?.message || 'client error' } }); + } + }; + + const consume = (line) => { + let message; + try { message = JSON.parse(line); } catch { + protocolError(`agent wrote a line that is not JSON (${line.length} chars)`); + return; + } + if (!isObject(message) || message.jsonrpc !== '2.0') { + protocolError('agent wrote a message that is not JSON-RPC 2.0'); + return; + } + if (typeof message.method === 'string') { + if (hasId(message)) answerRequest(message); + // Unknown notifications are ignored by contract: ACP adds update kinds + // faster than any client learns them. + else onNotification(message.method, message.params); + return; + } + if (!hasId(message) || !pending.has(message.id)) { + protocolError(`agent answered a request id it was never sent: ${JSON.stringify(message.id ?? null)}`); + return; + } + const { resolve, reject, method } = pending.get(message.id); + // JSON-RPC 2.0: an error is an object whose code is an integer. Anything + // else is a broken answer, not an error the adapter can classify, and it + // fails every pending request with it. + if (message.error != null && (!isObject(message.error) || !Number.isInteger(message.error.code))) { + protocolError(`agent answered ${method} with an error whose code is not an integer: ${JSON.stringify(message.error?.code ?? null).slice(0, 40)}`); + return; + } + pending.delete(message.id); + if (message.error != null) reject(new AcpRpcError(method, message.error)); + else resolve(message.result); + }; + + input.on('data', (chunk) => { + if (broken) return; + buffer += decoder.write(chunk); + let newline = buffer.indexOf('\n'); + while (newline !== -1) { + if (newline > maxLineChars) { + protocolError(`agent line exceeded ${maxLineChars} chars`); + return; + } + const line = buffer.slice(0, newline).trim(); + buffer = buffer.slice(newline + 1); + if (line) consume(line); + if (broken) return; + newline = buffer.indexOf('\n'); + } + if (buffer.length > maxLineChars) protocolError(`agent line exceeded ${maxLineChars} chars`); + }); + + return { + request(method, params) { + if (broken) return Promise.reject(broken); + const id = nextId; + nextId += 1; + return new Promise((resolve, reject) => { + pending.set(id, { resolve, reject, method }); + write({ id, method, params }); + }); + }, + notify(method, params) { return write({ method, params }); }, + /** End the connection; pending requests reject with `failure`. */ + close(failure) { fail(failure); }, + get broken() { return broken; }, + }; +}; diff --git a/packages/seats/src/adapters/acp-launch.js b/packages/seats/src/adapters/acp-launch.js new file mode 100644 index 0000000000..e65857855d --- /dev/null +++ b/packages/seats/src/adapters/acp-launch.js @@ -0,0 +1,211 @@ +/** + * Launch functions for ACP agents. Each one turns a run ctx into a command, an + * isolated env and a cleanup; the protocol itself is adapters/acp.js. + * + * The provider is an OpenAI-compatible route from `ctx.provider = { baseURL, + * apiKey, model }`. The key travels in the child env (goose) or a 0600 file + * the agent's config points at (opencode), never in argv, which any same-user + * process can read. + * + * HOME comes from ctx.env.HOME when the caller gives one, because that is + * where an agent keeps the sessions a later resume needs. Without one, the + * spawn gets a throwaway HOME, so nothing of the operator's leaks in and + * nothing of the run survives it. + */ +import { spawnSync } from 'node:child_process'; +import { chmod, mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { runFailure } from '../failures.js'; +import { createAcpAdapter, readsAsRefusal } from './acp.js'; + +const SAFE_ENV = /^(?:PATH|LANG|LC_[A-Z_]+|TERM|USER|LOGNAME|NO_COLOR|CI|TMPDIR|SSL_CERT_FILE|SSL_CERT_DIR)$/; +const VERSION_TIMEOUT_MS = 5000; + +/** The provider route, checked: a plain HTTP(S) base URL, a key and a model. */ +export const providerOf = (ctx, label) => { + const { baseURL, apiKey, model } = ctx?.provider || {}; + let url; + try { url = new URL(String(baseURL)); } catch { + throw runFailure('unsupported', `${label}: ctx.provider.baseURL must be an HTTP(S) URL`); + } + if (!/^https?:$/.test(url.protocol) || url.username || url.password || url.search || url.hash) { + throw runFailure('unsupported', `${label}: ctx.provider.baseURL must be a plain HTTP(S) URL with no credentials, query or fragment`); + } + if (typeof apiKey !== 'string' || !apiKey) throw runFailure('unsupported', `${label}: ctx.provider.apiKey is missing`); + if (typeof model !== 'string' || !model) throw runFailure('unsupported', `${label}: ctx.provider.model is missing`); + return { url, apiKey, model }; +}; + +/** + * Safe launch variables from the caller's env, plus HOME and XDG roots. Only + * the safe names are read from process.env, and only when ctx.env is absent. + */ +export const isolatedEnv = async (ctx, label) => { + const source = ctx.env || process.env; + const env = Object.fromEntries(Object.entries(source).filter(([name]) => SAFE_ENV.test(name))); + const callerHome = ctx.env?.HOME; + const home = callerHome || await mkdtemp(join(tmpdir(), `seats-${label}-home-`)); + env.HOME = home; + env.XDG_CONFIG_HOME = ctx.env?.XDG_CONFIG_HOME || join(home, '.config'); + env.XDG_DATA_HOME = ctx.env?.XDG_DATA_HOME || join(home, '.local', 'share'); + env.XDG_STATE_HOME = ctx.env?.XDG_STATE_HOME || join(home, '.local', 'state'); + env.XDG_CACHE_HOME = ctx.env?.XDG_CACHE_HOME || join(home, '.cache'); + const removeHome = callerHome ? async () => {} : () => rm(home, { recursive: true, force: true }); + return { env, removeHome }; +}; + +const detectVersion = (binary, pattern) => async () => { + const probe = spawnSync(binary, ['--version'], { encoding: 'utf8', timeout: VERSION_TIMEOUT_MS }); + if (probe.error || probe.status !== 0) return null; + const version = pattern.exec(String(probe.stdout || ''))?.[1] || 'unknown'; + return { path: binary, version }; +}; + +/** + * goose 1.54.0 ends a turn whose provider call failed with stopReason + * end_turn and the error as its reply: "Ran into this error: …". Without this + * reader that error would be posted as the agent's answer. Measured against + * the fake provider: a 400 content-policy body, a 500 and a 429 all take this + * shape, after goose's own retries. + */ +export const readGooseReplyFailure = (text, { credentials = [] } = {}) => { + const match = /^\s*Ran into this error:\s*([\s\S]*)$/.exec(String(text ?? '')); + if (!match) return null; + const detail = match[1].split('\n\nPlease retry')[0].trim(); + return runFailure(readsAsRefusal(detail) ? 'refusal' : 'protocol', + `goose ended the turn on a provider error: ${detail}`, { credentials, limit: 500 }); +}; + +/** `goose acp` with goose's OpenAI provider pointed at ctx.provider. */ +export const launchGoose = async (ctx = {}) => { + const { url, apiKey, model } = providerOf(ctx, 'goose'); + const { env, removeHome } = await isolatedEnv(ctx, 'goose'); + Object.assign(env, { + GOOSE_PROVIDER: 'openai', + GOOSE_MODEL: model, + // approve: every tool call asks over session/request_permission, so the + // caller's decidePermission (or the reject default) is the policy. + GOOSE_MODE: 'approve', + GOOSE_TELEMETRY_ENABLED: 'false', + // No keychain prompts or keychain reads: the key arrives in env only. + GOOSE_DISABLE_KEYRING: '1', + OPENAI_HOST: url.origin, + OPENAI_BASE_PATH: `${url.pathname.replace(/\/$/, '')}/chat/completions`.replace(/^\//, ''), + OPENAI_API_KEY: apiKey, + }); + return { + command: ctx._binaryPath || 'goose', args: ['acp'], env, cwd: ctx.cwd, cleanup: removeHome, + }; +}; + +const OPENCODE_PROVIDER_ID = 'seats'; +const OPENCODE_DISABLED_ENV = [ + 'OPENCODE_DISABLE_CLAUDE_CODE', + 'OPENCODE_DISABLE_CLAUDE_CODE_PROMPT', + 'OPENCODE_DISABLE_CLAUDE_CODE_SKILLS', + 'OPENCODE_DISABLE_EXTERNAL_SKILLS', + 'OPENCODE_DISABLE_DEFAULT_PLUGINS', + 'OPENCODE_DISABLE_AUTOUPDATE', +]; + +/** + * The opencode config for one spawn: a single enabled openai-compatible + * provider and it as the default model. Same shape as the cli opencode + * adapter's buildProviderConfig; the default model matters more here, because + * `opencode acp` has no -m flag and its built-in default is a hosted service. + */ +export const opencodeConfig = ({ baseURL, model, keyFile }) => ({ + share: 'disabled', + autoupdate: false, + enabled_providers: [OPENCODE_PROVIDER_ID], + model: `${OPENCODE_PROVIDER_ID}/${model}`, + provider: { + [OPENCODE_PROVIDER_ID]: { + npm: '@ai-sdk/openai-compatible', + options: { baseURL, apiKey: `{file:${keyFile}}` }, + models: { [model]: { name: model } }, + }, + }, +}); + +/** + * opencode 2.x reports each provider retry as a session_info_update whose + * `_meta["opencode/retry"]` is {attempt, nextRetryAt, error: {type, message, + * status, response: {body}}}, and sends the key as null when the waited-for + * attempt goes out. Only attempt, nextRetryAt, status and the error's type + * are read: message and body are provider text. + */ +export const readOpencodeRetry = (update) => { + const retry = update?._meta?.['opencode/retry']; + if (retry === null) return { state: 'started' }; + if (typeof retry !== 'object') return null; + return { + state: 'scheduled', + attempt: retry.attempt, + nextRetryAt: retry.nextRetryAt, + status: retry.error?.status, + errorType: retry.error?.type, + }; +}; + +/** `opencode acp` with a generated config; project and global config are off. */ +export const launchOpencode = async (ctx = {}) => { + const { url, apiKey, model } = providerOf(ctx, 'opencode'); + const { env, removeHome } = await isolatedEnv(ctx, 'opencode'); + const dir = await mkdtemp(join(tmpdir(), 'seats-opencode-')); + try { + await chmod(dir, 0o700); + const keyFile = join(dir, 'provider-key'); + const configPath = join(dir, 'opencode.json'); + const configDir = join(dir, 'config-dir'); + const xdgConfig = join(dir, 'xdg-config'); + await Promise.all([mkdir(configDir), mkdir(join(xdgConfig, 'opencode'), { recursive: true })]); + await writeFile(keyFile, apiKey, { mode: 0o600, flag: 'wx' }); + await writeFile(configPath, `${JSON.stringify(opencodeConfig({ baseURL: url.href.replace(/\/$/, ''), model, keyFile }), null, 2)}\n`, { mode: 0o600, flag: 'wx' }); + Object.assign(env, { + OPENCODE_CONFIG: configPath, + OPENCODE_CONFIG_DIR: configDir, + OPENCODE_DISABLE_PROJECT_CONFIG: '1', + OPENCODE_AUTO_SHARE: 'false', + // opencode merges global config from XDG_CONFIG_HOME; point it at an + // empty dir so only the generated file counts. Data (sessions) stays + // under the caller's XDG_DATA_HOME, which is what a resume reads. + XDG_CONFIG_HOME: xdgConfig, + ...Object.fromEntries(OPENCODE_DISABLED_ENV.map((name) => [name, '1'])), + }); + } catch (error) { + await rm(dir, { recursive: true, force: true }); + await removeHome(); + throw error; + } + return { + command: ctx._binaryPath || 'opencode', + args: ['acp'], + env, + cwd: ctx.cwd, + cleanup: async () => { + await rm(dir, { recursive: true, force: true }); + await removeHome(); + }, + }; +}; + +export const createAcpGooseAdapter = (host = {}) => createAcpAdapter({ + ...host, + name: 'acp-goose', + launch: launchGoose, + detect: detectVersion('goose', /(\d+\.\d+\.\d+)/), + readReplyFailure: readGooseReplyFailure, +}); + +export const createAcpOpencodeAdapter = (host = {}) => createAcpAdapter({ + ...host, + name: 'acp-opencode', + launch: launchOpencode, + readRetryStatus: readOpencodeRetry, + detect: detectVersion('opencode', /(\d+\.\d+\.\d+(?:-[0-9A-Za-z.-]+)?)/), +}); + +export const acpGoose = createAcpGooseAdapter(); +export const acpOpencode = createAcpOpencodeAdapter(); diff --git a/packages/seats/src/adapters/acp.js b/packages/seats/src/adapters/acp.js new file mode 100644 index 0000000000..83fa793716 --- /dev/null +++ b/packages/seats/src/adapters/acp.js @@ -0,0 +1,613 @@ +/** + * A generic adapter for any agent that speaks the Agent Client Protocol over + * stdio. One spawn is one process and one prompt turn: + * + * initialize → session/resume | session/load | session/new → session/prompt + * + * The agent-specific part is only how to launch it (`launch(ctx)`), so a new + * ACP agent is a launch function, not a new adapter. The agent's + * session/update stream maps onto ctx.trace one kind to one event, because the + * trace vocabulary was named after ACP's update kinds. + */ +import { spawn as childSpawn } from 'node:child_process'; +import { AcpRpcError, createAcpClient, DEFAULT_MAX_LINE_CHARS } from '../acp/client.js'; +import { RunFailure, runFailure } from '../failures.js'; +import { startProviderWatch } from '../provider-watch.js'; +import { traceOf } from '../trace/events.js'; +import { + argsDigestOf, isUserId, PARKED_CALL_ROLES, policyActorId, RETRY_ERROR_TYPES, +} from '../record/format.js'; +import { redactKnownCredentials, scrubAdapterFailure, spawnCredentials } from '../upstream-refusal.js'; + +export const ACP_PROTOCOL_VERSION = 1; +const DEFAULT_TIMEOUT_MS = 15 * 60 * 1000; +// After session/cancel the agent owes a prompt response with stopReason +// cancelled; this is how long it gets before the process is killed. +const CANCEL_GRACE_MS = 3000; +// On a normal end the agent gets a moment after stdin closes to exit by +// itself, so a session store it writes on shutdown is complete for a resume. +const CLOSE_GRACE_MS = 1000; +const KILL_GRACE_MS = 2000; +// A provider failure seen on the wire already decided the run; the agent gets +// only a moment to answer the cancel, so the run ends in about a second. +const WIRE_CANCEL_GRACE_MS = 300; +// 'exit' can arrive before the last stdout line is read; give 'close' this +// long before declaring the turn unsettled. +const EXIT_DRAIN_MS = 200; +const MAX_STDERR_CHARS = 4096; +const STDERR_IN_FAILURE = 400; +const FAILURE_LIMIT = 500; +const SECRET_NAME = /(?:^|_)(?:API_KEY|ACCESS_TOKEN|ACCESS_KEY|CLIENT_SECRET|PASSWORD|TOKEN|SECRET)$/i; +const REFUSAL_WORDING = /content[ _-]?policy|refus|rejected by|rate limit/i; +const STOP_FAILURES = Object.freeze({ + refusal: 'refusal', + max_tokens: 'max_tokens', + max_turn_requests: 'max_turn_requests', + cancelled: 'cancelled', +}); + +// The agent leads its own process group, so a kill reaches what it spawned: +// a CLI launcher's native binary, its MCP servers, a tool's shell. Killing +// only the leader can leave a grandchild holding the run's work and keys. +const OWN_GROUP = process.platform !== 'win32'; + +/** Signal the whole group; false when it has no members left (or no group). */ +const signalGroup = (child, signal) => { + if (!OWN_GROUP || child.pid === undefined) return false; + try { process.kill(-child.pid, signal); return true; } catch { return false; } +}; + +const sleep = (ms) => new Promise((resolve) => { setTimeout(resolve, ms).unref?.(); }); + +/** + * Every value this spawn holds that must never reach a failure detail: the + * runtime token, the provider key, and any secret-named variable in either the + * caller's env or the env the launch built. + */ +export const acpCredentials = (ctx = {}, launchEnv = {}) => { + const fromEnv = (env) => Object.entries(env || {}) + .filter(([name]) => SECRET_NAME.test(name)).map(([, value]) => value); + return [...new Set([ + ...spawnCredentials({ ...ctx, env: ctx.env || {} }, []), + ctx.provider?.apiKey, + ...fromEnv(ctx.env), + ...fromEnv(launchEnv), + ].filter((value) => typeof value === 'string' && value.length >= 8))]; +}; + +/** + * Does this error text read as the model route refusing? A 4xx the text names, + * or refusal wording. Agents disagree on the shape — opencode 1.19.0 answers + * the prompt with -32603 "Internal error: The request was rejected by the + * content policy." and no status; goose 1.54.0 writes "Bad request (400)" — + * so the parenthesised status is read too. + */ +export const readsAsRefusal = (text) => { + const raw = String(text ?? ''); + const status = scrubAdapterFailure(raw).status ?? Number(/\((\d{3})\)/.exec(raw)?.[1] ?? NaN); + return (status >= 400 && status <= 499) || REFUSAL_WORDING.test(raw); +}; + +/** A JSON-RPC error answer, as a named failure. */ +export const failureFromRpcError = (error, credentials) => { + const kind = readsAsRefusal(error.rpcMessage) ? 'refusal' : 'protocol'; + return runFailure(kind, `${error.method} answered error ${error.code}: ${error.rpcMessage}`, { + credentials, limit: FAILURE_LIMIT, + }); +}; + +/** The reject default: a reject_once option if offered, otherwise cancelled. */ +export const defaultPermissionDecision = (options = []) => { + const reject = options.find((option) => option?.kind === 'reject_once'); + return reject ? { outcome: 'selected', optionId: reject.optionId } : { outcome: 'cancelled' }; +}; + +/** A hook answer is used only when it names an option the agent offered; else null. */ +const checkedDecision = (decision, options) => { + if (decision?.outcome === 'cancelled') return { outcome: 'cancelled' }; + if (decision?.outcome === 'selected' && options.some((option) => option?.optionId === decision.optionId)) { + return { outcome: 'selected', optionId: decision.optionId }; + } + return null; +}; + +/** How a permission ask pending at stop is answered: cancelled, as ACP requires. */ +const ABORTED_ANSWER = Object.freeze({ decision: Object.freeze({ outcome: 'cancelled' }), by: 'aborted' }); + +const isAsker = (value) => value !== null && typeof value === 'object' + && isUserId(value.id) && PARKED_CALL_ROLES.includes(value.role); + +/** + * The roles a hook's answer may name: a person's role on a parked call. Not + * `policy` (the policy deciding is never an answer) and not `pod-member` + * (ADR-033 D5: a pod member's word decides no parked call). + */ +const ANSWER_ROLES = Object.freeze(PARKED_CALL_ROLES.filter((role) => role !== 'policy')); +/** A hook answer the adapter records must say who gave it, as the valet's must. */ +const isAttributed = (raw) => isUserId(raw?.decidedBy) && ANSWER_ROLES.includes(raw?.decidedByRole); + +/** + * The record's human-readable ask: the tool call's kind and title, scrubbed of + * this spawn's credentials (it is agent text) and kept short. + */ +const questionOf = (toolCall, credentials) => { + const kind = typeof toolCall?.kind === 'string' && toolCall.kind ? toolCall.kind : 'tool'; + const title = typeof toolCall?.title === 'string' ? toolCall.title.trim() : ''; + const text = title ? `${kind}: ${title}` : `${kind} call`; + return redactKnownCredentials(text, credentials).slice(0, QUESTION_CHARS); +}; + +const abortFailure = (signal) => (signal?.reason instanceof RunFailure + ? signal.reason : runFailure('cancelled', 'the run was aborted')); + +const textOf = (content) => (content?.type === 'text' && typeof content.text === 'string' ? content.text : ''); + +const QUESTION_CHARS = 200; +/** A context count the record accepts: an integer >= 0, else null (unknown). */ +const count = (value) => (Number.isInteger(value) && value >= 0 ? value : null); + +const usageFromPrompt = (usage) => ({ + inputTokens: usage.inputTokens ?? null, + outputTokens: usage.outputTokens ?? null, + totalTokens: usage.totalTokens ?? null, + cost: null, +}); + +/** + * A retry report from the launch's reader, as the record takes it: each field + * only in its recorded shape, and absent otherwise. A type outside the + * record's closed set is `other`, so no agent text reaches errorType. + */ +const retryData = (report, turn) => { + const attempt = Number.isInteger(report.attempt) && report.attempt >= 1 ? report.attempt : undefined; + if (report.state === 'started') { + const started = attempt ?? turn.retryAttempt; + return { state: 'started', ...(started ? { attempt: started } : {}) }; + } + turn.retryAttempt = attempt; + const at = typeof report.nextRetryAt === 'string' ? new Date(report.nextRetryAt) : null; + const { status, errorType } = report; + return { + state: 'scheduled', + ...(attempt ? { attempt } : {}), + ...(at && !Number.isNaN(at.getTime()) ? { nextRetryAt: at.toISOString() } : {}), + ...(Number.isInteger(status) && status >= 100 && status <= 599 ? { status } : {}), + ...(errorType != null ? { errorType: RETRY_ERROR_TYPES.includes(errorType) ? errorType : 'other' } : {}), + }; +}; + +/** One session/update → at most one trace event. Unknown kinds are dropped. */ +const traceUpdate = (trace, update, turn, { readRetry } = {}) => { + switch (update?.sessionUpdate) { + case 'agent_message_chunk': { + const text = textOf(update.content); + if (!text) return; + turn.text += text; + trace.emit('agent.message_chunk', { text }); + return; + } + case 'agent_thought_chunk': + trace.emit('agent.thought_chunk', { chars: textOf(update.content).length }); + return; + case 'tool_call': + trace.emit('tool.call', { + toolCallId: update.toolCallId, title: update.title, kind: update.kind, status: update.status, + }); + return; + case 'tool_call_update': + trace.emit('tool.update', { toolCallId: update.toolCallId, status: update.status ?? undefined, title: update.title ?? undefined }); + return; + case 'plan': + trace.emit('plan', { + entries: (Array.isArray(update.entries) ? update.entries : []) + .map((entry) => ({ content: entry?.content, status: entry?.status })), + }); + return; + case 'usage_update': + turn.contextTokens = count(update.used) ?? turn.contextTokens; + trace.emit('usage', { + inputTokens: null, + outputTokens: null, + totalTokens: null, + cost: update.cost ? { amount: update.cost.amount, currency: update.cost.currency } : null, + contextTokens: count(update.used), + contextSize: count(update.size), + }); + break; + case 'session_info_update': { + // Session state (a title) unless the launch's reader finds a retry + // report in it; the protocol has no retry update of its own. + const report = readRetry?.(update); + if (report) trace.emit('agent.retry', retryData(report, turn)); + return; + } + default: + } +}; + +const runTurn = ({ + name, launched, prompt, ctx, memoryPreamble, readReplyFailure, readRetryStatus, maxLineChars, timings, wire, +}) => new Promise((resolveRun, rejectRun) => { + const trace = traceOf(ctx); + // A trace writer that refuses an entry (record-v1's emit throws on data it + // cannot record) ends the run with that error; it never escapes into a + // stream handler, where it would crash the host instead. + const emit = (type, data) => { + try { trace.emit(type, data); } catch (error) { stop(error); } + }; + const credentials = acpCredentials(ctx, launched.env); + const cwd = launched.cwd ?? ctx.cwd ?? process.cwd(); + const turn = { text: '', contextTokens: undefined, retryAttempt: undefined }; + const pendingPermissions = new Set(); + let stderr = ''; + // Set only once session/new|resume|load has answered. Updates are read only + // for this id, which is also what drops a load's replayed history: the + // agent streams it before answering, while this is still null. + let sessionId = null; + let promptInFlight = null; + let stopping = null; + let settled = false; + let exitInfo = null; + let resolveGone; + const gone = new Promise((resolve) => { resolveGone = resolve; }); + + let child; + try { + child = (ctx._spawnImpl || childSpawn)(launched.command, launched.args || [], { + cwd, env: launched.env, stdio: ['pipe', 'pipe', 'pipe'], detached: OWN_GROUP, + }); + } catch (error) { + rejectRun(runFailure('exited', `could not start ${name}: ${error?.code || error?.message}`, { credentials })); + return; + } + + const exitedFailure = () => { + const tail = stderr.trim().slice(-STDERR_IN_FAILURE); + const how = exitInfo ? `code ${exitInfo.code}, signal ${exitInfo.signal}` : 'no exit status'; + return runFailure('exited', `${name} exited before the turn settled (${how})${tail ? `: ${tail}` : ''}`, { + credentials, limit: FAILURE_LIMIT, + }); + }; + + // Who decides, for the record. With a hook, the run's approver (ctx.askedOf) + // is asked, and its answer is {outcome, optionId?, decidedBy, decidedByRole}: + // a user id and an owner role. Without a hook, or when the hook fails or + // names nobody, the reject default is the policy deciding. A decider that writes its own permission entries (the + // valet's policy) carries recordsPermissions = true, and then the adapter + // writes none, so each ask is recorded exactly once. + const hasHook = typeof ctx.decidePermission === 'function'; + const grantId = typeof ctx.grantId === 'string' && ctx.grantId ? ctx.grantId : undefined; + const policy = { id: policyActorId(grantId), role: 'policy' }; + const recordPermissions = ctx.decidePermission?.recordsPermissions !== true; + + /** Resolves { decision, by }: by is 'hook' (with its attribution), 'default', 'failed' or 'aborted'. */ + const decide = (params) => { + const options = Array.isArray(params?.options) ? params.options : []; + if (stopping) return Promise.resolve({ decision: { outcome: 'cancelled' }, by: 'aborted' }); + return new Promise((resolve) => { + const answer = (result) => { + if (!pendingPermissions.delete(answer)) return; + resolve(result); + }; + pendingPermissions.add(answer); + if (!hasHook) { answer({ decision: defaultPermissionDecision(options), by: 'default' }); return; } + Promise.resolve().then(() => ctx.decidePermission(params)).then((raw) => { + // An answer the adapter records decides only when it names who gave + // it; one naming nobody, or a pod member, is a failed ask and the + // policy default decides. A decider that records itself (the valet) + // has already held its own answer to that. + const decision = checkedDecision(raw, options); + if (!decision || (recordPermissions && !isAttributed(raw))) { + answer({ decision: defaultPermissionDecision(options), by: 'failed' }); + return; + } + answer({ decision, by: 'hook', decidedBy: raw?.decidedBy, decidedByRole: raw?.decidedByRole }); + }, () => answer({ decision: defaultPermissionDecision(options), by: 'failed' })); + }); + }; + + const onRequest = async (method, params) => { + if (method !== 'session/request_permission') return undefined; + const options = Array.isArray(params?.options) ? params.options : []; + const toolCall = params?.toolCall || {}; + const toolCallId = typeof toolCall.toolCallId === 'string' ? toolCall.toolCallId : null; + const argsDigest = argsDigestOf(toolCall.rawInput); + const shared = { ...(grantId ? { grantId } : {}), ...(argsDigest ? { argsDigest } : {}) }; + if (recordPermissions) { + emit('permission.requested', { + toolCallId, + options: options.map((option) => ({ optionId: option?.optionId, kind: option?.kind })), + question: questionOf(toolCall, credentials), + askedOf: hasHook && isAsker(ctx.askedOf) ? { id: ctx.askedOf.id, role: ctx.askedOf.role } : policy, + ...shared, + }); + } + const result = await decide(params); + const { decision } = result; + if (recordPermissions) { + const kind = options.find((option) => option?.optionId === decision.optionId)?.kind; + const asked = result.by !== 'default'; + const approval = { hook: kind?.startsWith('allow') ? 'approved' : 'rejected', failed: 'failed', aborted: 'aborted' }[result.by]; + emit('permission.decided', { + outcome: decision.outcome, + ...(decision.optionId ? { optionId: decision.optionId, kind } : {}), + toolCallId, + rule: { list: result.by === 'hook' ? 'ask' : 'default', pattern: null }, + ...(asked && hasHook ? { approval } : {}), + ...(result.by === 'hook' + ? { decidedBy: result.decidedBy, decidedByRole: result.decidedByRole } + : { decidedBy: 'policy', decidedByRole: 'policy' }), + ...shared, + }); + } + return { outcome: decision }; + }; + + const onNotification = (method, params) => { + if (method !== 'session/update' || !sessionId || params?.sessionId !== sessionId) return; + traceUpdate({ emit }, params.update, turn, { readRetry: readRetryStatus }); + }; + + const client = createAcpClient({ + input: child.stdout, output: child.stdin, onRequest, onNotification, maxLineChars, + }); + + child.stdin.on('error', () => { /* EPIPE after exit; the exit path reports it */ }); + child.stderr.on('data', (chunk) => { stderr = `${stderr}${chunk}`.slice(-MAX_STDERR_CHARS); }); + child.on('error', (error) => { + if (exitInfo) return; + client.close(runFailure('exited', `could not start ${name}: ${error?.code || error?.message}`, { credentials })); + if (child.pid === undefined) resolveGone(); + }); + child.once('exit', (code, signal) => { + exitInfo = { code, signal }; + emit('process.exited', { code, signal }); + resolveGone(); + const done = () => client.close(exitedFailure()); + if (child.stdout.readableEnded) { done(); return; } + const timer = setTimeout(done, EXIT_DRAIN_MS); + timer.unref?.(); + child.once('close', () => { clearTimeout(timer); done(); }); + }); + + let rejectStopped; + const stopped = new Promise((_, reject) => { rejectStopped = reject; }); + stopped.catch(() => {}); + + // Cancel, then kill. ACP requires every pending permission request to be + // answered cancelled once session/cancel is sent. + const stop = async (failure, { cancelGraceMs = timings.cancelGraceMs } = {}) => { + if (stopping || settled) return; + stopping = failure; + if (pendingPermissions.size) { + for (const answer of [...pendingPermissions]) answer(ABORTED_ANSWER); + // The answers are written a few microtasks later; let them reach the + // agent before the cancel does, so it never sees a cancel with its + // permission questions still open. + await new Promise((resolve) => { setImmediate(resolve); }); + } + if (promptInFlight && sessionId) { + client.notify('session/cancel', { sessionId }); + await Promise.race([promptInFlight.catch(() => {}), sleep(cancelGraceMs), gone]); + } + rejectStopped(failure); + }; + + const requestedTimeout = Number(ctx.timeoutMs); + const timeoutMs = Number.isFinite(requestedTimeout) && requestedTimeout > 0 ? requestedTimeout : DEFAULT_TIMEOUT_MS; + const timer = setTimeout(() => stop(runFailure('timeout', `${name} exceeded ${timeoutMs}ms`)), timeoutMs); + const onAbort = () => stop(abortFailure(ctx.signal)); + ctx.signal?.addEventListener('abort', onAbort, { once: true }); + if (wire) { + const onWire = (failure) => stop(failure, { cancelGraceMs: timings.wireCancelGraceMs }); + if (wire.failure) onWire(wire.failure); else wire.stop = onWire; + } + + const call = async (method, params) => { + try { + return await client.request(method, params); + } catch (error) { + if (error instanceof AcpRpcError) throw failureFromRpcError(error, credentials); + throw error; + } + }; + + const openSession = async (capabilities) => { + const base = { cwd, mcpServers: launched.mcpServers || [] }; + const wanted = ctx.sessionId; + let fallback = null; + if (wanted) { + const ways = []; + if (capabilities?.sessionCapabilities?.resume) ways.push('resume'); + if (capabilities?.loadSession) ways.push('load'); + fallback = 'agent supports neither session/resume nor session/load'; + for (const way of ways) { + try { + await client.request(`session/${way}`, { ...base, sessionId: wanted }); + emit('session.opened', { method: way, sessionId: wanted }); + return { id: wanted, fresh: false }; + } catch (error) { + if (!(error instanceof AcpRpcError)) throw error; + fallback = `session/${way} answered error ${error.code}`; + } + } + } + const created = await call('session/new', base); + if (typeof created?.sessionId !== 'string' || !created.sessionId) { + throw runFailure('protocol', 'session/new returned no sessionId'); + } + emit('session.opened', { + method: 'new', sessionId: created.sessionId, ...(wanted ? { requestedSessionId: wanted, fallback } : {}), + }); + return { id: created.sessionId, fresh: true }; + }; + + const drive = async () => { + const init = await call('initialize', { + protocolVersion: ACP_PROTOCOL_VERSION, + clientCapabilities: { fs: { readTextFile: false, writeTextFile: false }, terminal: false }, + clientInfo: { name: 'seats', version: '0.0.0' }, + }); + if (init?.protocolVersion !== ACP_PROTOCOL_VERSION) { + throw runFailure('unsupported', `${name} negotiated ACP protocol ${JSON.stringify(init?.protocolVersion ?? null)}; this adapter speaks ${ACP_PROTOCOL_VERSION}`); + } + const session = await openSession(init.agentCapabilities); + sessionId = session.id; + const text = memoryPreamble(prompt, ctx.memoryLongTerm, { freshSession: session.fresh }); + if (stopping) throw stopping; + promptInFlight = call('session/prompt', { sessionId, prompt: [{ type: 'text', text }] }); + emit('turn.prompt_sent', { chars: text.length }); + const response = await promptInFlight; + if (stopping) throw stopping; + if (response?.usage) emit('usage', usageFromPrompt(response.usage)); + const reason = response?.stopReason; + if (reason === 'end_turn') { + const replyFailure = readReplyFailure?.(turn.text, { credentials }); + if (replyFailure) throw replyFailure; + return { + text: turn.text.trim(), + newSessionId: sessionId, + ...(turn.contextTokens !== undefined ? { contextTokens: turn.contextTokens } : {}), + }; + } + if (STOP_FAILURES[reason]) throw runFailure(STOP_FAILURES[reason], `${name} stopped the turn: ${reason}`); + throw runFailure('protocol', `${name} answered session/prompt with unknown stopReason ${JSON.stringify(reason ?? null)}`); + }; + + const terminate = async () => { + if (!exitInfo && !stopping) { + child.stdin.end(); + await Promise.race([gone, sleep(timings.closeGraceMs)]); + } + const kill = (signal) => { if (!signalGroup(child, signal)) child.kill(signal); }; + if (!exitInfo) { + kill('SIGTERM'); + await Promise.race([gone, sleep(timings.killGraceMs)]); + } + if (!exitInfo) kill('SIGKILL'); + await gone; + // The leader is gone; anything it left in its group goes too, on every + // end, so no descendant outlives the run. + if (signalGroup(child, 0)) { + signalGroup(child, 'SIGTERM'); + const deadline = Date.now() + timings.killGraceMs; + while (signalGroup(child, 0) && Date.now() < deadline) await sleep(20); + signalGroup(child, 'SIGKILL'); + } + }; + + const outcome = drive().catch((error) => { throw stopping || error; }); + Promise.race([outcome, stopped]).then( + (result) => ({ result }), + (error) => ({ error: stopping || error }), + ).then(async (end) => { + settled = true; + clearTimeout(timer); + ctx.signal?.removeEventListener('abort', onAbort); + for (const answer of [...pendingPermissions]) answer(ABORTED_ANSWER); + await terminate(); + if (end.error) rejectRun(end.error); else resolveRun(end.result); + }); +}); + +/** + * @param {import('../contract.js').SeatHost & { + * name: string, + * launch: (ctx: object) => Promise | AcpLaunch, + * detect: () => Promise<{ path: string, version: string } | null>, + * runtimeType?: string, + * providerKeyEnv?: string, + * readReplyFailure?: (text: string, opts: { credentials: string[] }) => RunFailure | null, + * readRetryStatus?: (update: object) => { state: 'scheduled'|'started', attempt?, nextRetryAt?, status?, errorType? } | null, + * maxLineChars?: number, + * timings?: { cancelGraceMs?: number, closeGraceMs?: number, killGraceMs?: number, wireCancelGraceMs?: number }, + * watchProvider?: boolean, + * }} options + * + * @typedef {object} AcpLaunch + * @property {string} command + * @property {string[]} [args] + * @property {Record} env the whole child env; nothing is inherited + * @property {string} [cwd] + * @property {object[]} [mcpServers] ACP McpServer entries for session/new|load|resume + * @property {() => Promise | void} [cleanup] + * + * `watchProvider` (default true) relays ctx.provider.baseURL through + * provider-watch.js when the ctx carries one: a 4xx ends the run as + * refusal_retry_loop, a 5xx/408/429 or an unreachable provider as + * provider_unavailable, at the first one. + * + * `readReplyFailure` exists because some agents end a failed turn with + * end_turn and the error as the reply text; the launch knows that shape, the + * protocol does not. `readRetryStatus` is the same idea for retries: an agent + * that reports its own provider retries does it in a session_info_update + * `_meta` key of its own, and the reader turns that into an agent.retry. + */ +export const createAcpAdapter = ({ + name, + launch, + detect, + runtimeType = name, + providerKeyEnv, + readReplyFailure, + readRetryStatus, + maxLineChars = DEFAULT_MAX_LINE_CHARS, + timings = {}, + memoryPreamble = (prompt) => prompt, + effectiveTrust = (trust) => trust, + seatModelRequired = () => false, + watchProvider = true, +} = {}) => { + if (!name || typeof launch !== 'function' || typeof detect !== 'function') { + throw new Error('createAcpAdapter needs a name, a launch function and a detect function'); + } + const resolvedTimings = { + cancelGraceMs: timings.cancelGraceMs ?? CANCEL_GRACE_MS, + closeGraceMs: timings.closeGraceMs ?? CLOSE_GRACE_MS, + killGraceMs: timings.killGraceMs ?? KILL_GRACE_MS, + wireCancelGraceMs: timings.wireCancelGraceMs ?? WIRE_CANCEL_GRACE_MS, + }; + return { + get requiresModel() { return seatModelRequired(); }, + name, + runtimeType, + ...(providerKeyEnv ? { providerKeyEnv } : {}), + detect, + + async validateEnvironment(environment) { + // The agent's own permission system plus decidePermission is all the + // confinement there is; nothing here can enforce a sandbox. + if (effectiveTrust(environment?.sandbox?.trust) === 'public' + || (environment?.sandbox?.mode && environment.sandbox.mode !== 'none')) { + throw runFailure('unsupported', `${name}: public or sandbox-declared seats are unsupported; no enforced confinement exists for ACP agents`); + } + }, + + async spawn(prompt, ctx = {}) { + await this.validateEnvironment(ctx.environment); + if (ctx.signal?.aborted) throw abortFailure(ctx.signal); + // The provider route runs through a relay that ends the run on the + // first failing answer; see provider-watch.js for why the wire is the + // only place that answer is visible. + const wire = { failure: null, stop: null }; + const watch = watchProvider && ctx.provider?.baseURL ? await startProviderWatch({ + baseURL: ctx.provider.baseURL, + credentials: acpCredentials(ctx), + onFailure: (failure) => { + wire.failure = failure; + wire.stop?.(failure); + }, + }) : null; + let launched; + try { + launched = await launch(watch ? { ...ctx, provider: { ...ctx.provider, baseURL: watch.baseURL } } : ctx); + return await runTurn({ + name, launched, prompt, ctx, memoryPreamble, readReplyFailure, readRetryStatus, maxLineChars, timings: resolvedTimings, wire, + }); + } finally { + try { await launched?.cleanup?.(); } catch { /* cleanup is best effort; the turn's outcome stands */ } + await watch?.close(); + } + }, + }; +}; diff --git a/packages/seats/src/failures.js b/packages/seats/src/failures.js index b3ff12fb1d..6c5a5ca148 100644 --- a/packages/seats/src/failures.js +++ b/packages/seats/src/failures.js @@ -22,6 +22,8 @@ export const FAILURE_KINDS = Object.freeze({ quota: { retryable: false, meaning: 'the account behind the CLI is out of quota or credits' }, provider_unsupported: { retryable: false, meaning: 'the provider does not serve this model to this account' }, capacity: { retryable: true, meaning: 'every entry the run config allows is at a host cap or held; it may run once one frees' }, + refusal_retry_loop: { retryable: false, meaning: 'the provider rejected a request (4xx) the CLI would retry; stopped at the first one on the wire' }, + provider_unavailable: { retryable: true, meaning: 'the provider answered 5xx, 408 or 429, or was unreachable; stopped at the first one on the wire' }, }); export class RunFailure extends Error { diff --git a/packages/seats/src/index.js b/packages/seats/src/index.js index 47e30b6def..6aca2cbc88 100644 --- a/packages/seats/src/index.js +++ b/packages/seats/src/index.js @@ -9,6 +9,15 @@ export { export { adapterFailure, spawnCredentials } from './upstream-refusal.js'; export { default as stub } from './adapters/stub.js'; export { createHermesAdapter } from './adapters/hermes.js'; +export { + createAcpAdapter, ACP_PROTOCOL_VERSION, defaultPermissionDecision, readsAsRefusal, +} from './adapters/acp.js'; +export { + acpGoose, acpOpencode, createAcpGooseAdapter, createAcpOpencodeAdapter, launchGoose, launchOpencode, + readOpencodeRetry, +} from './adapters/acp-launch.js'; +export { createAcpClient, AcpRpcError } from './acp/client.js'; +export { startProviderWatch, failureForStatus } from './provider-watch.js'; export { RunFailure, runFailure, failureKindOf, FAILURE_KINDS } from './failures.js'; export { EVENT_TYPES, isEventType, traceOf, NULL_TRACE } from './trace/events.js'; export { createTraceStore } from './trace/store.js'; diff --git a/packages/seats/src/provider-watch.js b/packages/seats/src/provider-watch.js new file mode 100644 index 0000000000..913d113b76 --- /dev/null +++ b/packages/seats/src/provider-watch.js @@ -0,0 +1,131 @@ +/** + * A loopback relay between a CLI and its model provider, so the adapter sees + * what the CLI hides: the provider's own answer. + * + * Measured 2026-10-25 against the fake provider: goose 1.54.0 answers a 400 + * content refusal by retrying it eight times over ~8 s, and opencode 1.19.0 + * retries a 503 with backoff until the run's timeout. Neither says anything + * on its ACP stream or stderr while it does. The provider wire is the only + * place the condition is visible, so that is where it is read. + * + * Rule: the FIRST failing answer to a POST ends the run. A 4xx is + * `refusal_retry_loop`; a 5xx, a 408 or a 429 is `provider_unavailable`, + * because those say "not now" rather than "not this". No threshold: a + * refusal does not become an answer on the eighth try, and a run that waits + * out a CLI's backoff is the silent hang this exists to remove. GETs (model + * listings) are relayed but never judged. + * + * Once tripped, the relay answers everything locally with 503 and forwards + * nothing, so the provider sees no further calls while the CLI is stopped. + * Headers and bodies pass through untouched; the relay holds no key. + */ +import { createServer, request as httpRequest } from 'node:http'; +import { request as httpsRequest } from 'node:https'; +import { runFailure } from './failures.js'; + +const BODY_SAMPLE_CHARS = 4096; + +/** The failure a provider status names, or null when the answer is fine. */ +export const failureForStatus = (status, body, { credentials = [] } = {}) => { + // The status leads the text so the shared keep-list reads `NNN: {body}` + // and keeps at most the body's one named message field. + const text = `${status}: ${String(body ?? '').slice(0, BODY_SAMPLE_CHARS)}`; + const transient = status === 408 || status === 429 || (status >= 500 && status <= 599); + if (transient) { + return runFailure('provider_unavailable', `provider answered ${text}`, { credentials, limit: 500 }); + } + if (status >= 400 && status <= 499) { + return runFailure('refusal_retry_loop', `provider answered ${text}`, { credentials, limit: 500 }); + } + return null; +}; + +/** + * @param {object} opts + * @param {string} opts.baseURL the real provider base URL, e.g. https://host/v1 + * @param {(failure: import('./failures.js').RunFailure) => void} opts.onFailure called once + * @param {string[]} [opts.credentials] scrubbed from the failure detail + * @returns {Promise<{ baseURL: string, close: () => Promise, tripped: boolean }>} + */ +export const startProviderWatch = async ({ baseURL, onFailure, credentials = [] }) => { + const upstream = new URL(baseURL); + const basePath = upstream.pathname.replace(/\/$/, ''); + const send = upstream.protocol === 'https:' ? httpsRequest : httpRequest; + const sockets = new Set(); + const forwards = new Set(); + let tripped = null; + + const trip = (failure) => { + if (tripped) return; + tripped = failure; + onFailure(failure); + }; + + const server = createServer((req, res) => { + if (tripped) { + req.resume(); + res.writeHead(503, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ error: { message: 'stopped by the seat after a provider failure' } })); + return; + } + const judged = req.method === 'POST'; + const headers = { ...req.headers, host: upstream.host }; + const forward = send({ + protocol: upstream.protocol, + hostname: upstream.hostname, + port: upstream.port || undefined, + method: req.method, + // The relay's base URL keeps the provider's path, so the CLI's request + // path is already the upstream path. + path: req.url, + headers, + }, (answer) => { + answer.on('aborted', () => res.destroy()); + const status = answer.statusCode || 0; + if (!judged || status < 400) { + res.writeHead(status, answer.headers); + answer.pipe(res); + return; + } + // Read the error body for the failure detail, then pass it on whole so + // the CLI's own handling still sees a well-formed answer. + const chunks = []; + answer.on('data', (chunk) => { chunks.push(chunk); }); + answer.on('end', () => { + const body = Buffer.concat(chunks); + const failure = failureForStatus(status, body.toString('utf8'), { credentials }); + if (failure) trip(failure); + res.writeHead(status, answer.headers); + res.end(body); + }); + answer.on('error', () => res.destroy()); + }); + forwards.add(forward); + forward.on('close', () => forwards.delete(forward)); + // A CLI that hangs up (or is killed) takes its upstream request with it, + // so the relay never holds a provider connection nobody is waiting on. + res.on('close', () => { if (!res.writableFinished) forward.destroy(); }); + forward.on('error', (error) => { + if (res.destroyed) return; + // An unreachable provider is a provider that is unavailable. + if (judged) trip(runFailure('provider_unavailable', `provider unreachable: ${error?.code || error?.message}`, { credentials })); + res.destroy(); + }); + req.pipe(forward); + }); + server.on('connection', (socket) => { + sockets.add(socket); + socket.on('close', () => sockets.delete(socket)); + }); + await new Promise((resolve) => { server.listen(0, '127.0.0.1', resolve); }); + const { port } = server.address(); + return { + baseURL: `http://127.0.0.1:${port}${basePath}`, + get tripped() { return tripped; }, + close: () => new Promise((resolve) => { + for (const forward of forwards) forward.destroy(); + for (const socket of sockets) socket.destroy(); + server.close(() => resolve()); + }), + }; +}; diff --git a/packages/seats/src/trace/events.js b/packages/seats/src/trace/events.js index 5304d4f671..45a8cf94d1 100644 --- a/packages/seats/src/trace/events.js +++ b/packages/seats/src/trace/events.js @@ -11,6 +11,7 @@ export const EVENT_TYPES = Object.freeze([ 'run.started', // { adapter, cli: {name, version}?, sessionId?, attempt } 'run.settled', // { outcome: 'ok'|'failed'|'stuck'|'cancelled', failure?: {kind, detail}, durationMs, attempts } + 'session.opened', // { method: 'new'|'resume'|'load', sessionId, requestedSessionId?, fallback? } fallback names why a requested session was not reopened 'turn.prompt_sent', // { chars } 'agent.message_chunk', // { text } 'agent.thought_chunk', // { chars } thoughts are counted, not stored @@ -20,7 +21,7 @@ export const EVENT_TYPES = Object.freeze([ 'plan', // { entries: [{content, status}] } 'permission.requested', // { toolCallId?, options: [{optionId, kind}], question, askedOf: {id, role}, grantId?, decisionCardId?, argsDigest? } 'permission.decided', // { outcome: 'selected'|'cancelled', optionId?, kind?, toolCallId?, rule?: {list, pattern}, approval? } - 'usage', // { inputTokens?, outputTokens?, totalTokens?, cost? } null-not-zero when unknown + 'usage', // { inputTokens?, outputTokens?, totalTokens?, cost?, contextTokens?, contextSize? } null-not-zero when unknown 'liveness.stalled', // { idleMs, windowMs } 'liveness.recovery', // { action: 'cancel'|'kill'|'retry', attempt } 'credential.granted', // { names: [...], injected?: [...] } names only, never values diff --git a/packages/seats/test-support/fake-acp-agent.mjs b/packages/seats/test-support/fake-acp-agent.mjs new file mode 100644 index 0000000000..55901a189d --- /dev/null +++ b/packages/seats/test-support/fake-acp-agent.mjs @@ -0,0 +1,256 @@ +#!/usr/bin/env node +/** + * A scripted ACP agent over stdio, for the ACP adapter's unit tests. + * + * Process-level behaviour comes from env: + * FAKE_ACP_LOG append every message received, one JSON line each + * FAKE_ACP_VERSION protocolVersion to answer initialize with (default 1) + * FAKE_ACP_CAPS comma list of `resume`, `load` + * FAKE_ACP_FAIL comma list of `resume`, `load` to answer with an error + * FAKE_ACP_FAIL_CODE that error's code as JSON (default -32002); `absent` omits it + * + * Turn behaviour comes from a directive in the prompt text: + * (none) thought, tool call + update, plan, usage, "PONG" + * [[reply:TEXT]] reply TEXT + * [[stop:REASON]] end with stopReason REASON + * [[permission]] ask permission (allow_once + reject_once), reply PERM: + * [[permission-noreject]] same, offering allow options only + * [[fs]] call fs/read_text_file and terminal/create, reply the error codes + * [[malformed]] write a non-JSON line, then hang + * [[huge:N]] write one N-char line, then hang + * [[exit]] write OPENAI_API_KEY to stderr and exit 3 + * [[hang]] never answer until session/cancel, then stopReason cancelled + * [[hang-deaf]] never answer, ignore cancel and SIGTERM + * [[hang-permission]] ask permission and wait; on cancel answer cancelled + * [[rpc-error:CODE:MSG]] answer the prompt with a JSON-RPC error + * [[echo-session]] reply SESSION: + * [[bad-usage]] a usage_update with a string `used` and a negative `size`, then OK + * [[opencode-retry]] opencode 2.x retry reports: two session_info_update + * _meta["opencode/retry"] reports, each followed by a + * null one, whose message and response body carry + * OPENAI_API_KEY; two malformed reports and a null; a + * report with no error; a title-only + * session_info_update; then OK + * [[grandchild:THEN]] start a grandchild that ignores SIGTERM, log its pid + * ({event:'grandchild', pid}), then THEN: `hang` + * (answers cancel), `reply` (end_turn) or `exit` (exit 0) + * [[provider:N]] POST the prompt to FAKE_ACP_PROVIDER_URL/chat/completions + * up to N times (default 8), 200 ms apart, like a CLI + * retrying a failing provider; ignores cancel; then + * reply PROVIDER: (PROVIDER: on a 200) + */ +import { spawn } from 'node:child_process'; +import { appendFileSync } from 'node:fs'; + +const log = (entry) => { if (process.env.FAKE_ACP_LOG) appendFileSync(process.env.FAKE_ACP_LOG, `${JSON.stringify(entry)}\n`); }; +const write = (message) => process.stdout.write(`${JSON.stringify({ jsonrpc: '2.0', ...message })}\n`); +const caps = new Set(String(process.env.FAKE_ACP_CAPS || '').split(',').filter(Boolean)); +const fails = new Set(String(process.env.FAKE_ACP_FAIL || '').split(',').filter(Boolean)); +const failError = () => { + const raw = process.env.FAKE_ACP_FAIL_CODE; + if (raw === 'absent') return { message: 'Resource not found' }; + return { code: raw === undefined ? -32002 : JSON.parse(raw), message: 'Resource not found' }; +}; + +let nextId = 1; +const waiting = new Map(); +const ask = (method, params) => new Promise((resolve) => { + const id = `agent-${nextId}`; + nextId += 1; + waiting.set(id, resolve); + write({ id, method, params }); +}); +const update = (sessionId, body) => write({ method: 'session/update', params: { sessionId, update: body } }); +const chunk = (sessionId, text) => update(sessionId, { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text } }); + +let activePrompt = null; +let deaf = false; + +const PERMISSION_OPTIONS = [ + { optionId: 'allow', name: 'Allow', kind: 'allow_once' }, + { optionId: 'reject', name: 'Reject', kind: 'reject_once' }, +]; +const ALLOW_ONLY = [ + { optionId: 'allow', name: 'Allow', kind: 'allow_once' }, + { optionId: 'always', name: 'Always', kind: 'allow_always' }, +]; + +const permission = (sessionId, options) => ask('session/request_permission', { + sessionId, toolCall: { toolCallId: 'tc-1', title: process.env.FAKE_ACP_PERMISSION_TITLE || 'Edit file', kind: 'edit' }, options, +}); + +const runPrompt = async (id, { sessionId, prompt }) => { + const text = (prompt || []).map((block) => block?.text || '').join(''); + const directive = /\[\[([a-z-]+)(?::([^\]]*))?\]\]/.exec(text); + const [, kind, arg = ''] = directive || []; + const answer = (stopReason, extra = {}) => write({ id, result: { stopReason, ...extra } }); + switch (kind) { + case 'reply': chunk(sessionId, arg); return answer('end_turn'); + case 'stop': return answer(arg); + case 'bad-usage': + update(sessionId, { sessionUpdate: 'usage_update', used: '42', size: -1 }); + update(sessionId, { sessionUpdate: 'usage_update', used: 1.5, size: 1000 }); + chunk(sessionId, 'OK'); + return answer('end_turn'); + case 'opencode-retry': { + const key = process.env.OPENAI_API_KEY || ''; + const info = (meta) => update(sessionId, { sessionUpdate: 'session_info_update', _meta: meta }); + const report = (attempt, type, status) => info({ + 'opencode/retry': { + attempt, + nextRetryAt: `2026-10-25T13:00:0${attempt}.000Z`, + error: { type, message: `upstream said ${key}`, status, response: { body: `{"error":"raw provider text ${key}"}` } }, + }, + }); + report(2, 'provider.internal', 503); + info({ 'opencode/retry': null }); + report(3, 'provider.rate-limit', 429); + info({ 'opencode/retry': null }); + info({ 'opencode/retry': { attempt: '4', nextRetryAt: 'soon', error: { type: key, status: '503' } } }); + info({ 'opencode/retry': { attempt: 0, nextRetryAt: 12, error: { type: 'provider.made-up', status: 99 } } }); + info({ 'opencode/retry': null }); + info({ 'opencode/retry': { attempt: 5, nextRetryAt: '2026-10-25T13:00:05Z' } }); + update(sessionId, { sessionUpdate: 'session_info_update', title: 'a title' }); + chunk(sessionId, 'OK'); + return answer('end_turn'); + } + case 'echo-session': chunk(sessionId, `SESSION:${sessionId}`); return answer('end_turn'); + case 'permission': + case 'permission-noreject': { + const response = await permission(sessionId, kind === 'permission' ? PERMISSION_OPTIONS : ALLOW_ONLY); + chunk(sessionId, `PERM:${JSON.stringify(response.result?.outcome ?? response.error)}`); + return answer('end_turn'); + } + case 'fs': { + const read = await ask('fs/read_text_file', { sessionId, path: '/etc/hosts' }); + const term = await ask('terminal/create', { sessionId, command: 'ls' }); + chunk(sessionId, `FS:${read.error?.code} TERM:${term.error?.code}`); + return answer('end_turn'); + } + case 'malformed': process.stdout.write('this is not json\n'); return undefined; + case 'huge': process.stdout.write(`${'x'.repeat(Number(arg) || 1000)}\n`); return undefined; + case 'exit': + process.stderr.write(`fatal: provider rejected key ${process.env.OPENAI_API_KEY}\n`); + setTimeout(() => process.exit(3), 10); + return undefined; + case 'hang': activePrompt = { id, sessionId }; return undefined; + case 'hang-deaf': deaf = true; process.on('SIGTERM', () => log({ event: 'sigterm-ignored' })); return undefined; + case 'hang-permission': { + activePrompt = { id, sessionId }; + // The answer is logged as it arrives (handle logs every message); a log + // written after this await can lose the race with the run's teardown. + await permission(sessionId, PERMISSION_OPTIONS); + return undefined; + } + case 'grandchild': { + const grandchild = spawn(process.execPath, ['-e', + "process.on('SIGTERM', () => {}); process.stdout.write('ready\\n'); setInterval(() => {}, 1000);"], + { stdio: ['ignore', 'pipe', 'ignore'] }); + await new Promise((resolve) => { grandchild.stdout.once('data', resolve); }); + log({ event: 'grandchild', pid: grandchild.pid }); + if (arg === 'reply') { chunk(sessionId, 'SPAWNED'); return answer('end_turn'); } + if (arg === 'exit') { setTimeout(() => process.exit(0), 10); return undefined; } + activePrompt = { id, sessionId }; + return undefined; + } + case 'provider': { + const statuses = []; + for (let attempt = 0; attempt < (Number(arg) || 8); attempt += 1) { + let status = 0; + let body = null; + try { + const answer = await fetch(`${process.env.FAKE_ACP_PROVIDER_URL}/chat/completions`, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${process.env.OPENAI_API_KEY}` }, + body: JSON.stringify({ model: 'm', messages: [{ role: 'user', content: text }] }), + }); + status = answer.status; + body = await answer.json().catch(() => null); + } catch { status = -1; } + log({ event: 'provider-attempt', status }); + statuses.push(status); + if (status === 200) { + chunk(sessionId, `PROVIDER:${body?.choices?.[0]?.message?.content}`); + return answer('end_turn'); + } + await new Promise((resolve) => { setTimeout(resolve, 200); }); + } + chunk(sessionId, `PROVIDER:${statuses.join(',')}`); + return answer('end_turn'); + } + case 'rpc-error': { + const [code, ...message] = arg.split(':'); + return write({ id, error: { code: Number(code), message: message.join(':') } }); + } + default: + update(sessionId, { sessionUpdate: 'agent_thought_chunk', content: { type: 'text', text: 'thinking hard' } }); + update(sessionId, { sessionUpdate: 'tool_call', toolCallId: 'tc-1', title: 'Read file', kind: 'read', status: 'pending' }); + update(sessionId, { sessionUpdate: 'tool_call_update', toolCallId: 'tc-1', status: 'completed' }); + update(sessionId, { sessionUpdate: 'plan', entries: [{ content: 'answer', priority: 'high', status: 'completed' }] }); + update(sessionId, { sessionUpdate: 'usage_update', used: 42, size: 1000 }); + update(sessionId, { sessionUpdate: 'a_kind_from_the_future', anything: true }); + write({ method: 'x/unknown_notification', params: {} }); + update('some-other-session', { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'WRONG' } }); + chunk(sessionId, 'PO'); + chunk(sessionId, 'NG'); + return answer('end_turn', { usage: { inputTokens: 10, outputTokens: 2, totalTokens: 12 } }); + } +}; + +const handle = (message) => { + log(message); + if (!message.method) { + const resolve = waiting.get(message.id); + waiting.delete(message.id); + resolve?.(message); + return; + } + const { id, method, params } = message; + switch (method) { + case 'initialize': + return write({ + id, + result: { + protocolVersion: Number(process.env.FAKE_ACP_VERSION || 1), + agentCapabilities: { + loadSession: caps.has('load'), + sessionCapabilities: caps.has('resume') ? { resume: {} } : {}, + }, + authMethods: [], + }, + }); + case 'session/new': return write({ id, result: { sessionId: 'sess-new-1' } }); + case 'session/resume': + if (fails.has('resume')) return write({ id, error: failError() }); + return write({ id, result: {} }); + case 'session/load': + if (fails.has('load')) return write({ id, error: failError() }); + chunk(params.sessionId, 'REPLAYED'); + return write({ id, result: {} }); + case 'session/prompt': return runPrompt(id, params); + case 'session/cancel': + if (deaf) return undefined; + if (activePrompt) { + write({ id: activePrompt.id, result: { stopReason: 'cancelled' } }); + activePrompt = null; + } + return undefined; + default: + if (id !== undefined) write({ id, error: { code: -32601, message: `unknown ${method}` } }); + return undefined; + } +}; + +let buffer = ''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data', (data) => { + buffer += data; + let newline = buffer.indexOf('\n'); + while (newline !== -1) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + if (line.trim()) handle(JSON.parse(line)); + newline = buffer.indexOf('\n'); + } +}); +process.stdin.on('end', () => { if (!deaf) process.exit(0); }); diff --git a/packages/seats/test-support/record-trace.js b/packages/seats/test-support/record-trace.js new file mode 100644 index 0000000000..5c13dad101 --- /dev/null +++ b/packages/seats/test-support/record-trace.js @@ -0,0 +1,29 @@ +/** + * A ctx.trace backed by a real record-v1 store, so adapter tests run under the + * strict writer: an entry the record would refuse throws here exactly as it + * would in production. `events` keeps what the adapter emitted, in order. + */ +import { randomUUID } from 'node:crypto'; +import { createTraceStore } from '../src/trace/store.js'; + +export const recordTrace = ({ root, secrets = [], actor = { type: 'agent', id: 'agent-test', owner: 'user-owner' } }) => { + const store = createTraceStore({ root }); + const runId = `run-${randomUUID().slice(0, 12)}`; + const run = store.startRun({ runId, adapter: 'acp-test', actor }); + run.redact(secrets); + const events = []; + return { + events, + run, + store, + lines: () => store.readRun(runId), + verify: () => store.verifyRun(runId), + trace: { + emit(type, data) { + const entry = run.emit(type, data); + events.push({ type, data }); + return entry; + }, + }, + }; +}; diff --git a/packages/seats/test/acp-real.test.js b/packages/seats/test/acp-real.test.js new file mode 100644 index 0000000000..d4b06a6548 --- /dev/null +++ b/packages/seats/test/acp-real.test.js @@ -0,0 +1,267 @@ +/** + * Real-binary ACP turns against the local fake provider. Each test runs only + * when its binary exists; nothing here reaches a hosted model or uses a real + * key. Override the paths with SEATS_GOOSE_BIN / SEATS_OPENCODE_BIN. + */ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { existsSync, mkdirSync, mkdtempSync, rmSync } from 'node:fs'; +import { createServer } from 'node:http'; +import { homedir, tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { + acpGoose, acpOpencode, createAcpGooseAdapter, createAcpOpencodeAdapter, +} from '../src/adapters/acp-launch.js'; +import { RunFailure } from '../src/failures.js'; +import { startFakeProvider } from '../test-support/fake-provider.js'; +import { recordTrace } from '../test-support/record-trace.js'; +import { after } from 'node:test'; + +const GOOSE = process.env.SEATS_GOOSE_BIN || join(homedir(), '.commonly', 'goose-1.54.0', 'goose'); +const OPENCODE = process.env.SEATS_OPENCODE_BIN + || join(homedir(), '.commonly', 'opencode-1.19.0', 'node_modules', 'opencode-darwin-arm64', 'bin', 'opencode'); +const KEY = 'sk-fake-not-a-real-key-0001'; + +/** + * A pass-through in front of the fake provider that remembers how many + * messages each chat request carried, which is how a resumed turn shows it + * brought the earlier turn along. + */ +const startRecorder = async (upstream) => { + const bodies = []; + const server = createServer((req, res) => { + let raw = ''; + req.on('data', (chunk) => { raw += chunk; }); + req.on('end', async () => { + try { bodies.push(JSON.parse(raw || '{}')); } catch { bodies.push({}); } + try { + const answer = await fetch(`${upstream.baseURL}${req.url.replace(/^\/v1/, '')}`, { + method: req.method, + headers: { + 'content-type': 'application/json', + ...(req.headers.authorization ? { authorization: req.headers.authorization } : {}), + }, + body: req.method === 'POST' ? raw : undefined, + }); + res.writeHead(answer.status, { 'content-type': answer.headers.get('content-type') || 'application/json' }); + res.end(Buffer.from(await answer.arrayBuffer())); + } catch { + res.destroy(); // a held [[stall]] request ends when the fake provider closes + } + }); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + return { + baseURL: `http://127.0.0.1:${server.address().port}/v1`, + bodies, + close: () => new Promise((resolve) => server.close(() => resolve())), + }; +}; + +const withProvider = async (fn) => { + const fake = await startFakeProvider(); + const recorder = await startRecorder(fake); + const root = mkdtempSync(join(tmpdir(), 'seats-acp-real-')); + const home = join(root, 'home'); + const cwd = join(root, 'ws'); + mkdirSync(home); + mkdirSync(cwd); + try { + await fn({ fake, recorder, home, cwd }); + } finally { + await recorder.close(); + await fake.close(); + rmSync(root, { recursive: true, force: true }); + } +}; + +// Every run writes a real record-v1 store, so a real agent's stream meets the +// strict writer: an entry the record would refuse fails the test. +const STORE_ROOT = mkdtempSync(join(tmpdir(), 'seats-acp-real-store-')); +const records = []; + +const runCtx = ({ binary, recorder, home, cwd, events, ...extra }) => { + const record = recordTrace({ root: STORE_ROOT, secrets: [KEY] }); + records.push(record); + return { + _binaryPath: binary, + cwd, + env: { PATH: '/usr/bin:/bin', HOME: home }, + provider: { baseURL: recorder.baseURL, apiKey: KEY, model: 'fake-model' }, + trace: { emit: (type, data) => { record.trace.emit(type, data); events.push({ type, data }); } }, + timeoutMs: 120000, + ...extra, + }; +}; + +after(() => { + try { + for (const record of records) assert.equal(record.verify()?.ok, true, 'every real run left a verifiable record'); + } finally { rmSync(STORE_ROOT, { recursive: true, force: true }); } +}); + +const chatRequests = (recorder) => recorder.bodies.filter((body) => Array.isArray(body.messages)); +const userMessages = (body) => body.messages.filter((m) => m.role === 'user').length; + +test('goose acp: one real turn against the fake provider', { skip: !existsSync(GOOSE) && `no goose at ${GOOSE}` }, async () => { + await withProvider(async ({ fake, recorder, home, cwd }) => { + const events = []; + const result = await acpGoose.spawn('Reply with exactly PONG.', runCtx({ binary: GOOSE, recorder, home, cwd, events })); + assert.equal(result.text, 'PONG'); + assert.ok(result.newSessionId); + const types = events.map((e) => e.type); + for (const type of ['session.opened', 'turn.prompt_sent', 'agent.message_chunk', 'usage', 'process.exited']) { + assert.ok(types.includes(type), `trace has ${type}: ${types}`); + } + assert.equal(types.at(-1), 'process.exited'); + assert.ok(fake.requests.length >= 1, 'the fake provider saw the turn'); + assert.ok(fake.requests.every((r) => r.model === 'fake-model' && r.authorization === 'present')); + + // goose keeps sessions under HOME, so a second spawn can reopen it. + const again = []; + const second = await acpGoose.spawn('Reply with exactly PONG.', runCtx({ + binary: GOOSE, recorder, home, cwd, events: again, sessionId: result.newSessionId, + })); + assert.equal(second.text, 'PONG', 'the history goose replays on load is not part of the reply'); + assert.deepEqual(again[0].data, { method: 'load', sessionId: result.newSessionId }); + assert.equal(userMessages(chatRequests(recorder).at(-1)), 2, 'the loaded turn carried the first turn'); + }); +}); + +test('opencode acp: one real turn, then a resumed second turn', { skip: !existsSync(OPENCODE) && `no opencode at ${OPENCODE}` }, async () => { + await withProvider(async ({ fake, recorder, home, cwd }) => { + const events = []; + const result = await acpOpencode.spawn('Reply with exactly PONG.', runCtx({ binary: OPENCODE, recorder, home, cwd, events })); + assert.equal(result.text, 'PONG'); + assert.match(result.newSessionId, /^ses_/); + const types = events.map((e) => e.type); + for (const type of ['session.opened', 'turn.prompt_sent', 'agent.message_chunk', 'usage', 'process.exited']) { + assert.ok(types.includes(type), `trace has ${type}: ${types}`); + } + assert.ok(fake.requests.length >= 1, 'the fake provider saw the turn'); + assert.ok(fake.requests.every((r) => r.model === 'fake-model' && r.authorization === 'present'), + 'every request went to the fake model, none to the hosted default'); + + const again = []; + const second = await acpOpencode.spawn('Reply with exactly PONG again.', runCtx({ + binary: OPENCODE, recorder, home, cwd, events: again, sessionId: result.newSessionId, + })); + assert.equal(second.text, 'PONG'); + assert.equal(second.newSessionId, result.newSessionId); + assert.deepEqual(again[0].data, { method: 'resume', sessionId: result.newSessionId }); + const turn = chatRequests(recorder).filter((body) => body.messages.some((m) => JSON.stringify(m.content).includes('PONG again'))); + assert.ok(turn.length >= 1); + assert.equal(userMessages(turn.at(-1)), 2, 'the resumed turn carried the first turn'); + }); +}); + +/** + * Measured 2026-10-25 with no watch: goose 1.54.0 sends a refused request 8 + * times over ~8.6 s and a 503 8 times over ~10.8 s before it says anything; + * opencode 1.19.0 retries a 503 with backoff until the run's own timeout. The + * provider watch ends each at the first failing answer. Time is measured from + * the provider's first request, so CLI start-up on a loaded host stays out of + * the "fast" claim. + */ +const wireFailure = async ({ adapter, binary, prompt }) => { + let outcome; + await withProvider(async ({ fake, recorder, home, cwd }) => { + const events = []; + let firstAt = null; + const poll = setInterval(() => { if (firstAt === null && fake.requests.length) firstAt = Date.now(); }, 5); + const error = await adapter.spawn(prompt, runCtx({ binary, recorder, home, cwd, events })) + .then(() => null, (failure) => failure); + clearInterval(poll); + outcome = { + error, events, afterFirstMs: Date.now() - (firstAt ?? Date.now()), requests: fake.requests.map((r) => r.directive), + }; + }); + return outcome; +}; + +const assertStoppedAtFirst = ({ error, events, afterFirstMs, requests }, kind, directive) => { + assert.ok(error instanceof RunFailure, `expected a RunFailure, got ${error}`); + assert.equal(error.kind, kind, error.message); + assert.ok(!error.message.includes(KEY)); + // Up to two failing requests: both CLIs open the turn with two concurrent + // calls, so the second can already be on the wire when the first answers. + const failing = requests.filter((r) => r === directive).length; + assert.ok(failing >= 1 && failing <= 2, `the provider saw ${failing} ${directive} requests`); + assert.ok(afterFirstMs < 5000, `settled ${afterFirstMs}ms after the first provider request`); + assert.equal(events.at(-1).type, 'process.exited'); +}; + +test('goose acp: the first provider refusal ends the run as refusal_retry_loop', { skip: !existsSync(GOOSE) && `no goose at ${GOOSE}` }, async () => { + const outcome = await wireFailure({ adapter: acpGoose, binary: GOOSE, prompt: '[[refuse]]' }); + assertStoppedAtFirst(outcome, 'refusal_retry_loop', 'refuse'); + assert.match(outcome.error.message, /content policy/); +}); + +test('goose acp: the first provider 503 ends the run as provider_unavailable', { skip: !existsSync(GOOSE) && `no goose at ${GOOSE}` }, async () => { + assertStoppedAtFirst(await wireFailure({ adapter: acpGoose, binary: GOOSE, prompt: '[[fail:503]]' }), 'provider_unavailable', 'fail'); +}); + +test('opencode acp: the first provider refusal ends the run as refusal_retry_loop', { skip: !existsSync(OPENCODE) && `no opencode at ${OPENCODE}` }, async () => { + assertStoppedAtFirst(await wireFailure({ adapter: acpOpencode, binary: OPENCODE, prompt: '[[refuse]]' }), 'refusal_retry_loop', 'refuse'); +}); + +test('opencode acp: the first provider 503 ends the run as provider_unavailable, not at a timeout', { skip: !existsSync(OPENCODE) && `no opencode at ${OPENCODE}` }, async () => { + assertStoppedAtFirst(await wireFailure({ adapter: acpOpencode, binary: OPENCODE, prompt: '[[fail:503]]' }), 'provider_unavailable', 'fail'); +}); + +// With the watch off, what each CLI itself reports is still read correctly. +test('goose acp, watch off: the refusal goose ends its turn with is a refusal, not a reply', { skip: !existsSync(GOOSE) && `no goose at ${GOOSE}` }, async () => { + await withProvider(async ({ recorder, home, cwd }) => { + const error = await createAcpGooseAdapter({ watchProvider: false }) + .spawn('[[refuse]]', runCtx({ binary: GOOSE, recorder, home, cwd, events: [] })) + .then(() => null, (failure) => failure); + assert.ok(error instanceof RunFailure); + assert.equal(error.kind, 'refusal'); + assert.match(error.message, /content policy/); + }); +}); + +test('opencode acp, watch off: its -32603 content-policy error is a refusal', { skip: !existsSync(OPENCODE) && `no opencode at ${OPENCODE}` }, async () => { + await withProvider(async ({ recorder, home, cwd }) => { + const error = await createAcpOpencodeAdapter({ watchProvider: false }) + .spawn('[[refuse]]', runCtx({ binary: OPENCODE, recorder, home, cwd, events: [] })) + .then(() => null, (failure) => failure); + assert.ok(error instanceof RunFailure); + assert.equal(error.kind, 'refusal'); + }); +}); + +// Measured 2026-10-25: opencode 1.19.0 answers session/cancel with +// stopReason cancelled at once; goose 1.54.0 does not answer it while its +// provider request is held, so the cancel grace runs out and it is killed. +const abortAfterStall = async ({ adapter, binary }) => { + let outcome; + await withProvider(async ({ fake, recorder, home, cwd }) => { + const events = []; + const controller = new AbortController(); + const run = adapter.spawn('[[stall]]', runCtx({ binary, recorder, home, cwd, events, signal: controller.signal })); + const deadline = Date.now() + 60000; + while (!fake.requests.some((r) => r.directive === 'stall') && Date.now() < deadline) { + await new Promise((resolve) => { setTimeout(resolve, 50); }); + } + const abortedAt = Date.now(); + controller.abort(); + const error = await run.then(() => null, (failure) => failure); + outcome = { error, events, settleMs: Date.now() - abortedAt }; + }); + return outcome; +}; + +test('opencode acp: abort mid-turn is answered by the agent within the cancel grace', { skip: !existsSync(OPENCODE) && `no opencode at ${OPENCODE}` }, async () => { + const { error, events, settleMs } = await abortAfterStall({ adapter: acpOpencode, binary: OPENCODE }); + assert.equal(error?.kind, 'cancelled'); + // Ignoring the cancel would cost the full 3000 ms grace before the kill even starts. + assert.ok(settleMs < 3000, `settled ${settleMs}ms after abort, inside the 3000ms cancel grace`); + assert.equal(events.at(-1).type, 'process.exited'); +}); + +test('goose acp: abort mid-turn still ends the run when the agent never answers the cancel', { skip: !existsSync(GOOSE) && `no goose at ${GOOSE}` }, async () => { + const { error, events } = await abortAfterStall({ adapter: acpGoose, binary: GOOSE }); + assert.equal(error?.kind, 'cancelled'); + assert.deepEqual(events.at(-1), { type: 'process.exited', data: { code: null, signal: 'SIGTERM' } }); +}); diff --git a/packages/seats/test/acp.test.js b/packages/seats/test/acp.test.js new file mode 100644 index 0000000000..7943a08501 --- /dev/null +++ b/packages/seats/test/acp.test.js @@ -0,0 +1,794 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { existsSync, mkdtempSync, readFileSync, rmSync, statSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { PassThrough } from 'node:stream'; +import { createAcpAdapter, defaultPermissionDecision, readsAsRefusal } from '../src/adapters/acp.js'; +import { readOpencodeRetry } from '../src/adapters/acp-launch.js'; +import { createAcpClient } from '../src/acp/client.js'; +import { + launchGoose, launchOpencode, readGooseReplyFailure, acpGoose, acpOpencode, +} from '../src/adapters/acp-launch.js'; +import { RunFailure, runFailure } from '../src/failures.js'; +import { recordTrace } from '../test-support/record-trace.js'; +import { createPermissionPolicy } from '../src/valet/permissions.js'; +import { startFakeProvider } from '../test-support/fake-provider.js'; + +const AGENT = fileURLToPath(new URL('../test-support/fake-acp-agent.mjs', import.meta.url)); +const KEY = ['sk', 'test', '0123456789abcdef'].join('-'); +// A hook answer the adapter records names who gave it (record-v1: a decision always does). +const BY_BEN = Object.freeze({ decidedBy: 'user-ben', decidedByRole: 'credential-owner' }); +const FAST = { cancelGraceMs: 300, closeGraceMs: 300, killGraceMs: 300 }; + +const scratch = () => mkdtempSync(join(tmpdir(), 'seats-acp-')); + +/** An adapter over the fake agent, plus what a test inspects afterwards. */ +const harness = ({ env = {}, ...options } = {}) => { + const dir = scratch(); + const logPath = join(dir, 'agent.log'); + const state = { launches: 0, cleanups: 0 }; + const adapter = createAcpAdapter({ + name: 'fake-acp', + detect: async () => null, + timings: FAST, + launch: (ctx) => { + state.launches += 1; + return { + command: process.execPath, + args: [AGENT], + env: { + PATH: process.env.PATH, OPENAI_API_KEY: ctx.provider?.apiKey || '', FAKE_ACP_LOG: logPath, + FAKE_ACP_PROVIDER_URL: ctx.provider?.baseURL || '', ...env, + }, + cwd: dir, + cleanup: () => { state.cleanups += 1; }, + }; + }, + ...options, + }); + // Every emit goes through a real record-v1 store: the strict writer. + const record = recordTrace({ root: join(dir, 'store'), secrets: [KEY] }); + const { events, trace } = record; + const received = () => (existsSync(logPath) + ? readFileSync(logPath, 'utf8').split('\n').filter(Boolean).map((line) => JSON.parse(line)) : []); + const methods = () => received().map((entry) => entry.method || entry.event || (entry.result !== undefined ? 'response' : 'error')); + const ctx = (extra = {}) => ({ trace, provider: { baseURL: 'http://127.0.0.1:9/v1', apiKey: KEY, model: 'm' }, ...extra }); + const done = () => rmSync(dir, { recursive: true, force: true }); + return { adapter, events, received, methods, ctx, state, done, record }; +}; + +const types = (events) => events.map((event) => event.type); +const rejectsKind = async (promise, kind) => { + const error = await promise.then(() => null, (failure) => failure); + assert.ok(error instanceof RunFailure, `expected a RunFailure, got ${error}`); + assert.equal(error.kind, kind, error.message); + return error; +}; + +test('happy path: initialize, new session, one prompt; the trace follows the update stream', async () => { + const h = harness(); + try { + const seen = []; + const adapter = createAcpAdapter({ + name: 'fake-acp', detect: async () => null, timings: FAST, + launch: () => ({ command: process.execPath, args: [AGENT], env: { PATH: process.env.PATH } }), + memoryPreamble: (prompt, longTerm, opts) => { seen.push(opts); return `[memory]\n${prompt}`; }, + }); + const result = await h.adapter.spawn('hello', h.ctx()); + assert.deepEqual(result, { text: 'PONG', newSessionId: 'sess-new-1', contextTokens: 42 }); + assert.deepEqual(types(h.events), [ + 'session.opened', 'turn.prompt_sent', 'agent.thought_chunk', 'tool.call', 'tool.update', 'plan', + 'usage', 'agent.message_chunk', 'agent.message_chunk', 'usage', 'process.exited', + ]); + const [opened, sent, thought, call, toolUpdate, plan, context, , , usage, exited] = h.events.map((e) => e.data); + assert.deepEqual(opened, { method: 'new', sessionId: 'sess-new-1' }); + assert.deepEqual(sent, { chars: 5 }); + assert.deepEqual(thought, { chars: 13 }, 'thoughts are counted, never stored'); + assert.deepEqual(call, { toolCallId: 'tc-1', title: 'Read file', kind: 'read', status: 'pending' }); + assert.deepEqual(toolUpdate, { toolCallId: 'tc-1', status: 'completed', title: undefined }); + assert.deepEqual(plan, { entries: [{ content: 'answer', status: 'completed' }] }); + assert.deepEqual(context, { inputTokens: null, outputTokens: null, totalTokens: null, cost: null, contextTokens: 42, contextSize: 1000 }); + assert.deepEqual(usage, { inputTokens: 10, outputTokens: 2, totalTokens: 12, cost: null }); + assert.deepEqual(exited, { code: 0, signal: null }, 'a finished agent exits on stdin EOF, not by signal'); + assert.equal(h.state.cleanups, 1); + + const init = h.received().find((m) => m.method === 'initialize'); + assert.equal(init.params.protocolVersion, 1); + assert.deepEqual(init.params.clientCapabilities, { fs: { readTextFile: false, writeTextFile: false }, terminal: false }); + const promptSent = h.received().find((m) => m.method === 'session/prompt'); + assert.deepEqual(promptSent.params, { sessionId: 'sess-new-1', prompt: [{ type: 'text', text: 'hello' }] }); + + const preambleRun = await adapter.spawn('hi', { trace: { emit() {} } }); + assert.equal(preambleRun.text, 'PONG'); + assert.deepEqual(seen, [{ freshSession: true }]); + } finally { h.done(); } +}); + +test('a requested session prefers session/resume over session/load', async () => { + const h = harness({ env: { FAKE_ACP_CAPS: 'resume,load' } }); + try { + const result = await h.adapter.spawn('[[echo-session]]', h.ctx({ sessionId: 'sess-old' })); + assert.deepEqual(result, { text: 'SESSION:sess-old', newSessionId: 'sess-old' }); + const sessionCalls = h.received().map((m) => m.method).filter((m) => m?.startsWith('session/') && m !== 'session/prompt'); + assert.deepEqual(sessionCalls, ['session/resume']); + assert.deepEqual(h.events[0].data, { method: 'resume', sessionId: 'sess-old' }); + } finally { h.done(); } +}); + +test('session/load is used without resume, and its replayed history is not the reply', async () => { + const h = harness({ env: { FAKE_ACP_CAPS: 'load' } }); + try { + const result = await h.adapter.spawn('[[reply:fresh]]', h.ctx({ sessionId: 'sess-old' })); + assert.equal(result.text, 'fresh'); + assert.ok(!h.events.some((e) => e.type === 'agent.message_chunk' && e.data.text === 'REPLAYED')); + assert.deepEqual(h.events[0].data, { method: 'load', sessionId: 'sess-old' }); + } finally { h.done(); } +}); + +test('a session that cannot be reopened falls back to session/new and the trace says why', async () => { + const none = harness(); + const failing = harness({ env: { FAKE_ACP_CAPS: 'resume,load', FAKE_ACP_FAIL: 'resume,load' } }); + try { + const result = await none.adapter.spawn('[[echo-session]]', none.ctx({ sessionId: 'sess-old' })); + assert.equal(result.newSessionId, 'sess-new-1'); + assert.deepEqual(none.events[0].data, { + method: 'new', sessionId: 'sess-new-1', requestedSessionId: 'sess-old', + fallback: 'agent supports neither session/resume nor session/load', + }); + await failing.adapter.spawn('[[echo-session]]', failing.ctx({ sessionId: 'sess-old' })); + const sessionCalls = failing.received().map((m) => m.method).filter((m) => /^session\/(resume|load|new)$/.test(m || '')); + assert.deepEqual(sessionCalls, ['session/resume', 'session/load', 'session/new']); + assert.equal(failing.events[0].data.fallback, 'session/load answered error -32002'); + } finally { none.done(); failing.done(); } +}); + +test('the memory preamble is told whether the session is fresh', async () => { + const seen = []; + const h = harness({ env: { FAKE_ACP_CAPS: 'resume' }, memoryPreamble: (prompt, longTerm, opts) => { seen.push([longTerm, opts.freshSession]); return `${longTerm}|${prompt}`; } }); + try { + await h.adapter.spawn('[[reply:a]]', h.ctx({ sessionId: 'sess-old', memoryLongTerm: 'MEM' })); + await h.adapter.spawn('[[reply:b]]', h.ctx({ memoryLongTerm: 'MEM' })); + assert.deepEqual(seen, [['MEM', false], ['MEM', true]]); + const prompts = h.received().filter((m) => m.method === 'session/prompt').map((m) => m.params.prompt[0].text); + assert.deepEqual(prompts, ['MEM|[[reply:a]]', 'MEM|[[reply:b]]']); + } finally { h.done(); } +}); + +test('any protocol version other than 1 is unsupported', async () => { + const h = harness({ env: { FAKE_ACP_VERSION: '2' } }); + try { + const error = await rejectsKind(h.adapter.spawn('hi', h.ctx()), 'unsupported'); + assert.match(error.message, /protocol 2/); + assert.ok(!h.received().some((m) => m.method === 'session/new'), 'nothing past initialize'); + assert.equal(h.state.cleanups, 1); + } finally { h.done(); } +}); + +test('stop reasons map to failure kinds', async () => { + const h = harness(); + try { + for (const reason of ['refusal', 'max_tokens', 'max_turn_requests', 'cancelled']) { + await rejectsKind(h.adapter.spawn(`[[stop:${reason}]]`, h.ctx()), reason); + } + await rejectsKind(h.adapter.spawn('[[stop:made_up]]', h.ctx()), 'protocol'); + } finally { h.done(); } +}); + +test('a JSON-RPC error on the prompt is protocol, unless it reads as an upstream refusal', async () => { + const h = harness(); + try { + // The shape opencode 1.19.0 answers with for a 400 content-policy body. + const refusal = await rejectsKind(h.adapter.spawn('[[rpc-error:-32603:Internal error: The request was rejected by the content policy.]]', h.ctx()), 'refusal'); + assert.match(refusal.message, /content policy/); + await rejectsKind(h.adapter.spawn('[[rpc-error:-32603:Internal error: disk on fire]]', h.ctx()), 'protocol'); + const echoed = await rejectsKind(h.adapter.spawn(`[[rpc-error:-32603:bad key ${KEY}]]`, h.ctx()), 'protocol'); + assert.doesNotMatch(echoed.message, new RegExp(KEY)); + } finally { h.done(); } +}); + +test('permission without a hook: reject_once if offered, otherwise cancelled', async () => { + const h = harness(); + try { + const rejected = await h.adapter.spawn('[[permission]]', h.ctx()); + assert.equal(rejected.text, 'PERM:{"outcome":"selected","optionId":"reject"}'); + const requested = h.events.find((e) => e.type === 'permission.requested').data; + assert.deepEqual(requested, { + toolCallId: 'tc-1', + options: [{ optionId: 'allow', kind: 'allow_once' }, { optionId: 'reject', kind: 'reject_once' }], + question: 'edit: Edit file', + askedOf: { id: 'policy', role: 'policy' }, + }); + assert.deepEqual(h.events.find((e) => e.type === 'permission.decided').data, { + outcome: 'selected', optionId: 'reject', kind: 'reject_once', toolCallId: 'tc-1', + rule: { list: 'default', pattern: null }, decidedBy: 'policy', decidedByRole: 'policy', + }); + + const cancelled = await h.adapter.spawn('[[permission-noreject]]', h.ctx()); + assert.equal(cancelled.text, 'PERM:{"outcome":"cancelled"}'); + assert.deepEqual(defaultPermissionDecision([]), { outcome: 'cancelled' }); + } finally { h.done(); } +}); + +test('permission with a hook: its answer is used only when it names an offered option and who gave it', async () => { + const h = harness(); + try { + const asked = []; + const allow = await h.adapter.spawn('[[permission]]', h.ctx({ + decidePermission: async (request) => { asked.push(request); return { outcome: 'selected', optionId: 'allow', ...BY_BEN }; }, + })); + assert.equal(allow.text, 'PERM:{"outcome":"selected","optionId":"allow"}'); + assert.equal(asked[0].sessionId, 'sess-new-1'); + assert.deepEqual(asked[0].toolCall, { toolCallId: 'tc-1', title: 'Edit file', kind: 'edit' }); + assert.equal(asked[0].options.length, 2); + + const bogus = await h.adapter.spawn('[[permission]]', h.ctx({ decidePermission: async () => ({ outcome: 'selected', optionId: 'root', ...BY_BEN }) })); + assert.equal(bogus.text, 'PERM:{"outcome":"selected","optionId":"reject"}'); + const thrown = await h.adapter.spawn('[[permission]]', h.ctx({ decidePermission: () => { throw new Error('hook broke'); } })); + assert.equal(thrown.text, 'PERM:{"outcome":"selected","optionId":"reject"}'); + const declined = await h.adapter.spawn('[[permission]]', h.ctx({ decidePermission: async () => ({ outcome: 'cancelled', ...BY_BEN }) })); + assert.equal(declined.text, 'PERM:{"outcome":"cancelled"}'); + // An answer naming nobody approves nothing. + const nobody = await h.adapter.spawn('[[permission]]', h.ctx({ decidePermission: async () => ({ outcome: 'selected', optionId: 'allow' }) })); + assert.equal(nobody.text, 'PERM:{"outcome":"selected","optionId":"reject"}'); + // So does one whose decidedBy is not a user id, whatever its role. + for (const decidedBy of ['user ben', '', '-ben']) { + const notAnId = await h.adapter.spawn('[[permission]]', h.ctx({ decidePermission: async () => ({ outcome: 'selected', optionId: 'allow', decidedBy, decidedByRole: 'credential-owner' }) })); + assert.equal(notAnId.text, 'PERM:{"outcome":"selected","optionId":"reject"}', JSON.stringify(decidedBy)); + } + // A hook that fails is recorded as the policy deciding, with approval failed. + const decided = h.events.filter((e) => e.type === 'permission.decided').map((e) => e.data); + assert.deepEqual(decided.map((d) => [d.rule.list, d.approval, d.decidedBy, d.decidedByRole]), [ + ['ask', 'approved', 'user-ben', 'credential-owner'], + ['default', 'failed', 'policy', 'policy'], + ['default', 'failed', 'policy', 'policy'], + ['ask', 'rejected', 'user-ben', 'credential-owner'], + ['default', 'failed', 'policy', 'policy'], + ['default', 'failed', 'policy', 'policy'], + ['default', 'failed', 'policy', 'policy'], + ['default', 'failed', 'policy', 'policy'], + ]); + assert.equal(h.record.verify().ok, true); + } finally { h.done(); } +}); + +test('permission records name the approver, the decider and the grant (record-v1)', async () => { + const h = harness(); + try { + const approver = { id: 'user-ben', role: 'credential-owner' }; + await h.adapter.spawn('[[permission]]', h.ctx({ + askedOf: approver, + grantId: 'g-42', + decidePermission: async () => ({ + outcome: 'selected', optionId: 'allow', decidedBy: 'user-ben', decidedByRole: 'credential-owner', + }), + })); + const requested = h.events.find((e) => e.type === 'permission.requested').data; + assert.deepEqual(requested.askedOf, approver); + assert.equal(requested.grantId, 'g-42'); + assert.equal(requested.question, 'edit: Edit file'); + assert.deepEqual(h.events.find((e) => e.type === 'permission.decided').data, { + outcome: 'selected', optionId: 'allow', kind: 'allow_once', toolCallId: 'tc-1', + rule: { list: 'ask', pattern: null }, approval: 'approved', + decidedBy: 'user-ben', decidedByRole: 'credential-owner', grantId: 'g-42', + }); + // The written record: the decision's actor is the human who decided. + const lines = h.record.lines(); + assert.equal(lines.find((l) => l.kind === 'permission.decided').actor.id, 'user-ben'); + assert.equal(h.record.verify().ok, true); + } finally { h.done(); } +}); + +test('an approver that is not an id token, or a decider that names nobody valid, is never recorded as a person', async () => { + const h = harness(); + try { + const result = await h.adapter.spawn('[[permission]]', h.ctx({ + askedOf: { id: 'Ben Smith', role: 'credential-owner' }, + grantId: 'g-7', + decidePermission: async () => ({ outcome: 'selected', optionId: 'allow', decidedBy: 'Ben Smith', decidedByRole: 'boss' }), + })); + assert.equal(result.text, 'PERM:{"outcome":"selected","optionId":"reject"}'); + assert.deepEqual(h.events.find((e) => e.type === 'permission.requested').data.askedOf, { id: 'grant:g-7', role: 'policy' }); + const decided = h.events.find((e) => e.type === 'permission.decided').data; + assert.deepEqual([decided.decidedBy, decided.decidedByRole, decided.approval, decided.rule.list], ['policy', 'policy', 'failed', 'default']); + } finally { h.done(); } +}); + +test('a pod member answering a parked call decides nothing: the policy default does', async () => { + const h = harness(); + try { + // askedOf may never be a pod member either. + const result = await h.adapter.spawn('[[permission]]', h.ctx({ + askedOf: { id: 'user-pat', role: 'pod-member' }, + decidePermission: async () => ({ outcome: 'selected', optionId: 'allow', decidedBy: 'user-pat', decidedByRole: 'pod-member' }), + })); + assert.equal(result.text, 'PERM:{"outcome":"selected","optionId":"reject"}'); + assert.deepEqual(h.events.find((e) => e.type === 'permission.requested').data.askedOf, { id: 'policy', role: 'policy' }); + const decided = h.events.find((e) => e.type === 'permission.decided').data; + assert.deepEqual([decided.decidedBy, decided.decidedByRole, decided.approval, decided.rule.list, decided.optionId], ['policy', 'policy', 'failed', 'default', 'reject']); + // Nor may a hook claim the policy's own role. + await h.adapter.spawn('[[permission]]', h.ctx({ decidePermission: async () => ({ outcome: 'selected', optionId: 'allow', decidedBy: 'user-pat', decidedByRole: 'policy' }) })); + assert.equal(h.events.filter((e) => e.type === 'permission.decided')[1].data.approval, 'failed'); + assert.equal(h.record.verify().ok, true); + } finally { h.done(); } +}); + +test('the question is agent text: credentials are scrubbed out of it and it stays short', async () => { + const h = harness({ env: { FAKE_ACP_PERMISSION_TITLE: `Upload with key ${KEY} ${'x'.repeat(400)}` } }); + try { + await h.adapter.spawn('[[permission]]', h.ctx()); + const { question } = h.events.find((e) => e.type === 'permission.requested').data; + assert.ok(!question.includes(KEY), question); + assert.match(question, /^edit: Upload with key \[redacted\]/); + assert.ok(question.length <= 200); + } finally { h.done(); } +}); + +test('the valet decider writes one permission pair and the adapter none; a plain hook gets the adapter pair', async () => { + const h = harness(); + try { + const owner = { id: 'user-ben', role: 'credential-owner' }; + const valet = createPermissionPolicy({ tools: { ask: ['edit'] } }, { + trace: h.record.trace, askedOf: owner, grantId: 'g-9', + approve: async () => ({ approved: true, by: 'user-ben', role: 'credential-owner' }), + }); + assert.equal(valet.recordsPermissions, true); + assert.ok(!Object.keys(valet).includes('recordsPermissions'), 'the marker is not enumerable'); + const allowed = await h.adapter.spawn('[[permission]]', h.ctx({ decidePermission: valet, askedOf: owner, grantId: 'g-9' })); + assert.equal(allowed.text, 'PERM:{"outcome":"selected","optionId":"allow"}'); + const pair = h.record.lines().filter((line) => line.kind.startsWith('permission.')); + assert.deepEqual(pair.map((line) => line.kind), ['permission.requested', 'permission.decided']); + assert.equal(pair[1].data.decidedBy, 'user-ben'); + assert.equal(pair[1].actor.id, 'user-ben'); + + await h.adapter.spawn('[[permission]]', h.ctx({ decidePermission: async () => ({ outcome: 'selected', optionId: 'allow', ...BY_BEN }) })); + const all = h.record.lines().filter((line) => line.kind.startsWith('permission.')); + assert.deepEqual(all.map((line) => line.kind), ['permission.requested', 'permission.decided', 'permission.requested', 'permission.decided']); + assert.equal(h.record.verify().ok, true); + } finally { h.done(); } +}); + +test('fs and terminal requests are refused with -32601', async () => { + const h = harness(); + try { + const result = await h.adapter.spawn('[[fs]]', h.ctx()); + assert.equal(result.text, 'FS:-32601 TERM:-32601'); + } finally { h.done(); } +}); + +test('a malformed line or an oversized line is a protocol failure', async () => { + const h = harness({ maxLineChars: 1000 }); + try { + const malformed = await rejectsKind(h.adapter.spawn('[[malformed]]', h.ctx()), 'protocol'); + assert.match(malformed.message, /not JSON/); + const huge = await rejectsKind(h.adapter.spawn('[[huge:5000]]', h.ctx()), 'protocol'); + assert.match(huge.message, /exceeded 1000 chars/); + assert.equal(h.events.filter((e) => e.type === 'process.exited').length, 2, 'the child is killed either way'); + } finally { h.done(); } +}); + +test('an agent that exits mid-turn is exited, with credentials scrubbed from the stderr tail', async () => { + const h = harness(); + try { + const error = await rejectsKind(h.adapter.spawn('[[exit]]', h.ctx()), 'exited'); + assert.match(error.message, /code 3/); + assert.match(error.message, /fatal: provider rejected key \[redacted\]/); + assert.doesNotMatch(error.message, new RegExp(KEY)); + assert.deepEqual(h.events.at(-1), { type: 'process.exited', data: { code: 3, signal: null } }); + assert.equal(h.state.cleanups, 1); + } finally { h.done(); } +}); + +test('a command that cannot start is exited, and cleanup still runs', async () => { + let cleaned = 0; + const adapter = createAcpAdapter({ + name: 'missing', detect: async () => null, timings: FAST, + launch: () => ({ command: '/nonexistent/acp-agent', args: [], env: {}, cleanup: () => { cleaned += 1; } }), + }); + await rejectsKind(adapter.spawn('hi', {}), 'exited'); + assert.equal(cleaned, 1); +}); + +test('abort mid-turn sends session/cancel, waits for the cancelled response, and rejects cancelled', async () => { + const h = harness(); + try { + const controller = new AbortController(); + const run = h.adapter.spawn('[[hang]]', h.ctx({ signal: controller.signal })); + await waitFor(() => h.events.some((e) => e.type === 'turn.prompt_sent')); + controller.abort(); + await rejectsKind(run, 'cancelled'); + assert.ok(h.methods().includes('session/cancel')); + assert.equal(h.state.cleanups, 1); + assert.equal(h.events.at(-1).type, 'process.exited'); + } finally { h.done(); } +}); + +test('abort answers a pending permission request cancelled before the turn ends', async () => { + const h = harness(); + try { + const controller = new AbortController(); + const run = h.adapter.spawn('[[hang-permission]]', h.ctx({ signal: controller.signal, decidePermission: () => new Promise(() => {}) })); + await waitFor(() => h.events.some((e) => e.type === 'permission.requested')); + controller.abort(); + await rejectsKind(run, 'cancelled'); + // What the agent read, logged as it read it. Anything the agent does after + // answering the cancel can be cut off: the run ends on that answer and its + // process group is killed, so a loaded host may never run the agent again. + const answer = h.received().find((entry) => entry.id === 'agent-1' && !entry.method); + assert.deepEqual(answer?.result, { outcome: { outcome: 'cancelled' } }, JSON.stringify(h.received())); + const order = h.methods(); + assert.ok(order.indexOf('response') < order.indexOf('session/cancel'), `permission answered before cancel: ${order}`); + assert.deepEqual(h.events.find((e) => e.type === 'permission.decided').data, { + outcome: 'cancelled', toolCallId: 'tc-1', rule: { list: 'default', pattern: null }, + approval: 'aborted', decidedBy: 'policy', decidedByRole: 'policy', + }); + } finally { h.done(); } +}); + +test('an abort whose reason is a RunFailure rejects with that failure', async () => { + const h = harness(); + try { + const controller = new AbortController(); + const run = h.adapter.spawn('[[hang]]', h.ctx({ signal: controller.signal })); + await waitFor(() => h.events.some((e) => e.type === 'turn.prompt_sent')); + const reason = runFailure('stuck', 'no progress'); + controller.abort(reason); + assert.equal(await run.catch((error) => error), reason); + } finally { h.done(); } +}); + +test('an agent deaf to cancel and SIGTERM is SIGKILLed after the grace periods', async () => { + const h = harness(); + try { + const controller = new AbortController(); + const run = h.adapter.spawn('[[hang-deaf]]', h.ctx({ signal: controller.signal })); + await waitFor(() => h.events.some((e) => e.type === 'turn.prompt_sent')); + const started = Date.now(); + controller.abort(); + await rejectsKind(run, 'cancelled'); + assert.ok(Date.now() - started >= FAST.cancelGraceMs + FAST.killGraceMs - 50); + assert.deepEqual(h.events.at(-1).data, { code: null, signal: 'SIGKILL' }); + } finally { h.done(); } +}); + +test('an already-aborted signal never launches', async () => { + const h = harness(); + try { + const controller = new AbortController(); + controller.abort(); + await rejectsKind(h.adapter.spawn('hi', h.ctx({ signal: controller.signal })), 'cancelled'); + assert.equal(h.state.launches, 0); + } finally { h.done(); } +}); + +test('timeoutMs takes the cancel path and rejects timeout', async () => { + // The clock starts at spawn, so on a loaded host a short budget can run out + // before the prompt is even sent; widen it until the prompt is in flight + // when the timeout fires, which is the case under test. + for (const timeoutMs of [300, 2000, 8000, 20000]) { + const h = harness(); + try { + const error = await rejectsKind(h.adapter.spawn('[[hang]]', h.ctx({ timeoutMs })), 'timeout'); + assert.match(error.message, new RegExp(`exceeded ${timeoutMs}ms`)); + if (!types(h.events).includes('turn.prompt_sent')) continue; + assert.ok(h.methods().includes('session/cancel')); + return; + } finally { h.done(); } + } + assert.fail('the prompt was never in flight when the timeout fired'); +}); + +test('public or sandbox-declared seats are refused before launch', async () => { + const h = harness(); + try { + await rejectsKind(h.adapter.spawn('hi', h.ctx({ environment: { sandbox: { trust: 'public' } } })), 'unsupported'); + await rejectsKind(h.adapter.spawn('hi', h.ctx({ environment: { sandbox: { mode: 'bwrap' } } })), 'unsupported'); + assert.equal(h.state.launches, 0); + } finally { h.done(); } +}); + +test('client: an answer to an id never sent is a protocol failure that rejects pending requests', async () => { + const input = new PassThrough(); + const output = new PassThrough(); + const failures = []; + const client = createAcpClient({ input, output, onProtocolError: (failure) => failures.push(failure) }); + const pending = client.request('initialize', {}); + input.write('{"jsonrpc":"2.0","id":999,"result":{}}\n'); + const error = await pending.catch((failure) => failure); + assert.equal(error.kind, 'protocol'); + assert.equal(failures.length, 1); + assert.match(output.read().toString(), /"method":"initialize"/); +}); + +test('refusal reading covers the measured agent shapes', () => { + assert.ok(readsAsRefusal('Internal error: The request was rejected by the content policy.')); + assert.ok(readsAsRefusal('Request failed: Bad request (400): nope')); + assert.ok(readsAsRefusal('Rate limit exceeded: fake provider failure.')); + assert.ok(!readsAsRefusal('Server error: Server error (500 Internal Server Error) at http://x')); + assert.ok(!readsAsRefusal('Internal error: disk on fire')); +}); + +test('goose reply failures: the measured "Ran into this error" texts become named failures', () => { + const refusal = readGooseReplyFailure('Ran into this error: Request failed: Bad request (400): The request was rejected by the content policy..\n\nPlease retry if you think this is a transient or recoverable error.'); + assert.equal(refusal.kind, 'refusal'); + assert.doesNotMatch(refusal.message, /Please retry/); + assert.equal(readGooseReplyFailure('Ran into this error: Rate limit exceeded: fake provider failure.').kind, 'refusal'); + assert.equal(readGooseReplyFailure('Ran into this error: Server error: Server error (500 Internal Server Error) at http://127.0.0.1:1/v1').kind, 'protocol'); + assert.equal(readGooseReplyFailure(`Ran into this error: key ${KEY} bad`, { credentials: [KEY] }).message.includes(KEY), false); + assert.equal(readGooseReplyFailure('PONG'), null); + assert.equal(readGooseReplyFailure('I ran into this error yesterday'), null); +}); + +test('launchGoose: provider in env, never argv; caller HOME kept; throwaway HOME removed', async () => { + const home = scratch(); + try { + const provider = { baseURL: 'http://127.0.0.1:4321/v1', apiKey: KEY, model: 'fake-model' }; + const launched = await launchGoose({ provider, cwd: home, env: { PATH: '/usr/bin', HOME: home, OPENROUTER_API_KEY: ['sk', 'or', 'must', 'not', 'pass', '123'].join('-') } }); + assert.deepEqual([launched.command, launched.args], ['goose', ['acp']]); + assert.equal(launched.env.HOME, home); + assert.equal(launched.env.OPENAI_HOST, 'http://127.0.0.1:4321'); + assert.equal(launched.env.OPENAI_BASE_PATH, 'v1/chat/completions'); + assert.equal(launched.env.OPENAI_API_KEY, KEY); + assert.equal(launched.env.GOOSE_PROVIDER, 'openai'); + assert.equal(launched.env.GOOSE_MODEL, 'fake-model'); + assert.equal(launched.env.OPENROUTER_API_KEY, undefined, 'only safe names cross from ctx.env'); + assert.ok(!launched.args.join(' ').includes(KEY)); + await launched.cleanup(); + assert.ok(existsSync(home), 'a caller HOME is never removed'); + + const throwaway = await launchGoose({ provider }); + assert.notEqual(throwaway.env.HOME, process.env.HOME); + assert.ok(existsSync(throwaway.env.HOME)); + await throwaway.cleanup(); + assert.ok(!existsSync(throwaway.env.HOME)); + await assert.rejects(launchGoose({ provider: { ...provider, baseURL: 'http://u:p@host/v1' } }), (e) => e.kind === 'unsupported'); + await assert.rejects(launchGoose({}), (e) => e.kind === 'unsupported'); + } finally { rmSync(home, { recursive: true, force: true }); } +}); + +test('launchOpencode: the key lives in a 0600 file the config names, never in config, env or argv', async () => { + const home = scratch(); + try { + const launched = await launchOpencode({ provider: { baseURL: 'http://127.0.0.1:4321/v1', apiKey: KEY, model: 'fake-model' }, env: { HOME: home, PATH: '/usr/bin' } }); + assert.deepEqual([launched.command, launched.args], ['opencode', ['acp']]); + const configText = readFileSync(launched.env.OPENCODE_CONFIG, 'utf8'); + const config = JSON.parse(configText); + assert.equal(config.model, 'seats/fake-model'); + assert.deepEqual(config.enabled_providers, ['seats']); + const { options } = config.provider.seats; + assert.equal(options.baseURL, 'http://127.0.0.1:4321/v1'); + const keyFile = /^\{file:(.+)\}$/.exec(options.apiKey)[1]; + assert.equal(readFileSync(keyFile, 'utf8'), KEY); + assert.equal(statSync(keyFile).mode & 0o777, 0o600); + assert.ok(!configText.includes(KEY)); + assert.ok(!Object.values(launched.env).some((value) => String(value).includes(KEY))); + assert.equal(launched.env.OPENCODE_DISABLE_PROJECT_CONFIG, '1'); + assert.equal(launched.env.XDG_DATA_HOME, join(home, '.local', 'share'), 'sessions live under the caller HOME'); + assert.notEqual(launched.env.XDG_CONFIG_HOME, join(home, '.config'), 'global config is isolated'); + await launched.cleanup(); + assert.ok(!existsSync(keyFile)); + assert.ok(existsSync(home)); + } finally { rmSync(home, { recursive: true, force: true }); } +}); + +test('the shipped adapters satisfy the Adapter contract', () => { + for (const adapter of [acpGoose, acpOpencode]) { + assert.equal(typeof adapter.name, 'string'); + assert.equal(typeof adapter.detect, 'function'); + assert.equal(typeof adapter.spawn, 'function'); + } + assert.deepEqual([acpGoose.name, acpOpencode.name], ['acp-goose', 'acp-opencode']); +}); + +/** + * How long a run took to settle after the provider first saw a request. Timing + * from that moment, not from spawn, keeps process start-up on a loaded host + * out of a "fires fast" claim. + */ +const settleAfterFirstRequest = async (provider, settling) => { + let firstAt = null; + const poll = setInterval(() => { if (firstAt === null && provider.requests.length) firstAt = Date.now(); }, 5); + try { + const error = await settling; + return { error, afterFirstMs: Date.now() - (firstAt ?? Date.now()) }; + } finally { clearInterval(poll); } +}; + +const alive = (pid) => { try { process.kill(pid, 0); return true; } catch { return false; } }; + +/** Runs a turn whose agent leaves a SIGTERM-deaf grandchild; it must not survive. */ +const grandchildOutlives = async (prompt, ctxExtra, settle) => { + const h = harness(); + let pid; + try { + await settle(h.adapter.spawn(prompt, h.ctx(ctxExtra(h))), h); + pid = h.received().find((entry) => entry.event === 'grandchild')?.pid; + assert.ok(pid, 'the agent started a grandchild'); + // A killed orphan is reaped by init asynchronously; give it a moment. + const deadline = Date.now() + 2000; + while (alive(pid) && Date.now() < deadline) await new Promise((r) => { setTimeout(r, 20); }); + return alive(pid); + } finally { + if (pid && alive(pid)) process.kill(pid, 'SIGKILL'); + h.done(); + } +}; + +test('abort kills the whole process group: a SIGTERM-deaf grandchild does not outlive the run', async () => { + const controller = new AbortController(); + const survived = await grandchildOutlives('[[grandchild:hang]]', () => ({ signal: controller.signal }), async (run, h) => { + await waitFor(() => h.received().some((entry) => entry.event === 'grandchild')); + controller.abort(); + await rejectsKind(run, 'cancelled'); + }); + assert.equal(survived, false); +}); + +test('a timeout kills the whole process group too', async () => { + const survived = await grandchildOutlives('[[grandchild:hang]]', () => ({ timeoutMs: 1500 }), async (run) => { + await rejectsKind(run, 'timeout'); + }); + assert.equal(survived, false); +}); + +test('an agent that exits mid-turn or ends normally leaves no descendant behind', async () => { + assert.equal(await grandchildOutlives('[[grandchild:exit]]', () => ({}), (run) => rejectsKind(run, 'exited')), false); + assert.equal(await grandchildOutlives('[[grandchild:reply]]', () => ({}), async (run) => { + assert.equal((await run).text, 'SPAWNED'); + }), false); +}); + +test('a session error whose code is not an integer is a protocol failure, never a fallback', async () => { + for (const code of ['"NOT_FOUND"', 'absent', '1.5', 'null']) { + const h = harness({ env: { FAKE_ACP_CAPS: 'resume', FAKE_ACP_FAIL: 'resume', FAKE_ACP_FAIL_CODE: code } }); + try { + const error = await rejectsKind(h.adapter.spawn('hi', h.ctx({ sessionId: 'sess-old' })), 'protocol'); + assert.match(error.message, /session\/resume with an error whose code is not an integer/, code); + assert.ok(!h.events.some((e) => e.type === 'session.opened'), 'no fallback is recorded'); + assert.equal(h.record.verify().ok, true); + } finally { h.done(); } + } +}); + +test('a prompt answered with a non-integer error code is a protocol failure', async () => { + const h = harness(); + try { + await rejectsKind(h.adapter.spawn('[[rpc-error:oops:content policy]]', h.ctx()), 'protocol'); + } finally { h.done(); } +}); + +test('context counts the record cannot hold are recorded as unknown, not refused', async () => { + const h = harness(); + try { + const result = await h.adapter.spawn('[[bad-usage]]', h.ctx()); + assert.equal(result.text, 'OK'); + assert.equal(result.contextTokens, undefined); + const usage = h.events.filter((e) => e.type === 'usage').map((e) => [e.data.contextTokens, e.data.contextSize]); + assert.deepEqual(usage, [[null, null], [null, 1000]]); + } finally { h.done(); } +}); + +test('a trace writer that refuses an entry ends the run with its error instead of crashing', async () => { + const h = harness(); + try { + const refusing = { emit(type) { if (type === 'tool.call') throw new Error('record: refused'); } }; + const error = await h.adapter.spawn('hello', h.ctx({ trace: refusing })).then(() => null, (e) => e); + assert.match(String(error?.message), /record: refused/); + } finally { h.done(); } +}); + +const providerAttempts = (h) => h.received().filter((entry) => entry.event === 'provider-attempt').map((entry) => entry.status); + +test('the first provider refusal on the wire ends the run as refusal_retry_loop, fast', async () => { + const provider = await startFakeProvider(); + const h = harness(); + try { + const { error, afterFirstMs } = await settleAfterFirstRequest(provider, rejectsKind(h.adapter.spawn('[[provider:30]] [[refuse]]', + h.ctx({ provider: { baseURL: provider.baseURL, apiKey: KEY, model: 'm' } })), 'refusal_retry_loop')); + // The agent's own loop would run 30 attempts 200 ms apart (6 s and more). + assert.ok(afterFirstMs < 3000, `settled ${afterFirstMs}ms after the first provider call`); + assert.equal(provider.requests.length, 1, 'the provider sees the refused request once, never a retry'); + assert.match(error.detail, /400/); + assert.ok(!error.message.includes(KEY)); + // The CLI's retries after the trip are answered by the relay, not the provider. + assert.equal(providerAttempts(h)[0], 400); + assert.ok(providerAttempts(h).slice(1).every((status) => status === 503)); + assert.equal(h.events.at(-1).type, 'process.exited'); + } finally { await provider.close(); h.done(); } +}); + +test('the first provider 5xx on the wire ends the run as provider_unavailable, fast', async () => { + const provider = await startFakeProvider(); + const h = harness(); + try { + const { error, afterFirstMs } = await settleAfterFirstRequest(provider, rejectsKind(h.adapter.spawn('[[provider:30]] [[fail:503]]', + h.ctx({ provider: { baseURL: provider.baseURL, apiKey: KEY, model: 'm' } })), 'provider_unavailable')); + assert.ok(afterFirstMs < 3000, `settled ${afterFirstMs}ms after the first provider call`); + assert.equal(error.retryable, true); + assert.equal(provider.requests.length, 1); + } finally { await provider.close(); h.done(); } +}); + +test('an unreachable provider ends the run as provider_unavailable', async () => { + const h = harness(); + try { + await rejectsKind(h.adapter.spawn('[[provider:8]]', h.ctx()), 'provider_unavailable'); + } finally { h.done(); } +}); + +test('a healthy provider passes through the relay untouched', async () => { + const provider = await startFakeProvider(); + const h = harness(); + try { + const result = await h.adapter.spawn('[[provider:3]] [[reply:HELLO]]', + h.ctx({ provider: { baseURL: provider.baseURL, apiKey: KEY, model: 'm' } })); + assert.equal(result.text, 'PROVIDER:HELLO'); + assert.equal(provider.requests.length, 1); + assert.equal(provider.requests[0].authorization, 'present'); + } finally { await provider.close(); h.done(); } +}); + +test('with watchProvider off the CLI\'s retries reach the provider (the watch is what stops them)', async () => { + const provider = await startFakeProvider(); + const h = harness({ watchProvider: false }); + try { + const result = await h.adapter.spawn('[[provider:3]] [[refuse]]', + h.ctx({ provider: { baseURL: provider.baseURL, apiKey: KEY, model: 'm' } })); + assert.equal(result.text, 'PROVIDER:400,400,400'); + assert.equal(provider.requests.length, 3); + } finally { await provider.close(); h.done(); } +}); + +async function waitFor(predicate, timeoutMs = 5000) { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() > deadline) throw new Error('waitFor timed out'); + await new Promise((resolve) => { setTimeout(resolve, 10); }); + } +} + +test('opencode retry reports reach the record as agent.retry, without the provider text', async () => { + const h = harness({ readRetryStatus: readOpencodeRetry }); + try { + const result = await h.adapter.spawn('[[opencode-retry]]', h.ctx()); + assert.equal(result.text, 'OK'); + const retries = h.events.filter((e) => e.type === 'agent.retry').map((e) => e.data); + assert.deepEqual(retries, [ + { state: 'scheduled', attempt: 2, nextRetryAt: '2026-10-25T13:00:02.000Z', status: 503, errorType: 'provider.internal' }, + { state: 'started', attempt: 2 }, + { state: 'scheduled', attempt: 3, nextRetryAt: '2026-10-25T13:00:03.000Z', status: 429, errorType: 'provider.rate-limit' }, + { state: 'started', attempt: 3 }, + // Malformed fields are absent, never null; a type outside the record's set is `other`. + { state: 'scheduled', errorType: 'other' }, + { state: 'scheduled', errorType: 'other' }, + { state: 'started' }, + { state: 'scheduled', attempt: 5, nextRetryAt: '2026-10-25T13:00:05.000Z' }, + ]); + // The adapter's own output, before the store redacts anything. + const emitted = JSON.stringify(h.events); + assert.ok(!emitted.includes(KEY), 'the key is never emitted'); + assert.ok(!emitted.includes('raw provider text') && !emitted.includes('upstream said'), 'neither the message nor the body is emitted'); + assert.equal(h.record.lines().filter((e) => e.type === 'agent.retry').length, 8); + assert.equal(h.record.verify().ok, true); + } finally { h.done(); } +}); + +test('without a retry reader, session_info_update is dropped as session state', async () => { + const h = harness(); + try { + assert.equal((await h.adapter.spawn('[[opencode-retry]]', h.ctx())).text, 'OK'); + assert.ok(!h.events.some((e) => e.type === 'agent.retry')); + } finally { h.done(); } +}); + +test('readOpencodeRetry reads only the opencode/retry key', () => { + assert.equal(readOpencodeRetry({ sessionUpdate: 'session_info_update', title: 't' }), null); + assert.equal(readOpencodeRetry({ _meta: { 'other/retry': {} } }), null); + assert.equal(readOpencodeRetry({ _meta: { retry: { attempt: 2 } } }), null); + assert.equal(readOpencodeRetry({ _meta: 'opencode/retry' }), null); + assert.equal(readOpencodeRetry({ _meta: { 'opencode/retry': 'x' } }), null); + assert.deepEqual(readOpencodeRetry({ _meta: { 'opencode/retry': null } }), { state: 'started' }); + const report = readOpencodeRetry({ _meta: { 'opencode/retry': { attempt: 2, nextRetryAt: 'n', error: { type: 't', status: 500, message: 'm', response: { body: 'b' } } } } }); + assert.deepEqual(report, { state: 'scheduled', attempt: 2, nextRetryAt: 'n', status: 500, errorType: 't' }); +}); diff --git a/packages/seats/test/provider-watch.test.js b/packages/seats/test/provider-watch.test.js new file mode 100644 index 0000000000..539526a1ea --- /dev/null +++ b/packages/seats/test/provider-watch.test.js @@ -0,0 +1,135 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { failureForStatus, startProviderWatch } from '../src/provider-watch.js'; +import { startFakeProvider } from '../test-support/fake-provider.js'; +import { RunFailure } from '../src/failures.js'; +import { createServer } from 'node:http'; + +const KEY = ['sk', 'test', '0123456789abcdef'].join('-'); + +const post = (baseURL, content, { stream = false, signal } = {}) => fetch(`${baseURL}/chat/completions`, { + method: 'POST', + signal, + headers: { 'content-type': 'application/json', authorization: `Bearer ${KEY}` }, + body: JSON.stringify({ model: 'fake-model', stream, messages: [{ role: 'user', content }] }), +}); + +const watched = async () => { + const provider = await startFakeProvider(); + const failures = []; + const watch = await startProviderWatch({ + baseURL: provider.baseURL, credentials: [KEY], onFailure: (failure) => failures.push(failure), + }); + return { + provider, watch, failures, close: async () => { await watch.close(); await provider.close(); }, + }; +}; + +test('failureForStatus: 4xx is refusal_retry_loop; 5xx, 408 and 429 are provider_unavailable; 2xx/3xx are fine', () => { + for (const status of [400, 401, 403, 404, 422]) assert.equal(failureForStatus(status, '').kind, 'refusal_retry_loop', String(status)); + for (const status of [408, 429, 500, 502, 503, 504]) assert.equal(failureForStatus(status, '').kind, 'provider_unavailable', String(status)); + for (const status of [200, 201, 204, 301, 399]) assert.equal(failureForStatus(status, ''), null, String(status)); + assert.equal(failureForStatus(400, '').retryable, false); + assert.equal(failureForStatus(503, '').retryable, true); +}); + +test('failureForStatus scrubs a credential the provider echoed back', () => { + const failure = failureForStatus(401, JSON.stringify({ error: { message: `bad key ${KEY}` } }), { credentials: [KEY] }); + assert.ok(failure instanceof RunFailure); + assert.ok(!failure.message.includes(KEY), failure.message); +}); + +test('the relay passes healthy answers through, streamed and not, and keeps the base path', async () => { + const w = await watched(); + try { + assert.match(w.watch.baseURL, /^http:\/\/127\.0\.0\.1:\d+\/v1$/); + const plain = await post(w.watch.baseURL, 'hi'); + assert.equal(plain.status, 200); + assert.equal((await plain.json()).choices[0].message.content, 'PONG'); + const streamed = await post(w.watch.baseURL, '[[reply:STREAMED]]', { stream: true }); + assert.match(await streamed.text(), /STREAMED[\s\S]*\[DONE\]/); + assert.equal(w.provider.requests.length, 2); + assert.equal(w.provider.requests[0].url, '/v1/chat/completions'); + assert.equal(w.provider.requests[0].authorization, 'present'); + assert.equal(w.failures.length, 0); + assert.equal(w.watch.tripped, null); + } finally { await w.close(); } +}); + +test('the first refusal trips once, its body reaches the CLI, and nothing more is forwarded', async () => { + const w = await watched(); + try { + const first = await post(w.watch.baseURL, '[[refuse]]'); + assert.equal(first.status, 400); + assert.match(JSON.stringify(await first.json()), /content policy/, 'the CLI still sees the real error body'); + assert.equal(w.failures.length, 1); + assert.equal(w.failures[0].kind, 'refusal_retry_loop'); + assert.match(w.failures[0].detail, /400/); + for (let attempt = 0; attempt < 3; attempt += 1) { + assert.equal((await post(w.watch.baseURL, '[[refuse]]')).status, 503); + } + assert.equal(w.provider.requests.length, 1, 'retries after the trip never reach the provider'); + assert.equal(w.failures.length, 1, 'onFailure fires once'); + assert.equal(w.watch.tripped, w.failures[0]); + } finally { await w.close(); } +}); + +test('the first 503 trips as provider_unavailable', async () => { + const w = await watched(); + try { + assert.equal((await post(w.watch.baseURL, '[[fail:503]]')).status, 503); + assert.deepEqual(w.failures.map((failure) => failure.kind), ['provider_unavailable']); + } finally { await w.close(); } +}); + +test('a failing GET is relayed but never judged', async () => { + const w = await watched(); + try { + const listing = await fetch(`${w.watch.baseURL}/models`); + assert.equal(listing.status, 200); + const missing = await fetch(`${w.watch.baseURL}/no-such-thing`); + assert.equal(missing.status, 404); + assert.equal(w.failures.length, 0); + } finally { await w.close(); } +}); + +test('an unreachable provider trips as provider_unavailable on the first POST', async () => { + const failures = []; + const watch = await startProviderWatch({ baseURL: 'http://127.0.0.1:9/v1', onFailure: (failure) => failures.push(failure) }); + try { + await post(watch.baseURL, 'hi').catch(() => null); + assert.deepEqual(failures.map((failure) => failure.kind), ['provider_unavailable']); + } finally { await watch.close(); } +}); + +test('a client that hangs up takes its upstream request with it', async () => { + // An upstream that never answers and counts the connections held open. + let open = 0; + let seen = 0; + const upstream = createServer(() => { seen += 1; }); + upstream.on('connection', (socket) => { open += 1; socket.on('close', () => { open -= 1; }); }); + await new Promise((resolve) => { upstream.listen(0, '127.0.0.1', resolve); }); + const failures = []; + const watch = await startProviderWatch({ + baseURL: `http://127.0.0.1:${upstream.address().port}/v1`, onFailure: (failure) => failures.push(failure), + }); + const waitUntil = async (predicate) => { + const deadline = Date.now() + 10000; + while (!predicate() && Date.now() < deadline) await new Promise((r) => { setTimeout(r, 10); }); + }; + try { + const controller = new AbortController(); + const held = post(watch.baseURL, 'hi', { signal: controller.signal }).catch(() => null); + await waitUntil(() => seen === 1); + assert.equal(open, 1); + controller.abort(); + await held; + await waitUntil(() => open === 0); + assert.equal(open, 0, 'the relay dropped the upstream request when the client hung up'); + assert.equal(failures.length, 0, 'a hang-up is not a provider failure'); + } finally { + await watch.close(); + upstream.closeAllConnections(); + await new Promise((resolve) => { upstream.close(resolve); }); + } +}); From 522e64bcd0380bef26a46a82d1e334ef560377f7 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Tue, 27 Oct 2026 23:34:22 -0700 Subject: [PATCH 2/3] fix(seats): register the ACP launch's provider key with the trace store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Otto showed the key the launch puts in the child's env was never registered, so agent text echoing it (a goose "Ran into this error: …" reply, a tool title) was written to the trace in clear. The test helper pre-registered the key, which hid it. - spawn registers acpCredentials(ctx, launched.env) right after launch, the one point every ACP launch passes. The new test uses a store with nothing pre-registered. - stderr is scrubbed whole before the 400-char tail, so the window can't split a key. - provider-watch scrubs `NNN: body`, so the keep-list matches and keeps only the body's message field, as its comment says. - SAFE_ENV names the locale variables, not any LC_ prefix. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MDs9DRK3nqxjeAV7xL8uhg --- packages/seats/src/adapters/acp-launch.js | 6 +- packages/seats/src/adapters/acp.js | 10 ++- packages/seats/src/provider-watch.js | 22 +++--- .../seats/test-support/fake-acp-agent.mjs | 3 +- packages/seats/test/acp.test.js | 68 ++++++++++++++++++- packages/seats/test/provider-watch.test.js | 11 +++ 6 files changed, 103 insertions(+), 17 deletions(-) diff --git a/packages/seats/src/adapters/acp-launch.js b/packages/seats/src/adapters/acp-launch.js index e65857855d..5956593ac4 100644 --- a/packages/seats/src/adapters/acp-launch.js +++ b/packages/seats/src/adapters/acp-launch.js @@ -19,7 +19,11 @@ import { join } from 'node:path'; import { runFailure } from '../failures.js'; import { createAcpAdapter, readsAsRefusal } from './acp.js'; -const SAFE_ENV = /^(?:PATH|LANG|LC_[A-Z_]+|TERM|USER|LOGNAME|NO_COLOR|CI|TMPDIR|SSL_CERT_FILE|SSL_CERT_DIR)$/; +// The locale names are the POSIX and glibc categories by name, never an LC_ +// prefix: a prefix match admits any variable that starts with it +// (LC_SECRET_TOKEN), and the child env is where the provider key lives. +const LOCALE_ENV = 'LC_(?:ALL|CTYPE|COLLATE|MESSAGES|MONETARY|NUMERIC|TIME|ADDRESS|IDENTIFICATION|MEASUREMENT|NAME|PAPER|TELEPHONE)'; +const SAFE_ENV = new RegExp(`^(?:PATH|LANG|${LOCALE_ENV}|TERM|USER|LOGNAME|NO_COLOR|CI|TMPDIR|SSL_CERT_FILE|SSL_CERT_DIR)$`); const VERSION_TIMEOUT_MS = 5000; /** The provider route, checked: a plain HTTP(S) base URL, a key and a model. */ diff --git a/packages/seats/src/adapters/acp.js b/packages/seats/src/adapters/acp.js index 83fa793716..70f9bd7afe 100644 --- a/packages/seats/src/adapters/acp.js +++ b/packages/seats/src/adapters/acp.js @@ -262,7 +262,9 @@ const runTurn = ({ } const exitedFailure = () => { - const tail = stderr.trim().slice(-STDERR_IN_FAILURE); + // Scrub the whole buffer before taking the tail: a window cut first can + // split a key, and the exact-match scrub cannot see half of one. + const tail = redactKnownCredentials(stderr, credentials).trim().slice(-STDERR_IN_FAILURE); const how = exitInfo ? `code ${exitInfo.code}, signal ${exitInfo.signal}` : 'no exit status'; return runFailure('exited', `${name} exited before the turn settled (${how})${tail ? `: ${tail}` : ''}`, { credentials, limit: FAILURE_LIMIT, @@ -601,6 +603,12 @@ export const createAcpAdapter = ({ let launched; try { launched = await launch(watch ? { ...ctx, provider: { ...ctx.provider, baseURL: watch.baseURL } } : ctx); + // The store scrubs only values registered with it, and the launch put + // the provider key (and any secret-named variable) in the child's + // env, so agent text can echo it: a goose "Ran into this error: …" + // reply, a tool title, a plan entry. Every launch path passes here, + // before the agent can speak. + traceOf(ctx).redact?.(acpCredentials(ctx, launched?.env)); return await runTurn({ name, launched, prompt, ctx, memoryPreamble, readReplyFailure, readRetryStatus, maxLineChars, timings: resolvedTimings, wire, }); diff --git a/packages/seats/src/provider-watch.js b/packages/seats/src/provider-watch.js index 913d113b76..787e9510cb 100644 --- a/packages/seats/src/provider-watch.js +++ b/packages/seats/src/provider-watch.js @@ -21,23 +21,23 @@ */ import { createServer, request as httpRequest } from 'node:http'; import { request as httpsRequest } from 'node:https'; -import { runFailure } from './failures.js'; +import { RunFailure, runFailure } from './failures.js'; +import { scrubAdapterFailure } from './upstream-refusal.js'; const BODY_SAMPLE_CHARS = 4096; /** The failure a provider status names, or null when the answer is fine. */ export const failureForStatus = (status, body, { credentials = [] } = {}) => { - // The status leads the text so the shared keep-list reads `NNN: {body}` - // and keeps at most the body's one named message field. - const text = `${status}: ${String(body ?? '').slice(0, BODY_SAMPLE_CHARS)}`; const transient = status === 408 || status === 429 || (status >= 500 && status <= 599); - if (transient) { - return runFailure('provider_unavailable', `provider answered ${text}`, { credentials, limit: 500 }); - } - if (status >= 400 && status <= 499) { - return runFailure('refusal_retry_loop', `provider answered ${text}`, { credentials, limit: 500 }); - } - return null; + const kind = transient ? 'provider_unavailable' : status >= 400 && status <= 499 ? 'refusal_retry_loop' : null; + if (!kind) return null; + // The status must LEAD the text the shared keep-list reads, so it parses + // `NNN: {body}` and keeps at most the body's one named message field. The + // readable prefix is added after the scrub; put first, it hides the status + // and the whole body sample is kept instead. + const text = `${status}: ${String(body ?? '').slice(0, BODY_SAMPLE_CHARS)}`; + const { detail } = scrubAdapterFailure(text, { credentials, limit: 500 }); + return new RunFailure(kind, `provider answered ${status}${detail ? `: ${detail}` : ''}`); }; /** diff --git a/packages/seats/test-support/fake-acp-agent.mjs b/packages/seats/test-support/fake-acp-agent.mjs index 55901a189d..01ea6447a0 100644 --- a/packages/seats/test-support/fake-acp-agent.mjs +++ b/packages/seats/test-support/fake-acp-agent.mjs @@ -19,6 +19,7 @@ * [[malformed]] write a non-JSON line, then hang * [[huge:N]] write one N-char line, then hang * [[exit]] write OPENAI_API_KEY to stderr and exit 3 + * [[exit:N]] same, with N dots after the key on its line * [[hang]] never answer until session/cancel, then stopReason cancelled * [[hang-deaf]] never answer, ignore cancel and SIGTERM * [[hang-permission]] ask permission and wait; on cancel answer cancelled @@ -130,7 +131,7 @@ const runPrompt = async (id, { sessionId, prompt }) => { case 'malformed': process.stdout.write('this is not json\n'); return undefined; case 'huge': process.stdout.write(`${'x'.repeat(Number(arg) || 1000)}\n`); return undefined; case 'exit': - process.stderr.write(`fatal: provider rejected key ${process.env.OPENAI_API_KEY}\n`); + process.stderr.write(`fatal: provider rejected key ${process.env.OPENAI_API_KEY}${'.'.repeat(Number(arg) || 0)}\n`); setTimeout(() => process.exit(3), 10); return undefined; case 'hang': activePrompt = { id, sessionId }; return undefined; diff --git a/packages/seats/test/acp.test.js b/packages/seats/test/acp.test.js index 7943a08501..443a11bee6 100644 --- a/packages/seats/test/acp.test.js +++ b/packages/seats/test/acp.test.js @@ -1,12 +1,14 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { existsSync, mkdtempSync, readFileSync, rmSync, statSync } from 'node:fs'; +import { + existsSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, +} from 'node:fs'; import { tmpdir } from 'node:os'; -import { join } from 'node:path'; +import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { PassThrough } from 'node:stream'; import { createAcpAdapter, defaultPermissionDecision, readsAsRefusal } from '../src/adapters/acp.js'; -import { readOpencodeRetry } from '../src/adapters/acp-launch.js'; +import { isolatedEnv, readOpencodeRetry } from '../src/adapters/acp-launch.js'; import { createAcpClient } from '../src/acp/client.js'; import { launchGoose, launchOpencode, readGooseReplyFailure, acpGoose, acpOpencode, @@ -15,6 +17,8 @@ import { RunFailure, runFailure } from '../src/failures.js'; import { recordTrace } from '../test-support/record-trace.js'; import { createPermissionPolicy } from '../src/valet/permissions.js'; import { startFakeProvider } from '../test-support/fake-provider.js'; +import { superviseRun } from '../src/supervisor.js'; +import { createTraceStore } from '../src/trace/store.js'; const AGENT = fileURLToPath(new URL('../test-support/fake-acp-agent.mjs', import.meta.url)); const KEY = ['sk', 'test', '0123456789abcdef'].join('-'); @@ -792,3 +796,61 @@ test('readOpencodeRetry reads only the opencode/retry key', () => { const report = readOpencodeRetry({ _meta: { 'opencode/retry': { attempt: 2, nextRetryAt: 'n', error: { type: 't', status: 500, message: 'm', response: { body: 'b' } } } } }); assert.deepEqual(report, { state: 'scheduled', attempt: 2, nextRetryAt: 'n', status: 500, errorType: 't' }); }); + +test('the secrets the launch injects are scrubbed from the record even when nothing pre-registered them', async () => { + // A store with NOTHING registered: the adapter itself must hand the store + // the key it put in the child's env, or agent text carries it in clear. + // A secret-named variable the launch minted itself is not in ctx at all; + // only the launch env names it. + const MINTED = 'launch-minted-session-0123456789'; + const replies = [ + `Ran into this error: Authentication failed (401) for key ${KEY}`, + `debug key=${KEY}`, + `debug session=${MINTED}`, + ]; + for (const reply of replies) { + const h = harness({ env: { SESSION_TOKEN: MINTED } }); + const root = join(scratch(), 'store'); + try { + const store = createTraceStore({ root }); + const ctx = { ...h.ctx(), trace: undefined }; + const run = await superviseRun({ adapter: h.adapter, prompt: `[[reply:${reply}]]`, ctx, store }); + assert.equal(run.outcome, 'ok', JSON.stringify(run.failure?.message)); + const lines = store.readRun(run.runId); + const chunks = lines.filter((line) => line.type === 'agent.message_chunk'); + assert.ok(chunks.length > 0, 'the reply was recorded'); + assert.match(chunks.map((line) => line.data.text).join(''), /\[redacted\]/); + for (const file of readdirSync(root, { recursive: true, withFileTypes: true }).filter((d) => d.isFile())) { + const text = readFileSync(join(file.parentPath ?? file.path, file.name), 'utf8'); + assert.equal(text.includes(KEY), false, `the provider key appears in ${file.name} for reply ${JSON.stringify(reply)}`); + assert.equal(text.includes(MINTED), false, `the launch-minted secret appears in ${file.name}`); + } + } finally { h.done(); rmSync(dirname(root), { recursive: true, force: true }); } + } +}); + +test('a key the stderr window would cut is scrubbed whole before the tail is taken', async () => { + const h = harness(); + try { + // The key ends 390 characters before the end of the line, so a 400-char + // window taken before the scrub keeps its last ten characters. + const error = await rejectsKind(h.adapter.spawn('[[exit:390]]', h.ctx()), 'exited'); + assert.equal(error.message.includes(KEY.slice(-10)), false, error.message); + assert.match(error.message, /redacted\]\.{390}$/); + } finally { h.done(); } +}); + +test('isolatedEnv passes the real locale categories and no other LC_-named variable', async () => { + const locale = { + LC_ALL: 'C', LC_CTYPE: 'C', LC_COLLATE: 'C', LC_MESSAGES: 'C', LC_MONETARY: 'C', LC_NUMERIC: 'C', LC_TIME: 'C', + LC_ADDRESS: 'C', LC_IDENTIFICATION: 'C', LC_MEASUREMENT: 'C', LC_NAME: 'C', LC_PAPER: 'C', LC_TELEPHONE: 'C', + }; + const home = scratch(); + try { + const { env } = await isolatedEnv({ + env: { ...locale, HOME: home, LC_SECRET_TOKEN: KEY, LC_TERMINAL: 'iTerm2', LC_: 'x', PATH: '/bin' }, + }, 'test'); + for (const name of Object.keys(locale)) assert.equal(env[name], 'C', name); + for (const name of ['LC_SECRET_TOKEN', 'LC_TERMINAL', 'LC_']) assert.equal(name in env, false, name); + } finally { rmSync(home, { recursive: true, force: true }); } +}); diff --git a/packages/seats/test/provider-watch.test.js b/packages/seats/test/provider-watch.test.js index 539526a1ea..026cce7da2 100644 --- a/packages/seats/test/provider-watch.test.js +++ b/packages/seats/test/provider-watch.test.js @@ -133,3 +133,14 @@ test('a client that hangs up takes its upstream request with it', async () => { await new Promise((resolve) => { upstream.close(resolve); }); } }); + +test('failureForStatus reads its own status prefix: a JSON body keeps only its named message', () => { + const extra = 'x'.repeat(2000); + for (const [status, kind] of [[401, 'refusal_retry_loop'], [503, 'provider_unavailable']]) { + const failure = failureForStatus(status, JSON.stringify({ error: { message: 'model is overloaded' }, extra })); + assert.equal(failure.kind, kind); + assert.equal(failure.detail, `provider answered ${status}: model is overloaded`); + } + // A structured body with no named field contributes nothing; the status stays. + assert.equal(failureForStatus(400, JSON.stringify({ trace: extra })).detail, 'provider answered 400'); +}); From c96a064d4c420fafd45492ff98b66b539c73ab36 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Tue, 27 Oct 2026 23:59:09 -0700 Subject: [PATCH 3/3] fix(seats): keep-list reads a JSON provider body whole A JSON error body past the 4096-char sample stopped parsing, so the keep-list treated it as prose and kept its first 500 characters, fields outside the list included (CodeRabbit). A JSON body is now read whole; one past 256 KiB keeps no detail. The test fails with the old slice. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MDs9DRK3nqxjeAV7xL8uhg --- packages/seats/src/provider-watch.js | 9 ++++++++- packages/seats/test/provider-watch.test.js | 11 +++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/packages/seats/src/provider-watch.js b/packages/seats/src/provider-watch.js index 787e9510cb..b93641ce4f 100644 --- a/packages/seats/src/provider-watch.js +++ b/packages/seats/src/provider-watch.js @@ -25,6 +25,10 @@ import { RunFailure, runFailure } from './failures.js'; import { scrubAdapterFailure } from './upstream-refusal.js'; const BODY_SAMPLE_CHARS = 4096; +// A JSON body is read whole, because a slice of one is no longer JSON and the +// keep-list would then treat it as prose and keep fields outside the list. +// Past this size no detail is kept at all. +const JSON_BODY_MAX_CHARS = 256 * 1024; /** The failure a provider status names, or null when the answer is fine. */ export const failureForStatus = (status, body, { credentials = [] } = {}) => { @@ -35,7 +39,10 @@ export const failureForStatus = (status, body, { credentials = [] } = {}) => { // `NNN: {body}` and keeps at most the body's one named message field. The // readable prefix is added after the scrub; put first, it hides the status // and the whole body sample is kept instead. - const text = `${status}: ${String(body ?? '').slice(0, BODY_SAMPLE_CHARS)}`; + const raw = String(body ?? ''); + const json = /^\s*[[{]/.test(raw); + if (json && raw.length > JSON_BODY_MAX_CHARS) return new RunFailure(kind, `provider answered ${status}`); + const text = `${status}: ${json ? raw : raw.slice(0, BODY_SAMPLE_CHARS)}`; const { detail } = scrubAdapterFailure(text, { credentials, limit: 500 }); return new RunFailure(kind, `provider answered ${status}${detail ? `: ${detail}` : ''}`); }; diff --git a/packages/seats/test/provider-watch.test.js b/packages/seats/test/provider-watch.test.js index 026cce7da2..4a58c8bd26 100644 --- a/packages/seats/test/provider-watch.test.js +++ b/packages/seats/test/provider-watch.test.js @@ -144,3 +144,14 @@ test('failureForStatus reads its own status prefix: a JSON body keeps only its n // A structured body with no named field contributes nothing; the status stays. assert.equal(failureForStatus(400, JSON.stringify({ trace: extra })).detail, 'provider answered 400'); }); + +test('failureForStatus applies the keep-list to a JSON body longer than the prose sample', () => { + // Past 4096 chars a slice would no longer parse as JSON and be kept as prose. + const body = JSON.stringify({ internal: 'acct-9f3e private routing note', pad: 'p'.repeat(5000), error: { message: 'quota exceeded' } }); + const failure = failureForStatus(429, body); + assert.equal(failure.detail, 'provider answered 429: quota exceeded'); + assert.doesNotMatch(failure.detail, /acct-9f3e|routing note|ppp/); + // Too large to read whole: no detail rather than a partial one. + const huge = JSON.stringify({ internal: 'acct-9f3e', pad: 'p'.repeat(300 * 1024) }); + assert.equal(failureForStatus(500, huge).detail, 'provider answered 500'); +});