diff --git a/src/lib/execution/adapters.mjs b/src/lib/execution/adapters.mjs new file mode 100644 index 00000000..ae4365f8 --- /dev/null +++ b/src/lib/execution/adapters.mjs @@ -0,0 +1,7 @@ +// Built-in execution adapters. Importing this registry has no side effects: +// processes are only created when the runner selects an adapter for a worker. +import { OPENCODE_EXECUTION_ADAPTER } from './opencode.mjs'; + +export const EXECUTION_ADAPTERS = Object.freeze(new Map([ + ['opencode', OPENCODE_EXECUTION_ADAPTER], +])); diff --git a/src/lib/execution/runner.mjs b/src/lib/execution/runner.mjs new file mode 100644 index 00000000..17841d03 --- /dev/null +++ b/src/lib/execution/runner.mjs @@ -0,0 +1,129 @@ +// Host-neutral execution coordinator. It owns scheduling, deadlines, and +// lifecycle cleanup; adapters own each host's transport and protocol details. +import { validateExecutionAdapter, validateWorkerResult } from './schema.mjs'; +import { EXECUTION_ADAPTERS } from './adapters.mjs'; + +const nowIso = () => new Date().toISOString(); + +function workerFailure(worker, /** @type {{status?:string, exitCategory?:string, failure?:any, startedAt?:string, clock?:()=>string}} */ + { status = 'failed', exitCategory = 'worker_error', failure = null, startedAt, clock = nowIso } = {}) { + const endedAt = clock(); + return validateWorkerResult({ + workerId: worker.id, activity: worker.activity, role: worker.role, host: worker.host, + status, exitCategory, startedAt: startedAt ?? endedAt, endedAt, + durationMs: Math.max(0, Date.parse(endedAt) - Date.parse(startedAt ?? endedAt)), + provider: null, providerProvenance: 'unknown', configuredModel: worker.configuredModel ?? null, + observedModel: null, sessionId: null, transcriptRefs: [], failure, usage: null, + }); +} + +function boundedFailure(error) { + return { reason: String(error?.message ?? error ?? 'execution failed').slice(0, 240) }; +} + +async function observeBeforeDeadline(adapter, state, timeoutMs) { + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) return { timedOut: false, observation: await adapter.observe(state) }; + let timer; + try { + return await Promise.race([ + adapter.observe(state).then((observation) => ({ timedOut: false, observation })), + new Promise((resolve) => { timer = setTimeout(() => resolve({ timedOut: true }), timeoutMs); }), + ]); + } finally { clearTimeout(timer); } +} + +/** Execute one worker and guarantee cleanup after every prepared state. */ +export async function executeWorker(worker, adapter, { + cwd = process.cwd(), timeoutMs = 120_000, clock = nowIso, +} = /** @type {{cwd?:string, timeoutMs?:number, clock?:()=>string}} */ ({})) { + const startedAt = clock(); + let state = null; + try { + const ready = await adapter.readiness({ worker, cwd }); + 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); + const watched = await observeBeforeDeadline(adapter, state, timeoutMs); + if (watched.timedOut) { + await adapter.cancel(state); + return adapter.interpret(state, { type: 'timeout' }); + } + return validateWorkerResult(adapter.interpret(state, watched.observation)); + } catch (error) { + return workerFailure(worker, { exitCategory: 'protocol_error', failure: boundedFailure(error), startedAt, clock }); + } finally { + if (state) { + try { await adapter.cleanup(state); } catch { /* terminal result is already authoritative */ } + } + } +} + +function validatePlan(plan) { + if (!plan || !Array.isArray(plan.workers)) throw new TypeError('execution plan requires workers'); + const ids = new Set(); + for (const worker of plan.workers) { + if (!worker || typeof worker !== 'object' || typeof worker.id !== 'string' || !worker.id + || typeof worker.activity !== 'string' || typeof worker.role !== 'string' || typeof worker.host !== 'string') { + throw new TypeError('execution worker requires id, activity, role, and host'); + } + if (ids.has(worker.id)) throw new Error(`execution plan has duplicate worker id "${worker.id}"`); + ids.add(worker.id); + } + for (const worker of plan.workers) { + for (const dependency of worker.dependsOn ?? []) { + if (!ids.has(dependency)) throw new Error(`worker "${worker.id}" depends on unknown worker "${dependency}"`); + if (dependency === worker.id) throw new Error(`worker "${worker.id}" cannot depend on itself`); + } + } +} + +function adapterFor(adapters, host) { + const adapter = adapters instanceof Map ? adapters.get(host) : adapters?.[host]; + return adapter ? validateExecutionAdapter(adapter) : null; +} + +/** Run a materialized routing plan as a bounded-concurrency dependency DAG. + * A failed dependency blocks descendants; independent branches keep running. */ +export async function executeRunPlan(plan, { + adapters = EXECUTION_ADAPTERS, cwd = process.cwd(), maxConcurrent = 4, timeoutMs, clock = nowIso, +} = /** @type {{adapters?:Record|Map, cwd?:string, maxConcurrent?:number, timeoutMs?:number, clock?:()=>string}} */ ({})) { + validatePlan(plan); + if (!Number.isInteger(maxConcurrent) || maxConcurrent < 1) throw new TypeError('maxConcurrent must be a positive integer'); + const pending = new Map(plan.workers.map((worker) => [worker.id, worker])); + const results = new Map(); + const running = new Map(); + + const start = (worker) => { + const adapter = adapterFor(adapters, worker.host); + const promise = adapter + ? executeWorker(worker, adapter, { cwd, timeoutMs, clock }) + : Promise.resolve(workerFailure(worker, { exitCategory: 'cli_unavailable', failure: { reason: `no execution adapter for host "${worker.host}"` }, clock })); + running.set(worker.id, promise.then((result) => ({ id: worker.id, result }))); + pending.delete(worker.id); + }; + + while (pending.size || running.size) { + for (const worker of pending.values()) { + if (running.size >= maxConcurrent) break; + const deps = worker.dependsOn ?? []; + if (!deps.every((id) => results.has(id))) continue; + const failed = deps.filter((id) => results.get(id).status !== 'succeeded'); + if (failed.length) { + results.set(worker.id, workerFailure(worker, { + status: 'blocked', exitCategory: 'worker_error', failure: { reason: 'dependency_failed', dependencies: failed }, clock, + })); + pending.delete(worker.id); + } else start(worker); + } + if (!running.size) { + if (pending.size) throw new Error('execution plan contains a dependency cycle'); + break; + } + const { id, result } = await Promise.race(running.values()); + running.delete(id); + results.set(id, result); + } + return plan.workers.map((worker) => results.get(worker.id)); +} diff --git a/tests/kit/execution-runner.test.mjs b/tests/kit/execution-runner.test.mjs new file mode 100644 index 00000000..044ec72c --- /dev/null +++ b/tests/kit/execution-runner.test.mjs @@ -0,0 +1,56 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { EXECUTION_ADAPTERS } from '../../src/lib/execution/adapters.mjs'; +import { executeRunPlan } from '../../src/lib/execution/runner.mjs'; + +const worker = (id, host = 'opencode', dependsOn) => ({ id, activity: 'implementation', role: 'coder', host, prompt: id, ...(dependsOn ? { dependsOn } : {}) }); +const clock = () => '2026-07-29T00:00:00.000Z'; + +test('the built-in registry exposes the supervised OpenCode transport only', () => { + assert.equal(EXECUTION_ADAPTERS.get('opencode').id, 'opencode-server'); + assert.equal(EXECUTION_ADAPTERS.has('claude'), false); + assert.equal(EXECUTION_ADAPTERS.has('codex'), false); +}); + +function adapter({ observation = { type: 'idle' }, events = [] } = {}) { + return { + id: 'fake', + async readiness() { events.push('ready'); return { ready: true }; }, + async prepare({ worker: w }) { events.push(`prepare:${w.id}`); return { worker: w }; }, + async launch(state) { events.push(`launch:${state.worker.id}`); return state; }, + async observe(state) { events.push(`observe:${state.worker.id}`); return observation; }, + interpret(state, observed) { + events.push(`interpret:${state.worker.id}:${observed.type}`); + return { + workerId: state.worker.id, activity: state.worker.activity, role: state.worker.role, host: state.worker.host, + status: observed.type === 'timeout' ? 'timed_out' : 'succeeded', exitCategory: observed.type === 'timeout' ? 'timeout' : 'success', + startedAt: clock(), endedAt: clock(), durationMs: 0, provider: null, providerProvenance: 'unknown', + configuredModel: null, observedModel: null, sessionId: null, transcriptRefs: [], failure: null, usage: null, + }; + }, + async cancel(state) { events.push(`cancel:${state.worker.id}`); }, + async cleanup(state) { events.push(`cleanup:${state.worker.id}`); }, + }; +} + +test('runner schedules a dependency DAG and blocks only descendants of a failure', async () => { + const events = []; + const ok = adapter({ events }); + const failed = { ...adapter({ events }), interpret(state) { + return { workerId: state.worker.id, activity: state.worker.activity, role: state.worker.role, host: state.worker.host, + status: 'failed', exitCategory: 'worker_error', startedAt: clock(), endedAt: clock(), durationMs: 0, + provider: null, providerProvenance: 'unknown', configuredModel: null, observedModel: null, sessionId: null, transcriptRefs: [], failure: { reason: 'no' }, usage: null }; + } }; + const results = await executeRunPlan({ workers: [worker('a', 'opencode'), worker('b', 'bad'), worker('c', 'opencode', ['a']), worker('d', 'opencode', ['b'])] }, { + adapters: { opencode: ok, bad: failed }, maxConcurrent: 2, clock, + }); + assert.deepEqual(results.map((result) => result.status), ['succeeded', 'failed', 'succeeded', 'blocked']); + assert.ok(events.includes('launch:c')); + assert.ok(!events.includes('launch:d')); +}); + +test('runner reports an unimplemented host without attempting a lifecycle', async () => { + const [result] = await executeRunPlan({ workers: [worker('a', 'claude')] }, { clock }); + assert.equal(result.exitCategory, 'cli_unavailable'); + assert.match(result.failure.reason, /no execution adapter/); +});