diff --git a/src/commands/x/provider.mjs b/src/commands/x/provider.mjs index a5a0c44..32eb890 100644 --- a/src/commands/x/provider.mjs +++ b/src/commands/x/provider.mjs @@ -58,12 +58,12 @@ Two independent axes: which host CLIs run the ruflo loop, and which LLM the routers use (aqe + ruflo). Mirrors \`ak x mcp\`: detect → persist to kit.json → idempotent heal. \`ak sync\` reapplies your choice. -Host model — three managed host integrations, two routing hosts: +Host model — three managed hosts, all eligible for explicit activity routing: claude, codex routing hosts: env-wired, one is primary, dual-host seeds the per-activity routing policy (they drive QE work) - opencode managed integration host: config-file wiring (opencode.json + opencode config-file wiring (opencode.json MCP + skills + permissions), lifecycle plugin, converted ruflo - agents, platform skill — never a routing target, never primary, + agents, platform skill — routable through ak run, never primary, never an aqe provider \`pick\` manages ALL THREE: enable/disable opencode here exactly like claude/codex. @@ -78,7 +78,7 @@ Subcommands: Options (pick, all optional — omit for interactive): --host the complete desired enabled-host set, e.g. claude,codex or claude,opencode (opencode is - wired + guided, never routed; excluding an + wired + guided; excluding an enabled host here DISABLES it — ak-managed wiring is stripped, user config preserved) --primary-host claude|codex which host leads (default claude; routing hosts @@ -164,10 +164,7 @@ async function status({ flags, cwd }) { : dflt ? 'enabled (default — ruflo default-on, no env written)' : d.wired ? 'enabled, wired' : 'enabled, not wired → ak sync'; - // The two host tiers, visible rather than implicit: claude/codex are the - // routing pair (primary + per-activity policy); opencode is a managed - // integration host — wired + guided, never a routing target. - const tier = h.id === 'opencode' ? dim(' · integration host (never routed)') + const tier = h.id === 'opencode' ? dim(' · routing host (ak run; never primary/AQE)') : dim(' · routing host'); // auth/billing axis — subscription ($0) vs metered key, per host. const auth = d.present ? hostAuthState(h.id, { present: true }) : null; @@ -382,16 +379,12 @@ async function maybeWriteQeCourtDefaults({ nonInteractive, cwd, enabled, aqeProv async function pick({ flags, cwd, pkgRoot }) { const cfg = loadKitConfig(); const hosts = await detectHosts(cwd); - // pick manages all managed host integrations (ADR-0016/0017): - // routing hosts — primary + per-activity policy seeds (registry capability) - // integration hosts — config-file wiring via opencode.mjs's owner module; - // never primary, never routed, never aqe - // The split is DERIVED from the host descriptors' capability flag, not a - // hardcoded id list (the seam issue #71's capability registry lands on). + // Routing eligibility is capability-derived. OpenCode retains its independent + // lifecycle wiring even though it is now an execution host; it is never a + // primary/AQE host because those are separate registry capabilities. // --host is the complete desired enabled-host set on BOTH tiers; excluding an // enabled host disables it (ak-managed wiring stripped, user config kept). const ROUTING = new Set(routableHostIds()); - const INTEGRATION = new Set(HOSTS.map((h) => h.id).filter((id) => !ROUTING.has(id))); const prevOpencode = !!cfg.providers?.hosts?.opencode || cfg.providers?.opencodeMcp === 'ak'; let enabled; let aqeProvider = cfg.providers.aqeProvider ?? null; @@ -422,20 +415,16 @@ async function pick({ flags, cwd, pkgRoot }) { } if (flags.provider !== undefined) models = parseModels(flags.provider); } else { - const installedRouting = HOSTS.filter((h) => hosts[h.id].present && ROUTING.has(h.id)).map((h) => h.id); - const installedIntegration = HOSTS.filter((h) => hosts[h.id].present && INTEGRATION.has(h.id)).map((h) => h.id); - if (installedRouting.length === 0 && installedIntegration.length === 0) { fail('no frontier CLI (claude/codex/opencode) found on PATH'); return 1; } - if (installedIntegration.length) { - console.log(`Installed hosts: ${[...installedRouting, ...installedIntegration].join(', ')}` - + dim(` (${installedIntegration.join(', ')} = integration host — wired + guided, never a routing target)`)); - } else { - console.log(`Installed hosts: ${installedRouting.join(', ')}`); - } + const installedRouting = HOSTS.filter((h) => hosts[h.id].present && ROUTING.has(h.id) + && (h.id !== 'opencode' || cfg.providers.hosts.opencode)).map((h) => h.id); + const installedOpenCode = hosts.opencode?.present && !cfg.providers.hosts.opencode; + if (installedRouting.length === 0 && !installedOpenCode) { fail('no frontier CLI (claude/codex/opencode) found on PATH'); return 1; } + console.log(`Installed hosts: ${installedRouting.join(', ') || 'none'}${installedOpenCode ? dim(' (opencode is available; type it to opt in)') : ''}`); // Default: every currently ENABLED host (even one temporarily absent from // PATH — a bare enter must never tear down an enabled host it simply can't // see right now) ∪ newly detected routing hosts. An installed-but-disabled - // integration host is opt-in by typing it — a bare enter must not opt a - // third host's config home in sight unseen either (codex-review r3). + // OpenCode host remains opt-in by typing it — a bare enter must not opt a + // third host's config home in sight unseen either. const enabledHosts = HOSTS.filter((h) => cfg.providers.hosts[h.id]).map((h) => h.id); const dflt = [...new Set([...enabledHosts, ...installedRouting])]; const absentEnabled = enabledHosts.filter((h) => !hosts[h].present); @@ -469,13 +458,11 @@ async function pick({ flags, cwd, pkgRoot }) { fail(`unknown host(s): ${unknown.join(', ')} (valid: ${[...known].join(', ')}) — nothing changed`); return 2; } - // Split the tiers: routing hosts drive primary/seeds; integration hosts are - // wired + guided. The routing pair needs at least one member (primaryHost - // must be enabled) — fall back to claude, keeping any integration choice. + // The routing set needs at least one primary-capable member; OpenCode remains + // routable but cannot satisfy that primary-host invariant on its own. const routing = enabled.filter((h) => ROUTING.has(h)); - const integrations = enabled.filter((h) => INTEGRATION.has(h)); - if (!routing.length) routing.push('claude'); - enabled = [...routing, ...integrations]; + if (!routing.some((h) => PRIMARY_HOSTS.includes(h))) routing.unshift('claude'); + enabled = [...new Set(routing)]; // primary host — which host leads (default claude); must be a ROUTING host. let primaryHost = prevPrimary; if (flags['primary-host'] !== undefined) { @@ -519,7 +506,7 @@ async function pick({ flags, cwd, pkgRoot }) { hosts: { claude: routing.includes('claude'), codex: routing.includes('codex'), - opencode: integrations.includes('opencode'), + opencode: enabled.includes('opencode'), }, aqeProvider, aqeFallback, @@ -546,7 +533,7 @@ async function pick({ flags, cwd, pkgRoot }) { for (const w of warnings) warn(w); cfg.providers.dualRouting = { ...cfg.providers.dualRouting, ...policy }; } - const prunedRoutes = pruneRoutesForHosts(cfg.providers.dualRouting, { hosts: routing }); + const prunedRoutes = pruneRoutesForHosts(cfg.providers.dualRouting, { hosts: enabled }); cfg.providers.dualRouting = prunedRoutes.policy; for (const message of prunedRoutes.warnings) warn(message); diff --git a/src/lib/adapters/registries.mjs b/src/lib/adapters/registries.mjs index f27e7ee..d0a023d 100644 --- a/src/lib/adapters/registries.mjs +++ b/src/lib/adapters/registries.mjs @@ -139,7 +139,7 @@ const hostEntries = [ { id: 'opencode', label: 'OpenCode', install: { bin: 'opencode', npmPackage: 'opencode-ai', externalInstallPolicy: 'detect-never-overwrite' }, - capabilities: { canDriveSession: true, canBePrimary: false, canRouteActivities: false, commandStatusline: false, transcripts: true, usage: false, nativeMcpConfig: true, nativeGuidance: true }, + capabilities: { canDriveSession: true, canBePrimary: false, canRouteActivities: true, commandStatusline: false, transcripts: true, usage: false, nativeMcpConfig: true, nativeGuidance: true }, auth: { apiKeyEnv: [], loginFile: ['.local', 'share', 'opencode', 'auth.json'], keyOverridesLogin: false }, legacy: { guidanceFile: 'agents-opencode', configFormat: 'json', diff --git a/src/lib/execution/opencode.mjs b/src/lib/execution/opencode.mjs index 0f791aa..9f78593 100644 --- a/src/lib/execution/opencode.mjs +++ b/src/lib/execution/opencode.mjs @@ -29,6 +29,31 @@ function defaultReservePort() { }); } +async function waitForChildClose(child, timeoutMs) { + if (!child?.once || child.exitCode != null || child.signalCode != null) return true; + let timer; + try { + return await Promise.race([ + new Promise((resolve) => child.once('close', () => resolve(true))), + new Promise((resolve) => { timer = setTimeout(() => resolve(false), timeoutMs); }), + ]); + } finally { clearTimeout(timer); } +} + +/** Terminate only the direct server child. A TERM that does not produce a close + * event becomes explicit orphan evidence after one bounded KILL fallback. */ +async function stopChild(child, { terminationGraceMs, forceGraceMs }) { + if (!child?.kill || child.exitCode != null || child.signalCode != null) return { stopped: true }; + if (!child.once) { + try { child.kill('SIGTERM'); } catch { return { stopped: false }; } + return { stopped: true }; + } + try { child.kill('SIGTERM'); } catch { return { stopped: false }; } + if (await waitForChildClose(child, terminationGraceMs)) return { stopped: true }; + try { child.kill('SIGKILL'); } catch { return { stopped: false }; } + return { stopped: await waitForChildClose(child, forceGraceMs) }; +} + function templateText(readFileSync = fs.readFileSync) { return readFileSync(TEMPLATE_PATH, 'utf8'); } @@ -70,15 +95,30 @@ async function requestJson(fetchFn, endpoint, password, pathname, } async function requestNoContent(fetchFn, endpoint, password, pathname, - { method = 'POST', body } = /** @type {{method?:string, body?:any}} */ ({})) { + { method = 'POST', body, signal } = /** @type {{method?:string, body?:any, signal?:AbortSignal}} */ ({})) { const headers = basicHeaders(password); if (body !== undefined) headers['content-type'] = 'application/json'; const response = await fetchFn(`${endpoint}${pathname}`, { - method, headers, ...(body === undefined ? {} : { body: JSON.stringify(body) }), + method, headers, ...(body === undefined ? {} : { body: JSON.stringify(body) }), ...(signal ? { signal } : {}), }); if (!response?.ok) throw new Error(`${method} ${pathname} failed with HTTP ${response?.status ?? 'unknown'}`); } +async function requestWithin(fetchFn, endpoint, password, pathname, options, timeoutMs) { + const controller = new AbortController(); + let timer; + try { + return await Promise.race([ + requestNoContent(fetchFn, endpoint, password, pathname, { ...options, signal: controller.signal }), + new Promise((_, reject) => { timer = setTimeout(() => { + controller.abort(); + const error = Object.assign(new Error(`${options.method ?? 'POST'} ${pathname} timed out`), { code: 'ETIMEDOUT' }); + reject(error); + }, timeoutMs); }), + ]); + } finally { clearTimeout(timer); } +} + async function waitForHealth(fetchFn, endpoint, password, { attempts = 40, wait = delay } = {}) { let lastError = null; for (let attempt = 0; attempt < attempts; attempt++) { @@ -99,9 +139,11 @@ function normalizeEvent(value) { /** Read only the first terminal session event. The SSE parsing is deliberately * tolerant of chunk boundaries but rejects malformed data rather than inventing * completion. */ -async function waitForTerminalEvent(response, sessionId) { +async function waitForTerminalEvent(response, sessionId, { signal } = /** @type {{signal?:AbortSignal}} */ ({})) { if (!response?.ok || !response.body?.getReader) throw new Error('GET /global/event did not return an SSE body'); const reader = response.body.getReader(); + const abort = () => { void reader.cancel(); }; + signal?.addEventListener?.('abort', abort, { once: true }); const decoder = new TextDecoder(); let buffer = ''; let data = []; @@ -119,18 +161,22 @@ async function waitForTerminalEvent(response, sessionId) { if (event?.type === 'session.status' && properties.sessionID === sessionId && properties.status?.type === 'idle') return { type: 'idle' }; return null; }; - for (;;) { - const { done, value } = await reader.read(); - buffer += decoder.decode(value ?? new Uint8Array(), { stream: !done }); - const lines = buffer.split(/\r?\n/); - buffer = lines.pop() ?? ''; - for (const line of lines) { - const terminal = await consume(line); - if (terminal) { await reader.cancel(); return terminal; } + try { + for (;;) { + const { done, value } = await reader.read(); + buffer += decoder.decode(value ?? new Uint8Array(), { stream: !done }); + const lines = buffer.split(/\r?\n/); + buffer = lines.pop() ?? ''; + for (const line of lines) { + const terminal = await consume(line); + if (terminal) { await reader.cancel(); return terminal; } + } + if (done) break; } - if (done) break; + throw new Error(signal?.aborted ? 'OpenCode SSE stream cancelled' : 'OpenCode SSE stream ended before the session became terminal'); + } finally { + signal?.removeEventListener?.('abort', abort); } - throw new Error('OpenCode SSE stream ended before the session became terminal'); } function assistantFrom(messages) { @@ -144,6 +190,7 @@ function assistantFrom(messages) { function errorCategory(error) { if (error?.name === 'ProviderAuthError') return 'auth_required'; + if (error?.name === 'ProtocolError') return 'protocol_error'; return 'worker_error'; } @@ -161,6 +208,8 @@ function terminalResult(state, observation, clock) { status = 'timed_out'; exitCategory = 'timeout'; failure = { reason: 'timeout' }; } else if (observation.type === 'cancelled') { status = 'cancelled'; exitCategory = 'cancelled'; failure = { reason: 'cancelled' }; + } else if (observation.type === 'orphaned') { + status = 'failed'; exitCategory = 'orphaned'; failure = { reason: 'owned OpenCode server did not terminate' }; } else if (observation.type === 'error' || assistant?.error) { const error = observation.error ?? assistant?.error ?? null; status = 'failed'; exitCategory = errorCategory(error); failure = error ?? { reason: 'OpenCode session failed' }; @@ -186,9 +235,11 @@ function terminalResult(state, observation, clock) { */ export function createOpenCodeExecutionAdapter({ fetchFn = globalThis.fetch, spawnFn = nodeSpawn, haveFn = have, reservePort = defaultReservePort, - secret = defaultSecret, wait = delay, clock = nowIso, + secret = defaultSecret, wait = delay, clock = nowIso, terminationGraceMs = 1_500, forceGraceMs = 1_500, } = {}) { if (typeof fetchFn !== 'function') throw new TypeError('fetchFn is required'); + if (!Number.isInteger(terminationGraceMs) || terminationGraceMs < 1) throw new TypeError('terminationGraceMs must be a positive integer'); + if (!Number.isInteger(forceGraceMs) || forceGraceMs < 1) throw new TypeError('forceGraceMs must be a positive integer'); const adapter = { id: 'opencode-server', async readiness() { @@ -200,7 +251,7 @@ export function createOpenCodeExecutionAdapter({ if (!path.isAbsolute(cwd)) throw new TypeError('OpenCode worker cwd must be absolute'); return { worker, cwd, prompt: renderOpenCodeWorkerPrompt(worker), startedAt: clock() }; }, - async launch(state) { + async launch(state, { timeoutMs = 120_000 } = {}) { const port = await reservePort(); const password = secret(); const endpoint = `http://${LOOPBACK}:${port}`; @@ -216,14 +267,15 @@ export function createOpenCodeExecutionAdapter({ }); if (typeof session?.id !== 'string' || !session.id) throw new Error('OpenCode created a session without an id'); const headers = basicHeaders(password); - const eventResponse = await fetchFn(`${endpoint}/global/event`, { headers }); - const terminal = waitForTerminalEvent(eventResponse, session.id); - await requestNoContent(fetchFn, endpoint, password, `/session/${encodeURIComponent(session.id)}/prompt_async`, { + const eventAbort = new AbortController(); + const eventResponse = await fetchFn(`${endpoint}/global/event`, { headers, signal: eventAbort.signal }); + const terminal = waitForTerminalEvent(eventResponse, session.id, { signal: eventAbort.signal }); + await requestWithin(fetchFn, endpoint, password, `/session/${encodeURIComponent(session.id)}/prompt_async`, { body: { agent: 'build', ...(state.worker.configuredModel ? { model: state.worker.configuredModel } : {}), parts: [{ type: 'text', text: state.prompt }] }, - }); - return { ...state, endpoint, password, child, sessionId: session.id, terminal }; + }, timeoutMs); + return { ...state, endpoint, password, child, sessionId: session.id, terminal, eventAbort }; } catch (error) { - try { child.kill('SIGTERM'); } catch { /* cleanup is best-effort */ } + await stopChild(child, { terminationGraceMs, forceGraceMs }); throw error; } }, @@ -242,16 +294,20 @@ export function createOpenCodeExecutionAdapter({ }, interpret(state, observation) { return terminalResult(state, observation, clock); }, async cancel(state) { + state?.eventAbort?.abort(); if (state?.sessionId) { try { await requestNoContent(fetchFn, state.endpoint, state.password, `/session/${encodeURIComponent(state.sessionId)}/abort`); } catch { /* cleanup records the final truth */ } } - try { state?.child?.kill?.('SIGTERM'); return { type: 'cancelled' }; } catch { return { type: 'cancelled', orphaned: true }; } + const stopped = await stopChild(state?.child, { terminationGraceMs, forceGraceMs }); + return stopped.stopped ? { type: 'cancelled' } : { type: 'cancelled', orphaned: true }; }, async cleanup(state) { + state?.eventAbort?.abort(); if (state?.endpoint) { try { await requestNoContent(fetchFn, state.endpoint, state.password, '/instance/dispose'); } catch { /* child termination remains the fallback */ } } - try { state?.child?.kill?.('SIGTERM'); return { cleaned: true }; } catch { return { cleaned: false, orphaned: true }; } + const stopped = await stopChild(state?.child, { terminationGraceMs, forceGraceMs }); + return stopped.stopped ? { cleaned: true } : { cleaned: false, orphaned: true }; }, }; return validateExecutionAdapter(adapter); diff --git a/src/lib/execution/runner.mjs b/src/lib/execution/runner.mjs index 17841d0..40f010b 100644 --- a/src/lib/execution/runner.mjs +++ b/src/lib/execution/runner.mjs @@ -43,16 +43,21 @@ export async function executeWorker(worker, adapter, { if (!ready?.ready) return workerFailure(worker, { exitCategory: ready?.exitCategory ?? 'cli_unavailable', failure: { reason: 'host is not ready' }, startedAt, clock, }); - state = await adapter.prepare({ worker, cwd }); - state = await adapter.launch(state); + state = await adapter.prepare({ worker, cwd, timeoutMs }); + state = await adapter.launch(state, { timeoutMs }); const watched = await observeBeforeDeadline(adapter, state, timeoutMs); if (watched.timedOut) { - await adapter.cancel(state); - return adapter.interpret(state, { type: 'timeout' }); + const cancelled = await adapter.cancel(state); + return adapter.interpret(state, { type: cancelled?.orphaned ? 'orphaned' : 'timeout' }); } return validateWorkerResult(adapter.interpret(state, watched.observation)); } catch (error) { - return workerFailure(worker, { exitCategory: 'protocol_error', failure: boundedFailure(error), startedAt, clock }); + const timedOut = error?.code === 'ETIMEDOUT'; + return workerFailure(worker, { + status: timedOut ? 'timed_out' : 'failed', + exitCategory: timedOut ? 'timeout' : 'protocol_error', + failure: boundedFailure(error), startedAt, clock, + }); } finally { if (state) { try { await adapter.cleanup(state); } catch { /* terminal result is already authoritative */ } diff --git a/src/lib/providers.mjs b/src/lib/providers.mjs index 40a5392..0e1952f 100644 --- a/src/lib/providers.mjs +++ b/src/lib/providers.mjs @@ -18,8 +18,8 @@ // OpenAI models are reached via `openai`). // // Two independent axes: -// host axis — which agent CLI runs the ruflo loop (claude, codex). ruflo runs -// both at once (dual-mode). This is about the coding-agent CLI. +// host axis — which agent CLI executes a managed worker (claude, codex, +// opencode). Ruvlo's dual mode itself remains Claude/Codex. // provider axis — which LLM the *routers* use: ruflo's API-key providers // (`ruflo providers configure`) and aqe's `AQE_LLM_PROVIDER`. // Independent of the host axis; keys live in the env, never kit.json. @@ -302,8 +302,9 @@ export async function collectIntegrationFacts({ }); } -/** Hosts eligible for legacy setup/sync installation and teardown loops. */ -export const commandHosts = () => HOSTS.filter((host) => +/** Hosts eligible for legacy env-backed setup/sync loops. OpenCode has its own + * owner-module lifecycle because it has no ruflo enable-env projection. */ +export const commandHosts = () => HOSTS.filter((host) => host.enableEnv && HOST_REGISTRY.find((entry) => entry.id === host.id)?.capabilities.canRouteActivities); /** Where host-enable env lands: project settings.local.json inside a repo (same @@ -558,7 +559,7 @@ export function formatRoutingTable(cfg) { const routes = resolveRoutes(policy); const s = routingSummary(policy); const lines = [bold('\nper-activity routing') - + dim(` (${s.byHost.claude ?? 0} claude · ${s.byHost.codex ?? 0} codex · ${s.custom} custom · .agentic-qe/llm-config.json)`)]; + + dim(` (${s.byHost.claude ?? 0} claude · ${s.byHost.codex ?? 0} codex · ${s.byHost.opencode ?? 0} opencode · ${s.custom} custom · .agentic-qe/llm-config.json)`)]; // "diverges from", never "stale"/"outdated"/"superseded": the pinned model is // sometimes the better choice for an activity, so the wording must present a // decision rather than a lag (#55). diff --git a/src/lib/routing.mjs b/src/lib/routing.mjs index e1b1a06..6ba9394 100644 --- a/src/lib/routing.mjs +++ b/src/lib/routing.mjs @@ -19,7 +19,9 @@ export const ACTIVITIES = [ // Activities ak originated (no upstream rUv template) — flagged wherever surfaced (ADR-0002). export const AK_ORIGINATED = new Set(['packaging', 'release']); -// Host → aqe/router provider type. Both are subscription-billed ($0 marginal). +// Host → aqe/router provider type. OpenCode deliberately has no entry: its +// execution provider is observed per worker and must never be inferred from the +// host or silently projected into AQE's separate provider vocabulary. export const HOST_PROVIDER = { claude: 'claude-code', codex: 'codex' }; export const HOSTS = routableHostIds(); @@ -96,7 +98,7 @@ export function formatModelHelp() { ]; for (const host of HOSTS) { lines.push(` ${host}:`); - for (const m of MODEL_CATALOG[host]) lines.push(` ${m.id.padEnd(28)} ${m.tier.padEnd(10)} ${m.note}`); + for (const m of MODEL_CATALOG[host] ?? []) lines.push(` ${m.id.padEnd(28)} ${m.tier.padEnd(10)} ${m.note}`); } for (const [prov, models] of Object.entries(PROVIDER_MODEL_CATALOG)) { lines.push(` ${prov} (aqe-fallback provider — metered):`); @@ -454,6 +456,10 @@ export function materializeRunPlan(policy = {}, { template = 'feature', task = ' */ export function policyToDualRunConfig(policy = {}, opts = {}) { const plan = materializeRunPlan(policy, opts); + const unsupported = plan.workers.filter((worker) => !['claude', 'codex'].includes(worker.host)); + if (unsupported.length) { + throw new Error(`ak dual supports Claude/Codex workers only; use ak run for ${[...new Set(unsupported.map((worker) => worker.host))].join(', ')}`); + } return { workers: plan.workers.map((worker) => ({ id: worker.id, @@ -492,7 +498,11 @@ export function escalatePolicy(policy = {}) { */ export function routedVendors(policy = {}) { const routes = resolveRoutes(policy); - return new Set(Object.values(routes).map((r) => vendorOf(HOST_PROVIDER[r.host]))); + return new Set(Object.values(routes) + .map((r) => HOST_PROVIDER[r.host]) + .filter(Boolean) + .map((provider) => vendorOf(provider)) + .filter(Boolean)); } /** Compact summary for status rows / dashboard / tables. */ @@ -518,7 +528,7 @@ export function validateRoute(route = {}) { const { host, model } = route; const errs = []; if (!isRoutableHost(host)) errs.push(`unknown host "${host}" (expected: ${HOSTS.join('|')})`); - else if (!AQE_CONSTRUCTIBLE_PROVIDERS.includes(HOST_PROVIDER[host])) errs.push(`host "${host}" maps to a non-constructible provider`); + else if (HOST_PROVIDER[host] && !AQE_CONSTRUCTIBLE_PROVIDERS.includes(HOST_PROVIDER[host])) errs.push(`host "${host}" maps to a non-constructible provider`); if (model != null && (typeof model !== 'string' || model.trim() === '')) errs.push('model must be a non-empty string'); return errs; } diff --git a/tests/kit/integration-command-facts.test.mjs b/tests/kit/integration-command-facts.test.mjs index daae584..35676a5 100644 --- a/tests/kit/integration-command-facts.test.mjs +++ b/tests/kit/integration-command-facts.test.mjs @@ -24,10 +24,10 @@ test('commands share one immutable normalized integration snapshot', async () => assert.equal(JSON.stringify(facts).includes('present-only-in-memory'), false); }); -test('managed but non-routable OpenCode never enters legacy install/routing loops', () => { +test('routable OpenCode keeps its config lifecycle out of legacy env-backed loops', () => { const opencode = HOST_REGISTRY.find(({ id }) => id === 'opencode'); assert.ok(opencode); assert.equal(opencode.capabilities.canDriveSession, true); - assert.equal(opencode.capabilities.canRouteActivities, false); + assert.equal(opencode.capabilities.canRouteActivities, true); assert.equal(commandHosts().some(({ id }) => id === 'opencode'), false); }); diff --git a/tests/kit/opencode-execution.test.mjs b/tests/kit/opencode-execution.test.mjs index 47bcd3e..39a291a 100644 --- a/tests/kit/opencode-execution.test.mjs +++ b/tests/kit/opencode-execution.test.mjs @@ -1,6 +1,7 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import { createOpenCodeExecutionAdapter, renderOpenCodeWorkerPrompt } from '../../src/lib/execution/opencode.mjs'; +import { executeWorker } from '../../src/lib/execution/runner.mjs'; const worker = { id: 'worker-1', activity: 'implementation', role: 'coder', host: 'opencode', @@ -70,3 +71,86 @@ test('permission events are deterministically aborted and never converted into i assert.ok(paths.includes('POST /session/ses-2/abort')); assert.ok(!paths.some((p) => p.includes('/permissions/perm-1'))); }); + +test('malformed server events fail as protocol errors instead of manufacturing completion', async () => { + const fetchFn = async (url, init = {}) => { + if (url.endsWith('/global/health')) return response({ healthy: true }); + if (url.endsWith('/session') && init.method === 'POST') return response({ id: 'ses-3' }); + if (url.endsWith('/global/event')) return response(null, { body: sse('data: {not-json}\n\n') }); + if (url.endsWith('/prompt_async') || url.endsWith('/instance/dispose')) return response(null, { status: 204 }); + throw new Error(`unexpected URL ${url}`); + }; + const adapter = createOpenCodeExecutionAdapter({ + fetchFn, spawnFn: () => ({ kill: () => true }), reservePort: async () => 43125, secret: () => 'ephemeral', clock: () => '2026-07-29T00:00:00.000Z', + }); + const launched = await adapter.launch(await adapter.prepare({ worker, cwd: process.cwd() })); + const result = adapter.interpret(launched, await adapter.observe(launched)); + assert.equal(result.status, 'failed'); + assert.equal(result.exitCategory, 'protocol_error'); + await adapter.cleanup(launched); +}); + +test('runner timeout aborts the session event stream and terminates the owned server', async () => { + const paths = []; + let cancelled = false; + const pending = new ReadableStream({ + start(controller) { this.controller = controller; }, + cancel() { cancelled = true; }, + }); + const child = { signals: [], kill(signal) { this.signals.push(signal); return true; } }; + const fetchFn = async (url, init = {}) => { + paths.push(`${init.method ?? 'GET'} ${new URL(url).pathname}`); + if (url.endsWith('/global/health')) return response({ healthy: true }); + if (url.endsWith('/session') && init.method === 'POST') return response({ id: 'ses-4' }); + if (url.endsWith('/global/event')) return response(null, { body: pending }); + if (url.endsWith('/prompt_async') || url.endsWith('/abort') || url.endsWith('/instance/dispose')) return response(null, { status: 204 }); + throw new Error(`unexpected URL ${url}`); + }; + const adapter = createOpenCodeExecutionAdapter({ + fetchFn, spawnFn: () => child, haveFn: async () => true, reservePort: async () => 43126, secret: () => 'ephemeral', + }); + const result = await executeWorker(worker, adapter, { cwd: process.cwd(), timeoutMs: 100 }); + assert.equal(result.status, 'timed_out'); + assert.equal(result.exitCategory, 'timeout'); + assert.equal(cancelled, true); + assert.ok(paths.includes('POST /session/ses-4/abort')); + assert.deepEqual(child.signals, ['SIGTERM', 'SIGTERM']); +}); + +test('a server that ignores TERM receives one bounded KILL fallback and reports an orphan', async () => { + const child = new (await import('node:events')).EventEmitter(); + child.exitCode = null; + child.kill = () => true; + const fetchFn = async (url, init = {}) => { + if (url.endsWith('/global/health')) return response({ healthy: true }); + if (url.endsWith('/session') && init.method === 'POST') return response({ id: 'ses-5' }); + if (url.endsWith('/global/event')) return response(null, { body: new ReadableStream({ cancel() {} }) }); + if (url.endsWith('/prompt_async') || url.endsWith('/abort') || url.endsWith('/instance/dispose')) return response(null, { status: 204 }); + throw new Error(`unexpected URL ${url}`); + }; + const adapter = createOpenCodeExecutionAdapter({ + fetchFn, spawnFn: () => child, haveFn: async () => true, reservePort: async () => 43127, secret: () => 'ephemeral', terminationGraceMs: 1, forceGraceMs: 1, + }); + const result = await executeWorker(worker, adapter, { cwd: process.cwd(), timeoutMs: 100 }); + assert.equal(result.status, 'failed'); + assert.equal(result.exitCategory, 'orphaned'); +}); + +test('a stalled prompt submission becomes a timeout and tears down its owned server', async () => { + const child = { signals: [], kill(signal) { this.signals.push(signal); return true; } }; + const fetchFn = async (url, init = {}) => { + if (url.endsWith('/global/health')) return response({ healthy: true }); + if (url.endsWith('/session') && init.method === 'POST') return response({ id: 'ses-6' }); + if (url.endsWith('/global/event')) return response(null, { body: new ReadableStream({ cancel() {} }) }); + if (url.endsWith('/prompt_async')) return new Promise(() => {}); + throw new Error(`unexpected URL ${url}`); + }; + const adapter = createOpenCodeExecutionAdapter({ + fetchFn, spawnFn: () => child, haveFn: async () => true, reservePort: async () => 43128, secret: () => 'ephemeral', + }); + const result = await executeWorker(worker, adapter, { cwd: process.cwd(), timeoutMs: 20 }); + assert.equal(result.status, 'timed_out'); + assert.equal(result.exitCategory, 'timeout'); + assert.match(result.failure.reason, /prompt_async timed out/); + assert.deepEqual(child.signals, ['SIGTERM']); +}); diff --git a/tests/kit/provider-cli.test.mjs b/tests/kit/provider-cli.test.mjs index 58c9ff6..6f445df 100644 --- a/tests/kit/provider-cli.test.mjs +++ b/tests/kit/provider-cli.test.mjs @@ -316,14 +316,14 @@ test('excluding claude keeps codex-owned bridges while pruning claude routes', ( } }); -test('interactive pick: installed opencode is displayed as an integration host but only ENTER-enabled when already on', () => { +test('interactive pick keeps installed OpenCode opt-in until explicitly selected', () => { const sb = pickSandbox({ hosts: { claude: true, codex: false, opencode: false } }); try { // Blank answers to every prompt: accept the defaults. const r = akPick(['x', 'provider', 'pick'], sb, { input: '\n\n\n\n' }); assert.equal(r.status, 0, `interactive pick failed\nstdout: ${r.stdout}\nstderr: ${r.stderr}`); - assert.match(r.stdout, /integration host — wired \+ guided, never a routing target/, - 'opencode is displayed with its non-routing qualifier'); + assert.match(r.stdout, /opencode is available; type it to opt in/, + 'OpenCode is displayed but not implicitly enabled'); assert.equal(kitJson(sb.home).providers.hosts.opencode, false, 'a bare enter must not opt a third host in sight unseen'); } finally { diff --git a/tests/kit/providers.test.mjs b/tests/kit/providers.test.mjs index efee094..59b20b4 100644 --- a/tests/kit/providers.test.mjs +++ b/tests/kit/providers.test.mjs @@ -221,7 +221,7 @@ test('every host descriptor carries an npm package name for install/update', () test('managed and routable host sets come only from registry capabilities', () => { assert.deepEqual(managedHostIds(), ['claude', 'codex', 'opencode']); - assert.deepEqual(routableHostIds(), ['claude', 'codex']); + assert.deepEqual(routableHostIds(), ['claude', 'codex', 'opencode']); assert.deepEqual(HOSTS.map((h) => h.id), managedHostIds()); }); diff --git a/tests/kit/routing.test.mjs b/tests/kit/routing.test.mjs index a845629..8be3cf8 100644 --- a/tests/kit/routing.test.mjs +++ b/tests/kit/routing.test.mjs @@ -190,10 +190,13 @@ test('host-neutral run plan preserves every legacy dual worker assignment', () = assert.ok(plan.workers.every((worker) => worker.activity && worker.host && !('platform' in worker))); }); -test('a managed but non-routable host cannot materialize a runnable plan', () => { - assert.throws(() => materializeRunPlan({ implementation: { +test('an explicit OpenCode route materializes for ak run but not the legacy dual adapter', () => { + const policy = { implementation: { host: 'opencode', model: 'openrouter/example', source: 'user', - } }, { template: 'feature', task: 'x' }), /implementation.*opencode.*canRouteActivities/); + } }; + const plan = materializeRunPlan(policy, { template: 'feature', task: 'x' }); + assert.equal(plan.workers.find((worker) => worker.activity === 'implementation').host, 'opencode'); + assert.throws(() => policyToDualRunConfig(policy, { template: 'feature', task: 'x' }), /ak dual supports Claude\/Codex/); }); test('policyToDualRunConfig throws on an unknown template', () => { diff --git a/tests/kit/run-command.test.mjs b/tests/kit/run-command.test.mjs index 2147182..52e8da7 100644 --- a/tests/kit/run-command.test.mjs +++ b/tests/kit/run-command.test.mjs @@ -11,9 +11,12 @@ test('ak run materializes the host-neutral plan and keeps run-local route overri assert.equal(cfg.providers.dualRouting.implementation.host, 'codex'); }); -test('ak run rejects an OpenCode route until the capability proof is complete', () => { +test('ak run materializes an explicit OpenCode route', () => { const cfg = { providers: { dualRouting: { ...seedDualRouting(), 'security-scan': { host: 'opencode', model: 'openrouter/example', source: 'user', } } } }; - assert.throws(() => buildRunPlan(cfg, 'security', 'src/auth'), /canRouteActivities/); + const { plan } = buildRunPlan(cfg, 'security', 'src/auth'); + const scanner = plan.workers.find((entry) => entry.activity === 'security-scan'); + assert.equal(scanner.host, 'opencode'); + assert.equal(scanner.configuredModel, 'openrouter/example'); });