src/devin-acp.js
Rama master · réplica de lectura del repositorio original.
import { spawn } from 'child_process';
import { VERSION } from './version.js';
import { log } from './config.js';
import { sanitizeToolName } from './tool-names.js';
import { buildSubprocessEnv, buildMinimalSubprocessEnv } from './log-safety.js';
// Spawn seam: `spawn` is captured behind an indirection so a test can inject a
// fake child (EventEmitter stdio) to exercise the stdin 'error'/EPIPE guard
// without a real Devin CLI or network. Production always uses the real spawn.
let _spawn = spawn;
export function __setDevinAcpSpawnForTest(fn) { _spawn = fn || spawn; }
function intEnv(name, fallback, min = 0) {
const n = parseInt(process.env[name] || '', 10);
return Number.isFinite(n) && n >= min ? n : fallback;
}
function runTimeoutMs() {
return intEnv('DEVIN_TIMEOUT_MS', 10 * 60_000, 1000);
}
function outputLimitBytes() {
return intEnv('DEVIN_OUTPUT_LIMIT_BYTES', 4 * 1024 * 1024, 1024);
}
function parseAcpArgs() {
const raw = process.env.DEVIN_CLI_ACP_ARGS_JSON || '';
if (!raw.trim()) return ['acp'];
try {
const args = JSON.parse(raw);
if (!Array.isArray(args) || !args.every(x => typeof x === 'string')) {
throw new Error('must be a JSON string array');
}
return args;
} catch (err) {
throw Object.assign(new Error(`Invalid DEVIN_CLI_ACP_ARGS_JSON: ${err.message}`), {
status: 500,
type: 'backend_misconfigured',
});
}
}
// ── Zero-billable proactive availability probe (AC1 gap-e / §4) ──────────────
// AC1 left "probe before we spawn" as a nice-to-have: today the ACP runner only
// learns the Devin CLI is missing AFTER it has reserved a pool account and
// spawned `devin acp`, then catches ENOENT → 503. probeDevinCliAvailable lets a
// router fail FAST and gracefully before burning a checkout/spawn.
//
// CRITICAL — this probe is strictly NON-billable. It runs `devin --version`
// (exit-code + presence check) and NOTHING else. It never calls
// initialize/authenticate/session.new/session/prompt, never opens a session,
// never sends a prompt — so it cannot consume account quota or upstream tokens.
// It only answers "is the binary on this box runnable?", which mirrors exactly
// what the real `spawn(command, ['acp'])` would hit, so a probe ENOENT predicts
// a runtime ENOENT.
//
// Result is cached for a short TTL (DEVIN_ACP_PROBE_TTL_MS) so a hot request
// path doesn't fork a child on every call. DEVIN_ACP_PROBE=0 disables the probe
// entirely (callers fall back to the passive post-spawn ENOENT→503 path).
let _probeCache = null; // { at: epochMs, result }
function parseProbeArgs() {
const raw = process.env.DEVIN_ACP_PROBE_ARGS_JSON || '';
if (!raw.trim()) return ['--version'];
try {
const args = JSON.parse(raw);
if (!Array.isArray(args) || !args.every(x => typeof x === 'string')) {
throw new Error('must be a JSON string array');
}
return args;
} catch {
// A misconfigured probe-args override must never block routing — fall back
// to the safe default rather than throwing inside an availability check.
return ['--version'];
}
}
function probeOnce() {
const command = process.env.DEVIN_CLI_PATH || 'devin';
const args = parseProbeArgs();
const timeoutMs = intEnv('DEVIN_ACP_PROBE_TIMEOUT_MS', 5000, 100);
return new Promise((resolve) => {
let settled = false;
const done = (r) => { if (!settled) { settled = true; resolve(r); } };
let child;
try {
// STRICTEST env of any spawn in the repo: this binary is resolved via
// PATH (or an operator-supplied DEVIN_CLI_PATH), so a hijacked binary
// would have received every credential in process.env — including the
// credential-store KEK. `--version` needs nothing beyond exec basics.
child = spawn(command, args, {
env: buildMinimalSubprocessEnv(),
windowsHide: true,
stdio: ['ignore', 'ignore', 'ignore'],
});
} catch (err) {
done({ available: false, reason: 'spawn_failed', detail: err?.code || err?.message || 'spawn threw' });
return;
}
const timer = setTimeout(() => {
try { child.kill('SIGTERM'); } catch { /* already gone */ }
// A version check that hangs past the timeout means a wedged binary; treat
// it as unavailable so we don't route to something that can't respond.
done({ available: false, reason: 'probe_timeout' });
}, timeoutMs);
timer.unref?.();
child.on('error', (err) => {
clearTimeout(timer);
// ENOENT = not installed; EACCES = present but not executable. Both mean
// the same real `spawn(command, ['acp'])` would fail the same way.
const reason = err?.code === 'ENOENT' ? 'not_found'
: err?.code === 'EACCES' ? 'not_executable'
: 'spawn_error';
done({ available: false, reason, detail: err?.code || err?.message });
});
child.on('close', (code) => {
clearTimeout(timer);
// The binary exists and is executable (it ran to exit). exitCode 0 is the
// clean "version printed" signal; a non-zero exit still proves PRESENCE —
// some CLI versions don't grok `--version` — so we keep it available to
// avoid false-negatives that would wrongly disable a working backend, and
// surface the code for observability.
done({ available: true, reason: code === 0 ? 'ok' : 'present_nonzero_exit', exitCode: code });
});
});
}
export async function probeDevinCliAvailable({ force = false } = {}) {
// Opt-out: skip the active probe and assume available, deferring to the
// passive post-spawn ENOENT→503 path. Marked skipped so callers can tell.
if (String(process.env.DEVIN_ACP_PROBE || '').trim() === '0') {
return { available: true, skipped: true, reason: 'probe_disabled' };
}
const command = process.env.DEVIN_CLI_PATH || 'devin';
const argsKey = (process.env.DEVIN_ACP_PROBE_ARGS_JSON || '').trim();
const key = `${command} ${argsKey}`;
const ttlMs = intEnv('DEVIN_ACP_PROBE_TTL_MS', 60_000, 0);
const now = Date.now();
// Cache keys on the resolved command (+ probe-args override): a config change
// that points at a different binary must re-probe, not reuse a stale verdict.
if (!force && ttlMs > 0 && _probeCache && _probeCache.key === key && (now - _probeCache.at) < ttlMs) {
return { ..._probeCache.result, cached: true };
}
const result = await probeOnce();
_probeCache = { at: now, key, result };
return { ...result, cached: false };
}
// Test seam: drop the cached probe result so each test observes a fresh probe.
export function __resetDevinAcpProbeCache() {
_probeCache = null;
}
function writeJsonLine(child, payload) {
if (!child.stdin?.writable) return;
child.stdin.write(`${JSON.stringify(payload)}\n`);
}
function parseJsonLine(line) {
try {
return JSON.parse(line);
} catch {
return null;
}
}
// ── Transient-first ACP error classification ────────────────────────────────
// The ACP path spawns `devin acp` which talks to the SAME upstream as
// DEVIN_CONNECT (server.codeium.com — authenticate _meta.api_server_url). That
// upstream is observed (DEVIN_CONNECT live, free account <redacted>) to
// wrap TRANSIENT faults inside an auth-shaped envelope:
// - "We're currently facing high demand for this model..." (capacity)
// - "an internal error occurred (trace ID: ...)" (backend blip in a 401/403)
// If those reach the pool as a plain 502 backend_error they fall through to the
// windowed generic-error streak and DISABLE a perfectly healthy account after
// 3 hits — i.e. a retryable hiccup burns a working credential.
//
// So we classify ACP RPC errors with the SAME transient-first ordering as
// devin-connect.js classifyUpstreamError, then choose an (status, type, message)
// tuple that makes the EXISTING reportRunFailure (special-agent.js) route safely
// WITHOUT changing that file:
// - CAPACITY → status 429 → reportRunFailure → markRateLimited
// (fixed cooldown, rotates out, NOT disabled). retryable.
// - UPSTREAM_INTERNAL→ status 502 + message keeps "internal error" →
// reportRunFailure → reportInternalError (account-sticky
// quarantine, NOT a re-login storm). NON-retryable
// (observed persistent 3/3 — same-process replay just
// amplifies load, mirrors devin-connect isRetryable).
// - UNAUTHORIZED → status 401 → reportRunFailure → genericError; a GENUINE
// dead token disables after the streak. Transient blips in
// a 401 shell are already caught by the two branches above
// (transient-first), so only real auth failures land here.
// - RATE_LIMITED → status 429 → markRateLimited.
// - else → 502 backend_error (unchanged legacy behaviour).
export function classifyAcpError(message, rpcCode = null) {
const body = String(message || '').trim();
const lc = body.toLowerCase();
// Capacity / high-demand throttling — match BEFORE auth so a momentary blip
// never reads as a dead token. Pattern parity with devin-connect.js:597.
if (/high demand|try again later|currently (busy|overloaded|at capacity)|model is (busy|overloaded)|temporarily (busy|overloaded|unavailable)|server is busy|overloaded|(service|backend|model|server) (is )?(temporarily )?unavailable|capacity/i.test(lc)) {
return { code: 'CAPACITY', type: 'capacity_error', status: 429, retryable: true };
}
// "an internal error occurred (trace ID: ...)" — a transient BACKEND fault,
// NOT a dead session token, even inside a 401/403 shell. NON-retryable
// (devin-connect.js:643-647: observed persistent, same-token replay amplifies).
if (/internal error occurred|internal error/i.test(lc)) {
return { code: 'UPSTREAM_INTERNAL', type: 'upstream_internal', status: 502, retryable: false };
}
// Real auth failure.
if (/permission_denied|unauthenticated|unauthorized|invalid.*(api ?key|token)|authentication failed/i.test(lc)) {
return { code: 'UNAUTHORIZED', type: 'unauthorized', status: 401, retryable: false };
}
// Explicit rate limiting.
if (/rate.?limit|too many requests|resource_exhausted/i.test(lc)) {
return { code: 'RATE_LIMITED', type: 'rate_limited', status: 429, retryable: false };
}
return { code: rpcCode ? String(rpcCode) : 'UPSTREAM_ERROR', type: 'backend_error', status: 502, retryable: false };
}
function errorFromRpcResponse(resp, fallback = 'ACP request failed') {
const msg = resp?.error?.message || fallback;
const { code, type, status, retryable } = classifyAcpError(msg, resp?.error?.code);
const err = new Error(msg);
err.status = status;
err.type = type;
err.code = code;
err.retryable = retryable;
return err;
}
function extractAcpUpdate(params) {
const update = params?.update || params?.sessionUpdate || params;
const kind = update?.sessionUpdate || update?.type || update?.kind || update?.name || '';
let text = '';
const content = update?.content || update?.delta || update?.message || null;
if (typeof update?.text === 'string') text = update.text;
else if (typeof content === 'string') text = content;
else if (typeof content?.text === 'string') text = content.text;
else if (Array.isArray(content)) {
text = content
.map(part => typeof part === 'string' ? part : (part?.text || ''))
.filter(Boolean)
.join('');
}
return { kind: String(kind || ''), text };
}
// The assistant's user-visible reply. Only these land in the final text.
const MESSAGE_CHUNK_KINDS = new Set([
'agent_message_chunk',
'agent_message_delta',
'assistant_message_chunk',
]);
// The agent's thinking stream (verified live 2026-06-29 with SWE-1.6 over real
// ACP). It is intentionally kept OUT of the reply text and captured separately
// as reasoning so callers can drop it by default or surface it explicitly.
const THOUGHT_CHUNK_KINDS = new Set([
'agent_thought_chunk',
'agent_thought_delta',
'agent_reasoning_chunk',
]);
// ── Tool-call notifications (GFHarness P1) ──────────────────────────────────
// ACP reports the agent's tool use as `session/update` notifications:
// tool_call {toolCallId, title, kind, status, rawInput?, content?}
// tool_call_update {toolCallId, status?, rawInput?, content?, rawOutput?}
// They used to be dropped: the text pump only knew message/thought kinds, so a
// turn that ended in a tool call surfaced as plain text with finish_reason
// "stop" and the caller never learned a tool ran. Collect them keyed by
// toolCallId and expose them as protocol tool-call events. A `tool_call_update`
// carries the tool's output in `content`/`rawOutput`; that is deliberately NOT
// emitted as a tool result — the API client never produced that result, and
// synthesizing one would fabricate a conversation turn.
const TOOL_UPDATE_KINDS = new Set(['tool_call', 'tool_call_update']);
function normalizeAcpToolInput(rawInput) {
if (rawInput == null) return {};
if (typeof rawInput !== 'string') return rawInput;
// rawInput is ACP `unknown`: it may already be JSON text. Parse it so the
// egress serializes exactly once; a non-JSON string stays a plain value.
try { return JSON.parse(rawInput); } catch { return rawInput; }
}
function applyAcpToolUpdate(buffers, update, updateKind) {
const id = update?.toolCallId || update?.tool_call_id;
if (!id) return;
let call = buffers.toolCalls.get(id);
if (!call) {
if (updateKind !== 'tool_call') {
// ACP requires tool_call before its updates. An update for an unknown id
// carries no tool identity to forward; keep the turn alive and log it.
log.warn(`Devin ACP ${updateKind} for unknown toolCallId ${String(id)} ignored`);
return;
}
call = { id, title: '', kind: '', status: '', rawInput: undefined, hasRawInput: false };
buffers.toolCalls.set(id, call);
}
if (typeof update.title === 'string' && update.title) call.title = update.title;
if (typeof update.kind === 'string' && update.kind) call.kind = update.kind;
if (typeof update.status === 'string' && update.status) call.status = update.status;
// rawInput is authoritative when present; a status-only update must not
// erase the arguments announced by the original tool_call.
if (Object.prototype.hasOwnProperty.call(update, 'rawInput')) {
call.rawInput = update.rawInput;
call.hasRawInput = true;
} else if (Object.prototype.hasOwnProperty.call(update, 'raw_input')) {
call.rawInput = update.raw_input;
call.hasRawInput = true;
}
}
function collectAcpTextFromNotification(obj, buffers, onChunk) {
if (obj?.method !== 'session/update') return;
const { kind, text } = extractAcpUpdate(obj.params || {});
if (TOOL_UPDATE_KINDS.has(kind)) {
const update = obj.params?.update || obj.params || {};
applyAcpToolUpdate(buffers, update, kind);
return;
}
if (!text) return;
if (MESSAGE_CHUNK_KINDS.has(kind)) {
buffers.message.push(text);
// Real-time fan-out: fire the chunk as it arrives so callers can stream it
// verbatim. Buffers are still filled, so getText() remains the source of
// truth for non-streaming callers. onChunk is optional — when absent the
// behaviour is identical to before (collect-then-return).
if (onChunk) { try { onChunk({ kind: 'message', text }); } catch { /* never let a consumer error kill the pump */ } }
} else if (THOUGHT_CHUNK_KINDS.has(kind)) {
buffers.thought.push(text);
if (onChunk) { try { onChunk({ kind: 'thought', text }); } catch { /* ignore consumer error */ } }
}
// Remaining update kinds (plan, mode, commands, status) are not part of the
// text/reasoning/tool-call split and are ignored here on purpose.
}
function makeAcpClient({ command, args, env, signal, timeoutMs, outputLimit, onChunk }) {
const child = _spawn(command, args, {
env,
windowsHide: true,
stdio: ['pipe', 'pipe', 'pipe'],
});
let nextId = 1;
let stderr = '';
let outputBytes = 0;
let closed = false;
let closeCode = null;
let fatalError = null;
let stdoutBuffer = '';
// Active session id, set after session/new. Held here so an abort can send a
// graceful `session/cancel` notification (spec §6.1) before SIGTERM.
let activeSessionId = null;
// True once onAbort has sent session/cancel and scheduled the deferred kill.
// close() (called from runDevinAcpProcess's finally the instant the prompt
// rejects) must NOT pre-empt that grace window with an immediate SIGTERM —
// otherwise the child is killed before it can read the cancel notification.
let gracefulKillPending = false;
const pending = new Map();
const buffers = { message: [], thought: [], toolCalls: new Map() };
const cleanup = () => {
for (const { timer } of pending.values()) clearTimeout(timer);
pending.clear();
};
const failAll = (err) => {
fatalError = fatalError || err;
for (const { reject, timer } of pending.values()) {
clearTimeout(timer);
reject(err);
}
pending.clear();
};
const onData = (name, chunk) => {
const s = chunk.toString('utf8');
outputBytes += Buffer.byteLength(s);
if (outputBytes > outputLimit) {
const err = Object.assign(new Error(`Devin ACP output exceeded ${outputLimit} bytes`), {
status: 502,
type: 'backend_output_too_large',
});
failAll(err);
child.kill('SIGTERM');
return;
}
if (name === 'stderr') {
stderr += s;
if (stderr.length > 8192) stderr = stderr.slice(-8192);
return;
}
stdoutBuffer += s;
const lines = stdoutBuffer.split(/\r?\n/);
stdoutBuffer = lines.pop() || '';
for (const line of lines) {
if (!line.trim()) continue;
const obj = parseJsonLine(line);
if (!obj) continue;
collectAcpTextFromNotification(obj, buffers, onChunk);
if (obj.method === 'session/request_permission' && obj.id != null) {
writeJsonLine(child, {
jsonrpc: '2.0',
id: obj.id,
result: { outcome: 'cancelled' },
});
continue;
}
if (obj.id == null) continue;
const waiter = pending.get(obj.id);
if (!waiter) continue;
pending.delete(obj.id);
clearTimeout(waiter.timer);
waiter.resolve(obj);
}
};
child.stdout.on('data', c => onData('stdout', c));
child.stderr.on('data', c => onData('stderr', c));
// stdin 'error' guard (audit finding): every writeJsonLine already checks
// child.stdin?.writable, but that check races the child dying mid-write — if
// the CLI exits between the guard and the write, the write emits EPIPE on
// stdin as an async 'error' event. With no listener that becomes an
// uncaughtException and takes down the WHOLE proxy (every tenant), not just
// this request. Attach a swallow-and-log handler so a dead child only fails
// its own in-flight requests via the 'close'/'error' paths above. Wrapped in
// try/catch so the handler itself can never throw and re-trigger the crash.
if (child.stdin) {
child.stdin.on('error', err => {
try {
// EPIPE/ECONNRESET here just mean the child went away mid-write; the
// 'close' handler will fail all pending requests with a proper status.
// Log at debug for the expected teardown races, warn for anything else.
if (err?.code === 'EPIPE' || err?.code === 'ECONNRESET') {
log.debug(`Devin ACP stdin closed mid-write (${err.code}); child likely exited`);
} else {
log.warn(`Devin ACP stdin error: ${err?.code || err?.message || err}`);
}
} catch { /* a logging failure must never escalate into a crash */ }
});
}
child.on('error', err => {
if (err.code === 'ENOENT') {
failAll(Object.assign(new Error(`Devin CLI not found: ${command}`), {
status: 503,
type: 'backend_unavailable',
}));
} else {
failAll(Object.assign(err, { status: 502, type: 'backend_error' }));
}
});
child.on('close', code => {
closed = true;
closeCode = code;
const err = fatalError || Object.assign(new Error(`Devin ACP exited with code ${code}`), {
status: 502,
type: 'backend_error',
});
if (code !== 0) {
failAll(err);
} else if (pending.size > 0) {
// GAP-ACP-01: a clean exit (code 0) that still has in-flight requests
// means the CLI closed before answering — e.g. it exited without
// emitting the session/prompt result. cleanup() alone would clear the
// pending map WITHOUT settling those promises, hanging the awaiting
// caller forever and permanently leaking its concurrency slot (fatal
// under DEVIN_MAX_PROCS=1). Reject them so the slot releases.
failAll(Object.assign(new Error('Devin ACP exited (code 0) before responding'), {
status: 502,
type: 'backend_error',
}));
} else {
cleanup();
}
});
const onAbort = () => {
// Graceful cancel: tell the agent to abort in-flight work for the session
// (spec §6.1 `session/cancel` is a notification — no id, no response) so it
// can clean up before we hard-kill the process. We give it a short grace
// window (DEVIN_ACP_CANCEL_GRACE_MS, default 250ms, 0 disables) for the
// notification to be read and acted on, THEN SIGTERM. The grace timer is
// unref'd so it never holds the event loop open, and the close handler
// clears it if the child exits cleanly on its own first. Best-effort: if
// stdin is already gone the write is a no-op and we SIGTERM immediately.
// The reject below uses request_aborted (499) which classifyAcpError leaves
// alone — an abort is a CLIENT action, never an upstream fault, so it must
// not touch pool health.
const err = Object.assign(new Error('Request aborted'), { status: 499, type: 'request_aborted' });
failAll(err);
const graceMs = intEnv('DEVIN_ACP_CANCEL_GRACE_MS', 250, 0);
if (activeSessionId && child.stdin?.writable && graceMs > 0) {
gracefulKillPending = true;
writeJsonLine(child, { jsonrpc: '2.0', method: 'session/cancel', params: { sessionId: activeSessionId } });
const killTimer = setTimeout(() => { if (!closed) child.kill('SIGTERM'); }, graceMs);
killTimer.unref?.();
} else {
child.kill('SIGTERM');
}
};
if (signal) signal.addEventListener('abort', onAbort, { once: true });
const request = (method, params = {}, timeout = timeoutMs) => {
if (closed) {
return Promise.reject(Object.assign(new Error(`Devin ACP is closed (code ${closeCode})`), {
status: 502,
type: 'backend_error',
}));
}
const id = nextId++;
const payload = { jsonrpc: '2.0', id, method, params };
return new Promise((resolve, reject) => {
const timer = setTimeout(() => {
pending.delete(id);
reject(Object.assign(new Error(`Devin ACP ${method} timed out after ${timeout}ms`), {
status: 504,
type: 'backend_timeout',
}));
}, timeout);
timer.unref?.();
pending.set(id, { resolve, reject, timer });
writeJsonLine(child, payload);
});
};
const close = () => {
if (signal) signal.removeEventListener('abort', onAbort);
cleanup();
// Don't pre-empt a graceful cancel-then-kill already scheduled by onAbort.
if (!closed && !gracefulKillPending) child.kill('SIGTERM');
};
return {
request,
close,
// Register the live session id so onAbort can emit `session/cancel` for it.
setSessionId: (id) => { activeSessionId = id || null; },
getText: () => buffers.message.join('').trim(),
getReasoning: () => buffers.thought.join('').trim(),
// Protocol-neutral view of the turn's tool calls. `input` is the parsed
// rawInput ({} when the agent announced none); `name` is null when ACP gave
// neither a title nor a kind, which the caller must treat as unmappable.
getToolCalls: () => Array.from(buffers.toolCalls.values(), (call) => ({
id: call.id,
name: sanitizeToolName(call.title) || sanitizeToolName(call.kind) || null,
kind: call.kind || null,
title: call.title || null,
status: call.status || null,
input: normalizeAcpToolInput(call.hasRawInput ? call.rawInput : null),
})),
getStderr: () => stderr.trim(),
};
}
// Build the ACP `session/prompt` content-block array. VISION (gated by
// DEVIN_ACP_VISION): the ACP path is the CLEAN vision route — the real devin CLI
// builds the downstream GetChatMessage wire and the server signs its own thinking
// turns, so we NEVER forge the un-forgeable #12 signature that blocks the
// DEVIN_CONNECT synthetic-tool_result path. Devin's own `initialize` advertises
// promptCapabilities.image=true, and the ACP ContentBlock union has a first-class
// image variant { type:'image', data:<base64>, mimeType }.
//
// `prompt` may be a plain string (text-only, unchanged behavior) or a structured
// { text, images:[{base64_data, mime_type}] } object. Images are only emitted
// when DEVIN_ACP_VISION is on; otherwise they are dropped and we degrade to text
// (identical to prior behavior — no silent behavior change when the gate is off).
export function acpVisionEnabled(env = process.env) {
const raw = String(env.DEVIN_ACP_VISION ?? '').trim().toLowerCase();
return raw === '1' || raw === 'on' || raw === 'true';
}
export function buildAcpPromptBlocks(prompt, modelHint = '', env = process.env) {
// String prompt → text-only block (back-compat).
if (typeof prompt === 'string' || prompt == null) {
return [{ type: 'text', text: `${modelHint}${prompt ?? ''}` }];
}
const text = typeof prompt.text === 'string' ? prompt.text : '';
const images = Array.isArray(prompt.images) ? prompt.images : [];
const blocks = [{ type: 'text', text: `${modelHint}${text}` }];
if (acpVisionEnabled(env)) {
for (const img of images) {
const data = img?.base64_data || img?.data;
if (!data) continue;
// ACP ContentBlock image variant (route A, inline base64). mimeType is the
// ACP field name (camelCase), distinct from our internal mime_type.
blocks.push({ type: 'image', data, mimeType: img.mime_type || img.mimeType || 'image/png' });
}
}
return blocks;
}
// One bounded retry for TRANSPORT/5xx faults. The whole ACP turn is the retry
// unit: the child process, its handshake and the prompt are re-created, because
// a dead `devin acp` child cannot be resumed. Deliberately excluded:
// - tool-mapping failures (backend_tool_call_unmapped): deterministic given
// the same turn; replaying burns a session for the same answer;
// - UPSTREAM_INTERNAL: measured persistent (3/3), same-token replay just
// amplifies load — mirrors devin-connect.js isRetryable();
// - auth / rate-limit / abort / environment faults (missing CLI, oversized
// output): retrying cannot change the verdict.
const ACP_TRANSPORT_CODES = new Set([
'ECONNRESET', 'ETIMEDOUT', 'ECONNREFUSED', 'EPIPE', 'EAI_AGAIN',
'ENOTFOUND', 'EHOSTUNREACH', 'ENETUNREACH',
]);
function isAcpRetryable(err) {
if (!err) return false;
if (err.status === 499 || err.type === 'request_aborted') return false;
if (err.type === 'backend_tool_call_unmapped') return false;
if (err.type === 'backend_unavailable' || err.type === 'backend_output_too_large'
|| err.type === 'backend_mode_unsupported') return false;
if (err.type === 'upstream_internal' || err.code === 'UPSTREAM_INTERNAL') return false;
if (err.type === 'unauthorized' || err.code === 'UNAUTHORIZED') return false;
if (err.code === 'RATE_LIMITED') return false;
// Explicit capacity ("high demand") is classified retryable by the ACP
// classifier; one replay is exactly the mitigation it asks for.
if (err.retryable === true) return true;
if (err.type === 'backend_timeout' || err.status === 504) return true;
if (err.code && ACP_TRANSPORT_CODES.has(err.code)) return true;
// A raw RPC 5xx / dead child (502 backend_error) is transport-shaped.
if (typeof err.status === 'number' && err.status >= 500 && err.status !== 501) return true;
return false;
}
function acpRetryBaseMs() {
const raw = Number(process.env.DEVIN_ACP_RETRY_BASE_MS);
return Number.isFinite(raw) && raw >= 0 ? raw : 250;
}
export async function runDevinAcpProcess(prompt, { modelKey = '', apiKey = '', apiServerUrl = '', signal = null, onChunk = null } = {}) {
if (!apiKey) {
throw Object.assign(new Error('Devin ACP mode requires an upstream Windsurf account apiKey.'), {
status: 503,
type: 'backend_unavailable',
});
}
const command = process.env.DEVIN_CLI_PATH || 'devin';
// Denylist env: the CLI gets the operator's general environment (PATH, HOME,
// proxy vars, DEVIN_* config like DEVIN_PERMISSION_MODE) but no credentials —
// the upstream apiKey travels ONLY through the ACP `authenticate` RPC's
// _meta.api_key, never env. keep: SSH agent vars so the agent's local tools
// can still reach git-over-ssh; nothing else secret-shaped survives.
const env = buildSubprocessEnv(process.env, {
keep: ['SSH_AUTH_SOCK', 'SSH_AGENT_PID'],
});
const args = parseAcpArgs();
// One channel only: `devin acp --model` (do not also set DEVIN_MODEL — the
// two can disagree). Leave DEVIN_PERMISSION_MODE to the operator; do not
// default it to auto.
if (modelKey && !args.includes('--model')) {
args.push('--model', modelKey);
}
// A replay after ANY chunk crossed onChunk would duplicate the visible answer.
let emitted = false;
const trackedChunk = onChunk
? (chunk) => { emitted = true; onChunk(chunk); }
: null;
const maxRetries = 1;
for (let attempt = 0; ; attempt++) {
try {
return await runDevinAcpAttempt(prompt, {
command, args, env, modelKey, apiKey, apiServerUrl, signal, onChunk: trackedChunk,
});
} catch (err) {
if (attempt >= maxRetries || emitted || signal?.aborted || !isAcpRetryable(err)) throw err;
const backoff = acpRetryBaseMs();
log.warn(`Devin ACP: retryable ${err.code || err.type || err.status} (${err.message}); retry ${attempt + 1}/${maxRetries} in ${backoff}ms`);
if (backoff) await new Promise((r) => setTimeout(r, backoff));
// The caller may have gone away while we waited; a fresh child would
// outlive its request (an already-aborted signal never fires onAbort).
if (signal?.aborted) throw err;
}
}
}
async function runDevinAcpAttempt(prompt, { command, args, env, modelKey, apiKey, apiServerUrl, signal, onChunk }) {
const client = makeAcpClient({
command,
args,
env,
signal,
timeoutMs: runTimeoutMs(),
outputLimit: outputLimitBytes(),
onChunk,
});
try {
const init = await client.request('initialize', {
protocolVersion: 1,
clientCapabilities: {
fs: { readTextFile: false, writeTextFile: false },
terminal: false,
},
clientInfo: { name: process.env.DEVIN_ACP_CLIENT_NAME || 'WindsurfAPI', version: VERSION },
}, 30_000);
if (init.error) throw errorFromRpcResponse(init, 'Devin ACP initialize failed');
const authMeta = {
api_key: apiKey,
...(apiServerUrl ? { api_server_url: apiServerUrl } : {}),
};
const auth = await client.request('authenticate', {
methodId: 'windsurf-api-key',
_meta: authMeta,
}, 45_000);
if (auth.error) throw errorFromRpcResponse(auth, 'Devin ACP authenticate failed');
const session = await client.request('session/new', {
cwd: process.env.DEVIN_CLI_WORKDIR || process.cwd(),
mcpServers: [],
}, 60_000);
if (session.error) throw errorFromRpcResponse(session, 'Devin ACP session/new failed');
const sessionId = session?.result?.sessionId || session?.result?.session_id;
if (!sessionId) {
throw Object.assign(new Error('Devin ACP session/new did not return sessionId'), {
status: 502,
type: 'backend_error',
});
}
// Arm graceful cancellation: a later abort can now send session/cancel.
client.setSessionId(sessionId);
// Optional native config: the CLI advertises configOptions on session/new.
// Flag-off by default because the live option ids/values are not calibrated
// against a real ACP peer yet; the prompt hint remains the fallback.
if (modelKey && process.env.DEVIN_ACP_CONFIG_OPTIONS === '1') {
const rawOptions = session?.result?.configOptions || session?.result?.config_options || [];
const modelOption = Array.isArray(rawOptions)
? rawOptions.find((o) => {
const id = String(o?.id || o?.configId || o?.key || o?.name || '').toLowerCase();
return id === 'model' || id.includes('model');
})
: null;
const configId = modelOption?.id || modelOption?.configId || modelOption?.key || modelOption?.name;
if (configId) {
const choices = Array.isArray(modelOption.options) ? modelOption.options : [];
const chosen = choices.find((c) => {
const v = c?.value ?? c?.id ?? c?.name;
return v === modelKey;
});
const value = chosen ? (chosen.value ?? chosen.id ?? chosen.name) : modelKey;
const cfg = await client.request('session/set_config_option', {
sessionId,
configId,
value,
}, 15_000);
if (cfg.error) {
log.warn(`Devin ACP set_config_option ${configId}=${value} failed: ${cfg.error.message || cfg.error.code || 'unknown'}`);
}
}
}
const modelHint = modelKey ? `Model requested by caller: ${modelKey}\n\n` : '';
const result = await client.request('session/prompt', {
sessionId,
prompt: buildAcpPromptBlocks(prompt, modelHint, env),
}, runTimeoutMs());
if (result.error) throw errorFromRpcResponse(result, 'Devin ACP session/prompt failed');
const stopReason = result?.result?.stopReason || result?.result?.stop_reason || null;
const toolCalls = client.getToolCalls();
// A tool-ish stop is only representable as tool_calls/tool_use when the ACP
// pump actually captured the call. Degrading to finish_reason:"stop" here
// would tell the client a turn that stopped to use a tool was a normal
// answer (GFHarness probe D). No call (or a call without the title/kind
// needed for a protocol tool name) is an unmappable turn: fail explicitly.
if (/tool/i.test(String(stopReason || ''))) {
const unnamed = toolCalls.filter(tc => !tc.name);
if (!toolCalls.length || unnamed.length) {
throw Object.assign(new Error(
`Devin ACP stopped with stopReason "${stopReason}" but ` +
(toolCalls.length
? `${unnamed.length} tool call(s) had no title/kind to map to a tool name`
: 'no tool_call update was received') +
'; refusing to report a clean completion',
), { status: 502, type: 'backend_tool_call_unmapped' });
}
}
return {
text: client.getText(),
reasoning: client.getReasoning(),
stderr: client.getStderr(),
usage: result?.result?.usage || null,
stopReason,
// Only named calls can become protocol tool-call events. A non-tool stop
// may carry metadata-only calls; they are dropped with a trace.
toolCalls: toolCalls.filter(tc => {
if (tc.name) return true;
log.warn(`Devin ACP tool_call ${String(tc.id)} has no title/kind; not forwarded`);
return false;
}),
};
} finally {
client.close();
}
}