src/special-agent.js
Rama master · réplica de lectura del repositorio original.
/**
* Optional special-agent backend for models that do not work through the
* direct Cascade chat path (SWE/adaptive/arena).
*
* The default backend mode is a conservative Devin CLI print-mode adapter. It
* is disabled by default and intentionally does not execute or emulate caller
* tools. DEVIN_CLI_MODE=acp is available as an experimental stdio backend; it
* is kept small and separate in devin-acp.js so this routing module does not
* become the protocol implementation.
*/
import { spawn } from 'child_process';
import { randomUUID } from 'crypto';
import {
getApiKey,
releaseAccount,
reportError,
reportSuccess,
reportInternalError,
markRateLimited,
reportBanSignal,
looksLikeBanSignal,
refundReservation,
} from './auth.js';
import { config, log } from './config.js';
import { recordRequest } from './dashboard/stats.js';
import { sanitizeText, PathSanitizeStream } from './sanitize.js';
// The CLI child receives the pool credential in its env (:428-431) and its stderr
// is folded into the message the CALLER sees (:504, and :692/:925 hand that message
// to the client), so a CLI that prints its own token on an auth failure would leak
// it. sanitizeText is a path/XML redactor and does not know any credential shape —
// measured 2026-09-13: auth1_/devin-session-token$/JWT all passed through it.
import { redactCredentialFragments, buildSubprocessEnv } from './log-safety.js';
import { runDevinAcpProcess, probeDevinCliAvailable, acpVisionEnabled } from './devin-acp.js';
import { extractInlineImages as extractImagesFromContent } from './devin-connect.js';
import { systemFingerprint } from './system-fingerprint.js';
import { getBackendSwitch, strictUsageTotal } from './runtime-config.js';
import { toClientToolCallId } from './tool-ids.js';
const SPECIAL_BACKEND = 'special_agent';
const DEFAULT_SPECIAL_MODELS = new Set([
'swe-1.6',
'swe-1.6-fast',
'adaptive',
'arena-fast',
'arena-smart',
]);
function intEnv(name, fallback, min = 0) {
const n = parseInt(process.env[name] || '', 10);
return Number.isFinite(n) && n >= min ? n : fallback;
}
function isEnabled() {
// DEVIN_ONLY (Cascade retired) implies the special-agent backend is on — it's
// the only backend left, so a single switch must enable the whole path.
if (getBackendSwitch('devinOnly')) return true;
const backend = String(process.env.WINDSURFAPI_SPECIAL_AGENT_BACKEND || '').trim().toLowerCase();
return backend === 'devin-cli' || process.env.DEVIN_CLI_ENABLED === '1';
}
// Opt-in passthrough of the ACP thought stream (agent_thought_chunk) as
// OpenAI-style reasoning_content. Default OFF preserves the prior behaviour:
// result.reasoning is captured by the runner but discarded by the handler.
function exposeReasoning() {
return process.env.DEVIN_ACP_EXPOSE_REASONING === '1';
}
export function isSpecialAgentEnabled() {
return isEnabled();
}
function configuredBackend() {
if (!isEnabled()) return '';
return 'devin-cli';
}
function maxProcs() {
return intEnv('DEVIN_MAX_PROCS', 1, 1);
}
function queueTimeoutMs() {
return intEnv('DEVIN_QUEUE_TIMEOUT_MS', 30_000, 1000);
}
function runTimeoutMs() {
return intEnv('DEVIN_TIMEOUT_MS', 10 * 60_000, 1000);
}
function outputLimitBytes() {
return intEnv('DEVIN_OUTPUT_LIMIT_BYTES', 4 * 1024 * 1024, 1024);
}
// SSE keepalive interval for live ACP streams. Devin "thinking" can run for
// tens of seconds with no chunk; without periodic bytes, clients and
// intermediaries (nginx, CDNs) close the idle connection. Mirrors the 15s
// heartbeat on the Cascade path but uses a comment frame so OpenAI/Anthropic
// SSE parsers ignore it. Read at call time so tests can tune it.
function acpPingMs() {
return intEnv('DEVIN_ACP_PING_MS', 25_000, 10);
}
export function isSpecialAgentModelInfo(info) {
return info?.backend === SPECIAL_BACKEND;
}
export function isSpecialAgentModelKey(modelKey, info = null) {
if (info && isSpecialAgentModelInfo(info)) return true;
return DEFAULT_SPECIAL_MODELS.has(String(modelKey || ''));
}
function textFromContent(content) {
if (typeof content === 'string') return content;
if (!Array.isArray(content)) return '';
return content
.map(part => {
if (!part || typeof part !== 'object') return '';
if (typeof part.text === 'string') return part.text;
if (part.type === 'text' && typeof part.content === 'string') return part.content;
if (part.type === 'image_url' || part.type === 'input_image') return '[image omitted: print backend does not support media]';
return '';
})
.filter(Boolean)
.join('\n');
}
/**
* Render chat messages into a single prompt for the special-agent backend.
*
* The backend takes ONE text prompt per session/prompt call, so a multi-turn
* chat has to be flattened. Two shapes:
* - Single turn (one user message, no prior assistant/tool turns): send the
* user's text as-is (optionally prefixed with system instructions). No role
* labels — it's just a prompt, and labels would be noise.
* - Multi-turn: render the prior turns as a labeled transcript, then call out
* the latest message and instruct the agent to continue AS THE ASSISTANT.
* Without this framing the agent tends to summarize/react to the whole
* transcript instead of answering the last turn.
*
* System messages are pulled out of the turn flow and surfaced as explicit
* instructions, mirroring how real chat APIs treat the system role.
*/
export function buildSpecialAgentPrompt(messages) {
const list = Array.isArray(messages) ? messages : [];
const systemParts = [];
const turns = [];
for (const m of list) {
const role = m?.role || 'user';
const text = textFromContent(m?.content).trim();
if (!text) continue;
if (role === 'system') { systemParts.push(text); continue; }
turns.push({ role, text });
}
const systemBlock = systemParts.length ? `System instructions:\n${systemParts.join('\n\n')}` : '';
const label = r => (r === 'assistant' ? 'Assistant' : r === 'tool' ? 'Tool result' : 'User');
const priorTurns = turns.slice(0, -1);
const last = turns[turns.length - 1];
// Single turn (no prior conversation): just the user's text + optional system.
if (priorTurns.length === 0) {
return [systemBlock, last ? last.text : ''].filter(Boolean).join('\n\n').trim();
}
// Multi-turn: labeled history, highlighted latest message, continuation cue.
const history = priorTurns.map(t => `${label(t.role)}:\n${t.text}`).join('\n\n');
const instruction = 'Continue this conversation. Reply as the assistant to the latest message, using the prior turns as context.';
return [
systemBlock,
'Conversation so far:',
history,
`Latest ${label(last.role)} message:\n${last.text}`,
instruction,
].filter(Boolean).join('\n\n').trim();
}
// Collect inline images across ALL message contents (devin-connect's
// extractInlineImages works per-content; this walks the whole conversation).
// Returns [{ base64_data, mime_type }] in message order — the ACP runner turns
// them into image content blocks. Kiro-style hygiene: cap total images so a
// pathological history can't blow the request body.
const ACP_MAX_IMAGES = 20;
function extractInlineImages(messages) {
const out = [];
for (const m of Array.isArray(messages) ? messages : []) {
if (!Array.isArray(m?.content)) continue;
for (const img of extractImagesFromContent(m.content)) {
out.push(img);
if (out.length >= ACP_MAX_IMAGES) return out;
}
}
return out;
}
function hasUnsupportedMedia(messages) {
for (const m of Array.isArray(messages) ? messages : []) {
if (!Array.isArray(m?.content)) continue;
for (const part of m.content) {
if (part?.type && part.type !== 'text') return true;
}
}
return false;
}
function estimateTokens(messages, completionText) {
const promptChars = (Array.isArray(messages) ? messages : [])
.reduce((n, m) => n + textFromContent(m?.content).length, 0);
const prompt = Math.max(1, Math.ceil(promptChars / 4));
const completion = Math.max(1, Math.ceil(String(completionText || '').length / 4));
return {
prompt_tokens: prompt,
completion_tokens: completion,
total_tokens: prompt + completion,
input_tokens: prompt,
output_tokens: completion,
prompt_tokens_details: { cached_tokens: 0 },
completion_tokens_details: { reasoning_tokens: 0 },
};
}
// H3: prefer the runner's real ACP usage over chars/4 estimation. ACP usage
// shapes vary (camelCase from JSON-RPC: inputTokens/outputTokens/totalTokens,
// or the snake_case Connect shape); accept either and fall back to estimation
// for any field the runner did not report. Estimation stays the floor so the
// stream is never billed at zero when the runner omits usage entirely.
function pickUsage(rawUsage, messages, completionText) {
const est = estimateTokens(messages, completionText);
if (!rawUsage || typeof rawUsage !== 'object') return est;
const num = (...vals) => {
for (const v of vals) {
const n = Number(v);
if (Number.isFinite(n) && n >= 0) return n;
}
return null;
};
const input = num(rawUsage.input_tokens, rawUsage.inputTokens, rawUsage.prompt_tokens, rawUsage.promptTokens);
const output = num(rawUsage.output_tokens, rawUsage.outputTokens, rawUsage.completion_tokens, rawUsage.completionTokens);
const total = num(rawUsage.total_tokens, rawUsage.totalTokens);
const cached = num(rawUsage.cache_read_tokens, rawUsage.cacheReadTokens, rawUsage.cached_tokens, rawUsage.cachedTokens);
const reasoning = num(rawUsage.reasoning_tokens, rawUsage.reasoningTokens);
const prompt = input ?? est.prompt_tokens;
const completion = output ?? est.completion_tokens;
return {
prompt_tokens: prompt,
completion_tokens: completion,
// WINDSURFAPI_STRICT_USAGE_TOTAL=1 means the client validates OpenAI's arithmetic
// identity, so honour it here too rather than passing the runner's total through.
//
// This front was missed when the flag shipped: it forwards whatever the ACP runner
// reported, so with the flag ON a Devin CLI / ACP model could still emit
// total != prompt + completion and the flag would be two-thirds true. A flag that
// holds on some protocol fronts and not others is worse than no flag, because the
// client cannot tell which response it is looking at. Sixth occurrence of the
// partial-path trap in this repo.
//
// No billing consequence either way: this path records via recordRequest (dashboard
// counters), never recordAccountSpend, so no per-account spend tally reads this number.
total_tokens: strictUsageTotal() ? (prompt + completion) : (total ?? (prompt + completion)),
input_tokens: prompt,
output_tokens: completion,
prompt_tokens_details: { cached_tokens: cached ?? 0 },
completion_tokens_details: { reasoning_tokens: reasoning ?? 0 },
};
}
// H7: map the ACP/Connect stop_reason onto an OpenAI finish_reason instead of
// hardcoding 'stop' (which made truncation / tool-call / refusal all look like
// a clean completion). Unknown reasons fall back to 'stop' so well-behaved
// completions are unaffected.
//
// cline-01: never report finish_reason:'tool_calls' unless tool_calls were
// actually emitted. Cline's @ai-sdk client treats a 'tool_calls' finish with no
// tool_calls array as a broken response and retries to death (cline#9622). The
// print runner is TEXT-ONLY, so a tool-ish stop_reason from it must degrade to
// 'stop', not lie about a tool call that was never sent. The ACP runner now
// forwards captured tool_call/tool_call_update notifications and refuses a
// tool-ish stop with no mappable call (backend_tool_call_unmapped), so its
// call sites pass hasToolCalls=true exactly when tool_calls are present.
// The content_filter test is deliberately NARROW. A bare `includes('content')`
// classified any reason with the substring "content" as a refusal, and the words
// upstreams actually use for a NORMAL completion contain it: `content_complete`,
// `no_content`, `contents_delivered`, `end_turn_content` all resolved to
// content_filter. That is not a cosmetic mislabel — content_filter closes a
// /v1/responses turn as `response.incomplete` and surfaces on /v1/messages as
// Anthropic `stop_reason:'refusal'`, so a completed answer was reported to the
// client as the model refusing to answer. A real refusal is visible in the reply
// text; a fabricated one is not something the client can recover from.
//
// So the filter vocabulary is matched on whole tokens (split on non-letters), not
// as substrings. `refusal` / `content_filter` / `content_policy` / `safety` still
// classify; `content_complete` no longer does.
const FILTER_TOKENS = new Set(['filter', 'filtered', 'refusal', 'refused', 'refuse',
'safety', 'unsafe', 'blocked', 'policy', 'moderation', 'moderated', 'censored']);
function mapFinishReason(stopReason, hasToolCalls = false) {
const r = String(stopReason || '').toLowerCase();
if (!r) return 'stop';
if (r.includes('max_tokens') || r.includes('max_token') || r.includes('length') || r.includes('truncat')) return 'length';
if (r.includes('tool')) return hasToolCalls ? 'tool_calls' : 'stop';
const tokens = r.split(/[^a-z]+/).filter(Boolean);
if (tokens.some(t => FILTER_TOKENS.has(t))) return 'content_filter';
return 'stop';
}
export { mapFinishReason };
function errorResponse(status, type, message, extra = {}) {
return {
status,
body: {
error: {
message,
type,
...extra,
},
},
};
}
function parseArgsTemplate(prompt, modelKey) {
const raw = process.env.DEVIN_CLI_ARGS_JSON || '';
let args;
if (raw.trim()) {
try {
args = JSON.parse(raw);
} catch (err) {
throw Object.assign(new Error(`Invalid DEVIN_CLI_ARGS_JSON: ${err.message}`), {
status: 500,
type: 'backend_misconfigured',
});
}
if (!Array.isArray(args) || !args.every(x => typeof x === 'string')) {
throw Object.assign(new Error('DEVIN_CLI_ARGS_JSON must be a JSON string array'), {
status: 500,
type: 'backend_misconfigured',
});
}
} else {
args = ['-p', '{prompt}'];
}
return args.map(arg => arg
.replaceAll('{prompt}', prompt)
.replaceAll('{model}', modelKey || ''));
}
let activeProcs = 0;
const waiters = [];
function acquireSlot(signal) {
if (activeProcs < maxProcs()) {
activeProcs++;
return Promise.resolve(() => releaseSlot());
}
return new Promise((resolve, reject) => {
const started = Date.now();
const waiter = {
resolve: () => {
clearTimeout(timer);
activeProcs++;
resolve(() => releaseSlot());
},
reject,
};
const timer = setTimeout(() => {
const idx = waiters.indexOf(waiter);
if (idx !== -1) waiters.splice(idx, 1);
reject(Object.assign(new Error(`Devin CLI process pool is full after ${Date.now() - started}ms`), {
status: 503,
type: 'backend_pool_exhausted',
}));
}, queueTimeoutMs());
timer.unref?.();
if (signal) {
signal.addEventListener('abort', () => {
clearTimeout(timer);
const idx = waiters.indexOf(waiter);
if (idx !== -1) waiters.splice(idx, 1);
reject(Object.assign(new Error('Request aborted while waiting for Devin CLI slot'), {
status: 499,
type: 'request_aborted',
}));
}, { once: true });
}
waiters.push(waiter);
});
}
function releaseSlot() {
activeProcs = Math.max(0, activeProcs - 1);
const next = waiters.shift();
if (next) next.resolve();
}
export function getSpecialAgentStatus() {
return {
backend: configuredBackend() || 'disabled',
enabled: isEnabled(),
activeProcs,
queued: waiters.length,
maxProcs: maxProcs(),
queueTimeoutMs: queueTimeoutMs(),
runTimeoutMs: runTimeoutMs(),
outputLimitBytes: outputLimitBytes(),
mode: getBackendSwitch('devinCliMode'),
};
}
export async function runDevinAcp(prompt, { modelKey = '', apiKey = '', apiServerUrl = '', signal = null, onChunk = null } = {}) {
const release = await acquireSlot(signal);
try {
// GAP-ACP-05: forward onChunk so the live streamer's incremental chunks
// actually reach the ACP pump. Without this the wrapper silently drops it
// and the "real-time" stream degrades to a single buffered flush at the end.
return await runDevinAcpProcess(prompt, { modelKey, apiKey, apiServerUrl, signal, onChunk });
} finally {
release();
}
}
export async function runDevinPrint(prompt, { modelKey = '', apiKey = '', signal = null } = {}) {
const release = await acquireSlot(signal);
try {
const mode = getBackendSwitch('devinCliMode');
if (mode !== 'print') {
throw Object.assign(new Error(`DEVIN_CLI_MODE=${mode} is not implemented yet; use print for the first PoC`), {
status: 501,
type: 'backend_mode_unsupported',
});
}
const command = process.env.DEVIN_CLI_PATH || 'devin';
const args = parseArgsTemplate(prompt, modelKey);
const envName = process.env.DEVIN_CLI_API_KEY_ENV || 'WINDSURF_API_KEY';
// Denylist env (L3b env-leak fix): the CLI needs general env (PATH, HOME,
// proxy vars, operator DEVIN_* config) but must not inherit the proxy's
// credentials. keep: envName — the ONE deliberate credential channel:
// when a pool key exists it is written below (overwriting any ambient
// value); when none exists (or DEVIN_CLI_USE_ACCOUNT_POOL=0) an ambient
// value under that same name still passes through, matching the pre-fix
// reachability without leaking anything else. SSH_AUTH_SOCK/SSH_AGENT_PID
// stay so the agent's local tools keep git-over-ssh.
const env = buildSubprocessEnv(process.env, {
keep: [envName, 'SSH_AUTH_SOCK', 'SSH_AGENT_PID'],
});
if (apiKey && process.env.DEVIN_CLI_USE_ACCOUNT_POOL !== '0') {
env[envName] = apiKey;
}
return await new Promise((resolve, reject) => {
const child = spawn(command, args, {
env,
windowsHide: true,
stdio: ['ignore', 'pipe', 'pipe'],
});
let stdout = '';
let stderr = '';
let killedByTimeout = false;
let outputBytes = 0;
let fatalError = null;
const limit = outputLimitBytes();
const timer = setTimeout(() => {
killedByTimeout = true;
child.kill('SIGTERM');
setTimeout(() => {
if (!child.killed) child.kill('SIGKILL');
}, 1500).unref?.();
}, runTimeoutMs());
timer.unref?.();
const onAbort = () => {
fatalError = Object.assign(new Error('Request aborted'), { status: 499, type: 'request_aborted' });
child.kill('SIGTERM');
};
if (signal) signal.addEventListener('abort', onAbort, { once: true });
const collect = (name, chunk) => {
const s = chunk.toString('utf8');
outputBytes += Buffer.byteLength(s);
if (outputBytes > limit) {
fatalError = Object.assign(new Error(`Devin CLI output exceeded ${limit} bytes`), {
status: 502,
type: 'backend_output_too_large',
});
child.kill('SIGTERM');
return;
}
if (name === 'stdout') stdout += s;
else stderr += s;
};
child.stdout.on('data', c => collect('stdout', c));
child.stderr.on('data', c => collect('stderr', c));
child.on('error', err => {
clearTimeout(timer);
if (signal) signal.removeEventListener('abort', onAbort);
if (err.code === 'ENOENT') {
reject(Object.assign(new Error(`Devin CLI not found: ${command}`), {
status: 503,
type: 'backend_unavailable',
}));
} else {
reject(Object.assign(err, { status: 502, type: 'backend_error' }));
}
});
child.on('close', code => {
clearTimeout(timer);
if (signal) signal.removeEventListener('abort', onAbort);
if (killedByTimeout) {
reject(Object.assign(new Error(`Devin CLI timed out after ${runTimeoutMs()}ms`), {
status: 504,
type: 'backend_timeout',
}));
return;
}
if (fatalError) {
reject(fatalError);
return;
}
if (code !== 0) {
const detail = (stderr || stdout || `exit code ${code}`).trim();
// REDACT FIRST, THEN TRUNCATE. Slicing first splits a JWT at its first
// dot and the credential regex no longer matches — the exact trap
// documented in log-safety.js next to sliceRedactedJson.
reject(Object.assign(new Error(`Devin CLI failed: ${redactCredentialFragments(detail).slice(0, 1000)}`), {
status: 502,
type: 'backend_error',
exitCode: code,
}));
return;
}
resolve({ text: stdout.trim(), stderr: stderr.trim() });
});
});
} finally {
release();
}
}
// Project the ACP runner's protocol-neutral tool calls into OpenAI's
// assistant.tool_calls shape. The runner already refused a tool-ish stop that
// has no mappable call, so every entry here has an id and a name. `input` is
// the agent's rawInput; {} means the agent announced no arguments (never a
// synthesized result — the tool's output stays out of the egress).
function openAiToolCalls(toolCalls) {
if (!Array.isArray(toolCalls) || !toolCalls.length) return null;
return toolCalls.map((tc, index) => ({
// ACP's toolCallId is the CLI's own dialect (the raw id still keys the
// tool_call/tool_call_update notifications inside the pump). The client
// only ever sees the OpenAI `call_<opaque>` shape.
id: toClientToolCallId(tc.id, `call_${index}_${Date.now().toString(36)}`),
type: 'function',
function: {
name: tc.name,
arguments: JSON.stringify(tc.input ?? {}),
},
}));
}
function chatCompletionBody({ id, created, model, messages, text, reasoning = '', usage = null, stopReason = null, toolCalls = null }) {
const calls = openAiToolCalls(toolCalls);
const message = { role: 'assistant', content: text || null };
if (reasoning) message.reasoning_content = reasoning;
if (calls) message.tool_calls = calls;
return {
id,
object: 'chat.completion',
created,
model,
system_fingerprint: systemFingerprint(model),
service_tier: 'default',
choices: [{
index: 0,
message,
finish_reason: mapFinishReason(stopReason, !!calls),
}],
usage: pickUsage(usage, messages, text),
};
}
function streamFromText({ id, created, model, messages, text, reasoning = '', usage = null, stopReason = null, includeUsage = false }) {
return {
status: 200,
stream: true,
headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache' },
handler: async (res) => {
const fp = systemFingerprint(model);
const send = data => {
if (data && data.object === 'chat.completion.chunk') {
if (data.system_fingerprint == null) data.system_fingerprint = fp;
if (data.service_tier == null) data.service_tier = 'default';
}
if (!res.writableEnded) res.write(`data: ${JSON.stringify(data)}\n\n`);
};
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: { role: 'assistant', content: '' }, finish_reason: null }] });
if (reasoning) {
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: { reasoning_content: reasoning }, finish_reason: null }] });
}
if (text) {
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: { content: text }, finish_reason: null }] });
}
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: {}, finish_reason: mapFinishReason(stopReason) }] });
// O1: trailing usage frame only when the caller opted in via
// stream_options.include_usage (OpenAI omits it by default).
if (includeUsage) {
send({ id, object: 'chat.completion.chunk', created, model,
choices: [], usage: pickUsage(usage, messages, text) });
}
if (!res.writableEnded) {
res.write('data: [DONE]\n\n');
res.end();
}
},
};
}
/**
* Live SSE stream for ACP mode: the runner is driven INSIDE the handler so each
* agent_message_chunk is forwarded the moment it arrives (real streaming, not
* collect-then-slice). The handler owns the account lifecycle and request
* accounting because it runs after handleSpecialAgentChatCompletion returns.
*
* Deltas are sanitized incrementally with PathSanitizeStream so no sensitive
* literal can leak across a chunk boundary. Reasoning chunks are forwarded as
* reasoning_content only when explicitly opted in.
*/
function streamLiveAcp({ id, created, model, messages, prompt, modelKey, acct, runner, signal, onDone, includeUsage = false }) {
return {
status: 200,
stream: true,
headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache' },
handler: async (res) => {
const fp = systemFingerprint(model);
const send = data => {
if (data && data.object === 'chat.completion.chunk') {
if (data.system_fingerprint == null) data.system_fingerprint = fp;
if (data.service_tier == null) data.service_tier = 'default';
}
if (!res.writableEnded) res.write(`data: ${JSON.stringify(data)}\n\n`);
};
const msgSanitizer = new PathSanitizeStream();
const showReasoning = exposeReasoning();
let emittedText = '';
let emittedReasoning = false;
// SURV-02: own a per-request AbortController bound to client disconnect.
// The route-level signal (server shutdown etc.) is chained in. Without
// this, a client hangup leaves the spawned ACP process running until
// DEVIN_TIMEOUT_MS (10 min default); with DEVIN_MAX_PROCS=1 that wedges
// the whole backend on a single abandoned stream. runDevinAcpProcess
// already kills the child on abort — we just have to fire it.
const abortController = new AbortController();
const onUpstreamAbort = () => abortController.abort();
if (signal) {
if (signal.aborted) abortController.abort();
else signal.addEventListener('abort', onUpstreamAbort, { once: true });
}
// res is a Node http.ServerResponse in production (always has .on); guard
// for minimal test/writable mocks that omit the EventEmitter surface.
if (typeof res.on === 'function') {
res.on('close', () => {
if (!res.writableEnded) abortController.abort();
});
}
// P0-3: SSE keepalive. Devin "thinking" can be silent for tens of
// seconds; without bytes, idle timers (client, nginx, CDN) drop the
// stream. `:` comment frames are ignored by SSE parsers.
const ping = setInterval(() => {
if (!res.writableEnded) res.write(': ping\n\n');
}, acpPingMs());
const stopPing = () => clearInterval(ping);
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: { role: 'assistant', content: '' }, finish_reason: null }] });
const onChunk = ({ kind, text }) => {
if (kind === 'message') {
const safe = msgSanitizer.feed(text);
if (safe) {
emittedText += safe;
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: { content: safe }, finish_reason: null }] });
}
} else if (kind === 'thought' && showReasoning) {
emittedReasoning = true;
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: { reasoning_content: text }, finish_reason: null }] });
}
};
let ok = true;
let runErr = null;
try {
const result = await runner(prompt, { modelKey, apiKey: acct?.apiKey || '', apiServerUrl: acct?.apiServerUrl || '', signal: abortController.signal, onChunk });
const tail = msgSanitizer.flush();
if (tail) {
emittedText += tail;
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: { content: tail }, finish_reason: null }] });
}
// Fallback for runners that return text/reasoning without streaming via
// onChunk (non-chunking runners or test mocks): emit the returned value
// so the stream is never empty when the runner produced output.
if (showReasoning && !emittedReasoning && result?.reasoning) {
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: { reasoning_content: sanitizeText(result.reasoning) }, finish_reason: null }] });
}
if (!emittedText && result?.text) {
const safe = sanitizeText(result.text);
emittedText = safe;
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: { content: safe }, finish_reason: null }] });
}
// Forward the turn's tool calls before the terminal chunk. They are
// emitted as one complete delta per call (id/name/arguments together):
// ACP's tool_call_update can replace rawInput, so buffering to the end
// is the only lossless framing. No tool result is emitted — the client
// never produced one.
const calls = openAiToolCalls(result?.toolCalls);
if (calls) {
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: {
tool_calls: calls.map((tc, index) => ({ index, ...tc })),
}, finish_reason: null }] });
}
// H7: map the ACP stop_reason instead of hardcoding 'stop'.
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: {}, finish_reason: mapFinishReason(result?.stopReason, !!calls) }] });
// H3: bill real ACP usage when the runner reported it; estimate otherwise.
// O1: only forward the usage frame when the caller opted in.
if (includeUsage) {
send({ id, object: 'chat.completion.chunk', created, model,
choices: [], usage: pickUsage(result?.usage, messages, emittedText) });
}
} catch (err) {
ok = false;
runErr = err;
// Headers are already sent (200), so surface the failure as a terminal
// SSE error event rather than an HTTP status the client can no longer
// see. Use a non-stop finish_reason so clients that ignore the bare
// error frame don't mistake a failed run for a clean completion (H2).
send({ error: { type: err?.type || 'backend_error', message: redactCredentialFragments(sanitizeText(err?.message || 'Special-agent stream failed')) } });
send({ id, object: 'chat.completion.chunk', created, model,
choices: [{ index: 0, delta: {}, finish_reason: 'error' }] });
} finally {
stopPing();
if (signal) signal.removeEventListener('abort', onUpstreamAbort);
if (!res.writableEnded) { res.write('data: [DONE]\n\n'); res.end(); }
if (onDone) onDone(ok, runErr);
}
},
};
}
async function checkoutAccount(callerKey, modelKey = null) {
if (process.env.DEVIN_CLI_USE_ACCOUNT_POOL === '0') return null;
// AP-GAP-1: thread the real modelKey so getApiKey applies the same
// entitlement filter the Cascade path gets — without it a free account with
// no SWE/Devin entitlement is selected, then fails late at ACP authenticate,
// burning a process spawn + RPM slot.
return getApiKey([], modelKey, callerKey);
}
/**
* Report the outcome of a special-agent run back to the account pool so the
* pool's health tracking (error streak → status='error', rate-limit windows,
* ban detection) actually works on the Devin path. Without this the pool has
* zero feedback: a banned/rate-limited key keeps getting reselected and failing
* on every request (AP-GAP-2, the top P0 from both the self-audit and the
* kiro.rs study). Mirrors how the Cascade path classifies upstream errors.
*
* Cooldowns use FIXED durations, never an accumulating default — kiro.rs's
* hard-won lesson was that exponential default_duration snowballs (60→90→135s)
* into a self-inflicted outage. markRateLimited already clamps to a fixed
* window per call.
*/
function reportRunFailure(apiKey, err, deps = {}) {
if (!apiKey) return;
const status = Number(err?.status) || 0;
const type = String(err?.type || '');
const message = String(err?.message || '');
const banSignal = deps.reportBanSignal || reportBanSignal;
const rateLimited = deps.markRateLimited || markRateLimited;
const internalError = deps.reportInternalError || reportInternalError;
const genericError = deps.reportError || reportError;
// AC1 gap-c: a 503 backend_unavailable means the Devin CLI binary is missing
// or not executable — an ENVIRONMENT fault, not an account fault. It must NOT
// feed the windowed generic-error streak, or a missing/removed CLI would
// wrongly disable every healthy account it touches (3 hits → status='error').
// The proactive probe normally catches this before checkout, but the passive
// post-spawn ENOENT→503 (probe disabled, print mode, or a TOCTOU race where
// the binary vanishes between probe and spawn) still lands here, so guard it.
if (status === 503 || type === 'backend_unavailable') {
return;
}
// A client abort (499 request_aborted) is also not an upstream/account fault.
if (status === 499 || type === 'request_aborted') {
return;
}
// An ACP turn whose tool_use stop has no forwardable tool call is a
// protocol/CLI-shape fault (the agent contradicted its own notification
// stream), not an account fault. Feeding the error streak would disable a
// healthy credential over a CLI-version mismatch.
if (type === 'backend_tool_call_unmapped') {
return;
}
// Account-level ban/suspension shapes — promote to banned after 2 hits.
if (looksLikeBanSignal(message)) {
banSignal(apiKey, message);
return;
}
// 429 rate limit — fixed-window cooldown, optionally per-model upstream.
if (status === 429) {
rateLimited(apiKey);
return;
}
// 402 quota exhausted — treat as a longer rate-limit so the key rotates out
// without being permanently disabled (it recovers when quota resets).
if (status === 402) {
rateLimited(apiKey, 30 * 60 * 1000);
return;
}
// Upstream "internal error" shapes are account-specific and sticky.
if (status >= 500 && /internal error/i.test(message)) {
internalError(apiKey);
return;
}
// Generic failure — feeds the windowed error streak (3-in-window disables).
genericError(apiKey);
}
export async function handleSpecialAgentChatCompletion(body, route, deps = {}) {
const started = Date.now();
const modelKey = route?.modelKey || body?.model || '';
const model = route?.model || body?.model || modelKey || config.defaultModel;
const messages = Array.isArray(route?.messages) ? route.messages : (body?.messages || []);
const id = route?.id || 'chatcmpl-' + randomUUID().replace(/-/g, '').slice(0, 29);
const created = route?.created || Math.floor(Date.now() / 1000);
const tools = Array.isArray(body?.tools) ? body.tools : [];
// O1: OpenAI streams the trailing usage-only frame only when the caller sets
// stream_options.include_usage:true; otherwise it is omitted.
const includeUsage = body?.stream_options?.include_usage === true;
if (!configuredBackend()) {
return errorResponse(
503,
'backend_unavailable',
`${model} requires the optional special-agent backend. Set WINDSURFAPI_SPECIAL_AGENT_BACKEND=devin-cli and DEVIN_CLI_PATH to test Devin CLI/ACP routing.`,
{ backend: 'special_agent' },
);
}
if (tools.length && !getBackendSwitch('allowClientTools')) {
return errorResponse(
400,
'unsupported_tool_boundary',
'Devin CLI print backend does not safely expose caller-local tools yet. Use a normal Cascade model, or wait for ACP tool bridging.',
{ backend: 'devin-cli', tool_count: tools.length },
);
}
// Vision over ACP (gated by DEVIN_ACP_VISION): the ACP path forwards images as
// first-class content blocks and the real CLI + server produce genuinely
// signed turns — the clean route that works even for extended-thinking models
// (opus-4-8), which the DEVIN_CONNECT synthetic-tool_result path CANNOT serve
// (un-forgeable #12 signature). Verified end-to-end: opus saw a red/blue image.
const acpMode = getBackendSwitch('devinCliMode') === 'acp';
const visionOverAcp = acpMode && acpVisionEnabled() && hasUnsupportedMedia(messages);
if (hasUnsupportedMedia(messages) && !visionOverAcp && process.env.DEVIN_CLI_ALLOW_MEDIA !== '1') {
return errorResponse(
400,
'unsupported_media',
'Devin CLI print backend currently accepts text-only requests. Media/vision requires ACP (set DEVIN_CLI_MODE=acp + DEVIN_ACP_VISION=1) or an explicit media adapter.',
{ backend: 'devin-cli' },
);
}
const promptText = buildSpecialAgentPrompt(messages);
// When vision-over-ACP is active, carry the extracted images alongside the text
// as a structured prompt the ACP runner turns into image content blocks. Text
// prompt alone must still be non-empty (an image with no text is allowed — the
// images are the payload), so accept the request if EITHER text or images exist.
const visionImages = visionOverAcp ? extractInlineImages(messages) : [];
const prompt = visionImages.length ? { text: promptText, images: visionImages } : promptText;
if (!promptText && !visionImages.length) {
return errorResponse(400, 'invalid_request_error', 'Special-agent request has no text prompt.', { param: 'messages' });
}
let acct = null;
let handedOff = false;
try {
const mode = getBackendSwitch('devinCliMode');
// AC1 gap-e: proactively confirm the Devin CLI is runnable BEFORE we reserve
// a pool account / spawn the ACP child. A missing or non-executable binary
// is an ENVIRONMENT fault, not an account fault, so failing fast here avoids
// burning a checkout + RPM slot only to hit ENOENT post-spawn. The probe is
// zero-billable (devin --version only) and short-TTL cached. ACP-only:
// print mode keeps its own post-spawn ENOENT→503.
if (mode === 'acp') {
const probe = deps.probeDevinCliAvailable || probeDevinCliAvailable;
const status = await probe();
if (status && status.available === false) {
// 503 backend_unavailable — same shape the post-spawn ENOENT path emits,
// so the upper layer (failover / client) handles it identically. The
// pool is untouched because no account was checked out.
return errorResponse(
503,
'backend_unavailable',
`Devin CLI is not available for ACP routing (${status.reason || 'unavailable'}). Install the CLI or set DEVIN_CLI_PATH.`,
{ backend: 'devin-cli', probe_reason: status.reason || 'unavailable' },
);
}
}
if (process.env.DEVIN_CLI_USE_ACCOUNT_POOL !== '0') {
const checkout = deps.checkoutAccount || checkoutAccount;
acct = await checkout(route?.callerKey || '', modelKey);
if (!acct) {
return errorResponse(503, 'pool_exhausted', 'No active account is available for Devin CLI special-agent backend.');
}
}
const runner = mode === 'acp'
? (deps.runDevinAcp || runDevinAcp)
: (deps.runDevinPrint || runDevinPrint);
// Live streaming is only meaningful for ACP, whose runner emits chunks via
// onChunk. Print mode has no incremental output, so it keeps the buffered
// streamFromText path. The live streamer runs the runner itself, so it owns
// the account release + request accounting via onDone.
if (body?.stream && mode === 'acp') {
const releaser = deps.releaseAccount || releaseAccount;
const onSuccess = deps.reportSuccess || reportSuccess;
const stream = streamLiveAcp({
id, created, model, messages, prompt, modelKey, acct, runner, includeUsage,
signal: route?.signal || null,
onDone: (ok, err) => {
recordRequest(model, ok, Date.now() - started, acct?.id || null);
// AP-GAP-2: feed pool health from the streaming path too.
if (acct?.apiKey) {
if (ok) onSuccess(acct.apiKey);
else {
reportRunFailure(acct.apiKey, err, deps);
// AP-RISK-3: a failed request shouldn't keep burning an RPM slot
// (free tier RPM=10; consecutive failures would fake-saturate).
(deps.refundReservation || refundReservation)(acct.apiKey, acct.reservationTimestamp);
}
releaser(acct.apiKey);
}
},
});
handedOff = true;
return stream;
}
const result = await runner(prompt, {
modelKey,
apiKey: acct?.apiKey || '',
apiServerUrl: acct?.apiServerUrl || '',
signal: route?.signal || null,
});
const text = sanitizeText(result?.text || '');
const reasoning = exposeReasoning() ? sanitizeText(result?.reasoning || '') : '';
if (result?.stderr) log.debug(`special-agent devin stderr: ${String(result.stderr).slice(0, 240)}`);
recordRequest(model, true, Date.now() - started, acct?.id || null);
if (acct?.apiKey) (deps.reportSuccess || reportSuccess)(acct.apiKey);
// H3/H7: carry the runner's real usage + stop_reason through both shapes
// (null on print runners → pickUsage/mapFinishReason fall back gracefully).
const usage = result?.usage || null;
const stopReason = result?.stopReason || null;
const toolCalls = result?.toolCalls || null;
if (body?.stream) return streamFromText({ id, created, model, messages, text, reasoning, usage, stopReason, includeUsage });
return { status: 200, body: chatCompletionBody({ id, created, model, messages, text, reasoning, usage, stopReason, toolCalls }) };
} catch (err) {
const status = err?.status || 502;
const type = err?.type || 'backend_error';
log.warn(`special-agent ${modelKey} failed: ${err.message}`);
recordRequest(model, false, Date.now() - started, acct?.id || null);
// AP-GAP-2: classify the failure back to the pool (rate-limit / ban /
// internal / generic error) so a bad key stops being reselected.
if (acct?.apiKey) {
reportRunFailure(acct.apiKey, err, deps);
// AP-RISK-3: refund the RPM reservation so a failed request doesn't keep
// a free-tier slot (RPM=10) occupied and fake-saturate the account.
(deps.refundReservation || refundReservation)(acct.apiKey, acct.reservationTimestamp);
}
// sanitizeText handles paths; redactCredentialFragments handles credentials.
// Both are needed: err.message can carry CLI stderr, which can carry a token.
return errorResponse(status, type, redactCredentialFragments(sanitizeText(err.message || 'Special-agent backend failed')), {
backend: 'devin-cli',
});
} finally {
// The live streamer owns the account when handedOff: it releases in onDone
// after the stream completes. Releasing here would pull the key mid-stream.
if (!handedOff && acct?.apiKey) {
const releaser = deps.releaseAccount || releaseAccount;
releaser(acct.apiKey);
}
}
}
// Test-only surface. pickUsage decides this front's total_tokens shape, and it was the one
// usage builder the WINDSURFAPI_STRICT_USAGE_TOTAL flag missed on its first pass — exposing
// it means that front is checked by assertion rather than by reading.
export const __testing = { pickUsage, estimateTokens };