Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions bin/agentic-kit.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ const PORCELAIN = Object.assign(Object.create(null), {
setup: () => import('../src/commands/setup.mjs'),
dashboard: () => import('../src/commands/x/dashboard.mjs'),
admin: () => import('../src/commands/x/admin.mjs'),
run: () => import('../src/commands/run.mjs'),
dual: () => import('../src/commands/dual.mjs'),
host: () => import('../src/commands/x/provider.mjs'),
provider: () => import('../src/commands/x/provider.mjs'),
Expand Down Expand Up @@ -49,6 +50,7 @@ Usage (ak = alias of agentic-kit):
ak sync converge to good: upgrade + heal + verify [--dry-run] [--no-upgrade]
ak dashboard open the local web dashboard (localhost; auto-opens browser) [--port N] [--no-open]
ak admin maintainer-only telemetry admin (localhost; GitHub/npm egress) [--port N] [--no-open]
ak run execute a host-neutral activity pipeline [template "task"] [--dry-run]
ak dual run a Claude+Codex collaboration swarm (dual-host) [run <template> "task"] [--dry-run]
ak host manage agent hosts, routing, and provider bindings [status|pick|refresh|off]
ak provider deprecated alias for ak host; removed before the stable release
Expand Down
103 changes: 103 additions & 0 deletions src/commands/run.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
// ak run — host-neutral execution of the managed activity routing plan. `ak dual`
// remains the compatibility adapter for claude-flow-codex collaboration pipelines.
import { loadKitConfig } from '../lib/config.mjs';
import { fail, ok, dim, bold } from '../lib/output.mjs';
import { executeRunPlan } from '../lib/execution/runner.mjs';
import { DUAL_RUN_TEMPLATE_NAMES, materializeRunPlan, parseRouteSpecs } from '../lib/routing.mjs';

export const options = {
route: { type: 'string', multiple: true },
'dry-run': { type: 'boolean', default: false },
'max-concurrent': { type: 'string' },
timeout: { type: 'string' },
json: { type: 'boolean', default: false },
};

export const help = `ak run — execute a host-neutral activity pipeline

Materializes the managed per-activity routing policy and runs each worker through
its host adapter. OpenCode is accepted only after its routing capability is enabled.

Usage:
ak run <template> "<task>"

Templates: ${DUAL_RUN_TEMPLATE_NAMES.join(', ')}

Options:
--route 'act:host[:model]' per-run routing override (repeatable; not persisted)
--dry-run print the host-neutral execution plan only
--max-concurrent <n> max concurrent workers (default 4)
--timeout <ms> per-worker timeout (default 120000)
--json emit machine-readable plan/results

Examples:
ak run feature "add token-bucket rate limiting" --dry-run
ak run security "src/auth/" --route 'security-scan:opencode'`;

function positiveInt(value, name) {
if (value === undefined) return undefined;
if (!/^\d+$/.test(value) || Number(value) < 1) throw new TypeError(`${name} must be a positive integer`);
return Number(value);
}

export function buildRunPlan(cfg, template, task, routeFlags = []) {
let policy = { ...(cfg.providers?.dualRouting ?? {}) };
if (routeFlags.length) {
const { policy: overrides, warnings } = parseRouteSpecs(routeFlags);
policy = { ...policy, ...overrides };
return { plan: materializeRunPlan(policy, { template, task }), warnings };
}
return { plan: materializeRunPlan(policy, { template, task }), warnings: [] };
}

function printPlan(plan) {
console.log(bold(`run: ${plan.template}`));
for (const worker of plan.workers) {
const dependency = worker.dependsOn?.length ? `after ${worker.dependsOn.join(', ')}` : 'start';
console.log(` ${worker.id.padEnd(12)} ${worker.host.padEnd(9)} ${(worker.configuredModel ?? '').padEnd(24)} ${dim(dependency)}`);
}
}

function printResults(results) {
for (const result of results) {
const detail = result.failure?.reason ? ` — ${result.failure.reason}` : '';
console.log(` ${result.workerId.padEnd(12)} ${result.host.padEnd(9)} ${result.status} (${result.exitCategory})${dim(detail)}`);
}
}

export async function run({ flags, positionals }) {
const template = positionals[0];
const task = positionals.slice(1).join(' ').trim();
if (!template || !DUAL_RUN_TEMPLATE_NAMES.includes(template)) {
fail(`unknown template "${template ?? ''}" — expected: ${DUAL_RUN_TEMPLATE_NAMES.join(', ')}`);
return 2;
}
if (!task) { fail('a task description is required: ak run <template> "<task>"'); return 2; }
let plan;
let warnings;
try {
({ plan, warnings } = buildRunPlan(loadKitConfig(), template, task, flags.route ?? []));
} catch (error) {
fail(error.message);
return 2;
}
for (const warning of warnings) console.error(`warning: ${warning}`);
if (flags['dry-run']) {
if (flags.json) console.log(JSON.stringify({ plan }, null, 2));
else printPlan(plan);
return 0;
}
let maxConcurrent;
let timeoutMs;
try {
maxConcurrent = positiveInt(flags['max-concurrent'], 'max-concurrent');
timeoutMs = positiveInt(flags.timeout, 'timeout');
} catch (error) { fail(error.message); return 2; }
if (!flags.json) printPlan(plan);
const results = await executeRunPlan(plan, { maxConcurrent, timeoutMs });
if (flags.json) console.log(JSON.stringify({ plan, results }, null, 2));
else printResults(results);
if (results.every((result) => result.status === 'succeeded')) { ok('run complete'); return 0; }
fail('run finished with one or more non-successful workers');
return 1;
}
4 changes: 4 additions & 0 deletions src/lib/execution/adapters.mjs
Original file line number Diff line number Diff line change
@@ -1,7 +1,11 @@
// 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';
import { CLAUDE_EXECUTION_ADAPTER } from './claude.mjs';
import { CODEX_EXECUTION_ADAPTER } from './codex.mjs';

export const EXECUTION_ADAPTERS = Object.freeze(new Map([
['claude', CLAUDE_EXECUTION_ADAPTER],
['codex', CODEX_EXECUTION_ADAPTER],
['opencode', OPENCODE_EXECUTION_ADAPTER],
]));
17 changes: 17 additions & 0 deletions src/lib/execution/claude.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
import { createSubprocessExecutionAdapter } from './subprocess.mjs';

/** Claude Code's documented print/json mode. No permission bypass is passed. */
/** @param {Omit<Parameters<typeof createSubprocessExecutionAdapter>[0], 'id'|'host'|'command'|'argumentsFor'>} [options] */
export function createClaudeExecutionAdapter(options = {}) {
return createSubprocessExecutionAdapter({
id: 'claude-print-json', host: 'claude', command: 'claude',
argumentsFor: (worker) => [
'--print', '--output-format', 'json',
...(worker.configuredModel ? ['--model', worker.configuredModel] : []),
worker.prompt,
],
...options,
});
}

export const CLAUDE_EXECUTION_ADAPTER = createClaudeExecutionAdapter();
17 changes: 17 additions & 0 deletions src/lib/execution/codex.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
import { createSubprocessExecutionAdapter } from './subprocess.mjs';

/** Codex's documented exec/json mode. Its configured sandbox policy is retained. */
/** @param {Omit<Parameters<typeof createSubprocessExecutionAdapter>[0], 'id'|'host'|'command'|'argumentsFor'>} [options] */
export function createCodexExecutionAdapter(options = {}) {
return createSubprocessExecutionAdapter({
id: 'codex-exec-json', host: 'codex', command: 'codex',
argumentsFor: (worker, cwd) => [
'exec', '--json', '--cd', cwd,
...(worker.configuredModel ? ['--model', worker.configuredModel] : []),
worker.prompt,
],
...options,
});
}

export const CODEX_EXECUTION_ADAPTER = createCodexExecutionAdapter();
110 changes: 110 additions & 0 deletions src/lib/execution/subprocess.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
// Supervised subprocess adapters for the hosts whose native CLIs already offer
// non-interactive structured output. The runner owns timeout policy; this module
// owns only one direct child process and never invokes a shell or bypass mode.
import { spawn as nodeSpawn } from 'node:child_process';
import { have } from '../exec.mjs';
import { validateExecutionAdapter, validateWorkerResult } from './schema.mjs';

const nowIso = () => new Date().toISOString();
const OUTPUT_LIMIT = 256 * 1024;

function capture(stream) {
let text = '';
stream?.on?.('data', (chunk) => { text = `${text}${String(chunk)}`.slice(-OUTPUT_LIMIT); });
return () => text;
}

function waitForChild(child, stdout, stderr) {
return new Promise((resolve) => {
let settled = false;
const finish = (result) => {
if (settled) return;
settled = true;
resolve({ ...result, stdout: stdout(), stderr: stderr() });
};
child.once?.('error', (error) => finish({ code: null, signal: null, error }));
child.once?.('close', (code, signal) => finish({ code, signal, error: null }));
});
}

function categoryFor(completion) {
const detail = `${completion.stderr}\n${completion.stdout}\n${completion.error?.message ?? ''}`;
if (/auth(?:entication)?|login|required api key|unauthorized/i.test(detail)) return 'auth_required';
if (/model .*(not found|unavailable)|unknown model|model_not_found/i.test(detail)) return 'model_unavailable';
return 'worker_error';
}

function failureFor(completion) {
const detail = (completion.stderr || completion.stdout || completion.error?.message || 'host process failed').trim();
return { reason: detail.slice(0, 240) };
}

function resultFor(state, observation, host, clock) {
const endedAt = clock();
const durationMs = Math.max(0, Date.parse(endedAt) - Date.parse(state.startedAt));
const terminal = observation?.type;
let status = 'succeeded';
let exitCategory = 'success';
let failure = null;
if (terminal === 'timeout') {
status = 'timed_out'; exitCategory = 'timeout'; failure = { reason: 'timeout' };
} else if (terminal === 'cancelled') {
status = 'cancelled'; exitCategory = 'cancelled'; failure = { reason: 'cancelled' };
} else if (terminal === 'orphaned') {
status = 'failed'; exitCategory = 'orphaned'; failure = { reason: 'host process did not terminate' };
} else if (!observation || observation.code !== 0 || observation.error) {
status = 'failed'; exitCategory = categoryFor(observation ?? {}); failure = failureFor(observation ?? {});
}
return validateWorkerResult({
workerId: state.worker.id, activity: state.worker.activity, role: state.worker.role, host,
status, exitCategory, startedAt: state.startedAt, endedAt, durationMs,
provider: host, providerProvenance: 'configured', configuredModel: state.worker.configuredModel ?? null,
observedModel: null, sessionId: null, transcriptRefs: [], failure, usage: null,
});
}

async function terminate(state) {
if (state.finished) return { type: 'cancelled' };
try {
if (!state.child?.kill?.('SIGTERM')) return { type: 'orphaned' };
return { type: 'cancelled' };
} catch { return { type: 'orphaned' }; }
}

/** Build a lifecycle adapter for one host's structured non-interactive CLI.
* `argumentsFor` must return a fixed argv vector; prompts never pass through a
* shell. Permission modes are deliberately absent from this generic layer.
* @param {{id:string, host:string, command:string, argumentsFor:(worker:any, cwd:string)=>string[],
* spawnFn?:typeof nodeSpawn, haveFn?:typeof have, clock?:()=>string}} options */
export function createSubprocessExecutionAdapter({
id, host, command, argumentsFor, spawnFn = nodeSpawn, haveFn = have, clock = nowIso,
} = /** @type {any} */ ({})) {
if (!id || !host || !command || typeof argumentsFor !== 'function') throw new TypeError('subprocess adapter requires id, host, command, and argumentsFor');
const adapter = {
id,
async readiness() {
const installed = await haveFn(command);
return installed ? { ready: true } : { ready: false, exitCategory: 'cli_unavailable' };
},
async prepare({ worker, cwd = process.cwd() } = /** @type {{worker?:any, cwd?:string}} */ ({})) {
if (worker?.host !== host) throw new TypeError(`${host} adapter requires a ${host} worker`);
if (typeof worker?.prompt !== 'string' || !worker.prompt.trim()) throw new TypeError(`${host} worker.prompt is required`);
return { worker, cwd, args: argumentsFor(worker, cwd), startedAt: clock() };
},
async launch(state) {
const child = spawnFn(command, state.args, { cwd: state.cwd, env: process.env, stdio: ['ignore', 'pipe', 'pipe'] });
if (!child?.once) throw new Error(`${host} process did not expose child lifecycle events`);
const stdout = capture(child.stdout);
const stderr = capture(child.stderr);
const completion = waitForChild(child, stdout, stderr);
const next = { ...state, child, completion, finished: false };
completion.then(() => { next.finished = true; });
return next;
},
async observe(state) { return state.completion; },
interpret(state, observation) { return resultFor(state, observation, host, clock); },
async cancel(state) { return terminate(state); },
async cleanup(state) { return terminate(state); },
};
return validateExecutionAdapter(adapter);
}
2 changes: 1 addition & 1 deletion tests/kit/cli-help.test.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ test('mutating commands intercept both --help and -h before running', () => {
});

test('every command exposes an Examples section in its help', () => {
for (const cmd of [['setup'], ['status'], ['sync'], ['dual'], ['dashboard'], ['uninstall'],
for (const cmd of [['setup'], ['status'], ['sync'], ['run'], ['dual'], ['dashboard'], ['uninstall'],
['host'], ['provider'], ['x', 'mcp'], ['x', 'host'], ['x', 'provider'],
['x', 'verify'], ['x', 'reference'], ['x', 'daemon-gc']]) {
const r = ak(...cmd, '--help');
Expand Down
10 changes: 5 additions & 5 deletions tests/kit/execution-runner.test.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,10 @@ 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', () => {
test('the built-in registry exposes supervised transports for every managed host', () => {
assert.equal(EXECUTION_ADAPTERS.get('claude').id, 'claude-print-json');
assert.equal(EXECUTION_ADAPTERS.get('codex').id, 'codex-exec-json');
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 = [] } = {}) {
Expand Down Expand Up @@ -49,8 +49,8 @@ test('runner schedules a dependency DAG and blocks only descendants of a failure
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 });
test('runner reports an unknown host without attempting a lifecycle', async () => {
const [result] = await executeRunPlan({ workers: [worker('a', 'unknown')] }, { clock });
assert.equal(result.exitCategory, 'cli_unavailable');
assert.match(result.failure.reason, /no execution adapter/);
});
19 changes: 19 additions & 0 deletions tests/kit/run-command.test.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { buildRunPlan } from '../../src/commands/run.mjs';
import { seedDualRouting } from '../../src/lib/routing.mjs';

test('ak run materializes the host-neutral plan and keeps run-local route overrides ephemeral', () => {
const cfg = { providers: { dualRouting: seedDualRouting() } };
const { plan } = buildRunPlan(cfg, 'feature', 'add a queue', ['implementation:claude:claude-sonnet-5']);
assert.equal(plan.template, 'feature');
assert.equal(plan.workers.find((entry) => entry.id === 'coder').host, 'claude');
assert.equal(cfg.providers.dualRouting.implementation.host, 'codex');
});

test('ak run rejects an OpenCode route until the capability proof is complete', () => {
const cfg = { providers: { dualRouting: { ...seedDualRouting(), 'security-scan': {
host: 'opencode', model: 'openrouter/example', source: 'user',
} } } };
assert.throws(() => buildRunPlan(cfg, 'security', 'src/auth'), /canRouteActivities/);
});
60 changes: 60 additions & 0 deletions tests/kit/subprocess-execution.test.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
import { EventEmitter } from 'node:events';
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { createClaudeExecutionAdapter } from '../../src/lib/execution/claude.mjs';
import { createCodexExecutionAdapter } from '../../src/lib/execution/codex.mjs';
import { executeWorker } from '../../src/lib/execution/runner.mjs';

const worker = (host, model = 'model-1') => ({
id: `${host}-1`, activity: 'implementation', role: 'coder', host, configuredModel: model, prompt: 'Do the work.',
});
const clock = () => '2026-07-29T00:00:00.000Z';

function child({ code = 0, stderr = '' } = {}) {
const result = new EventEmitter();
result.stdout = new EventEmitter();
result.stderr = new EventEmitter();
result.kill = () => { queueMicrotask(() => result.emit('close', null, 'SIGTERM')); return true; };
queueMicrotask(() => {
if (stderr) result.stderr.emit('data', stderr);
result.emit('close', code, null);
});
return result;
}

test('Claude adapter uses print/json mode without a permission bypass', async () => {
const calls = [];
const adapter = createClaudeExecutionAdapter({
haveFn: async () => true, clock,
spawnFn: (command, args, options) => { calls.push({ command, args, options }); return child(); },
});
const result = await executeWorker(worker('claude'), adapter, { cwd: process.cwd(), clock });
assert.equal(result.status, 'succeeded');
assert.deepEqual(calls[0].args, ['--print', '--output-format', 'json', '--model', 'model-1', 'Do the work.']);
assert.ok(!calls[0].args.some((arg) => arg.includes('dangerously') || arg.includes('bypass')));
});

test('Codex adapter pins its supplied workspace and normalizes host failures', async () => {
const calls = [];
const adapter = createCodexExecutionAdapter({
haveFn: async () => true, clock,
spawnFn: (command, args, options) => { calls.push({ command, args, options }); return child({ code: 1, stderr: 'model not found' }); },
});
const result = await executeWorker(worker('codex'), adapter, { cwd: '/workspace/fixture', clock });
assert.equal(result.status, 'failed');
assert.equal(result.exitCategory, 'model_unavailable');
assert.deepEqual(calls[0].args, ['exec', '--json', '--cd', '/workspace/fixture', '--model', 'model-1', 'Do the work.']);
assert.ok(!calls[0].args.some((arg) => arg.includes('dangerously') || arg.includes('bypass')));
});

test('subprocess cleanup terminates a still-running direct child', async () => {
const result = new EventEmitter();
result.stdout = new EventEmitter();
result.stderr = new EventEmitter();
let signal = null;
result.kill = (value) => { signal = value; queueMicrotask(() => result.emit('close', null, value)); return true; };
const adapter = createClaudeExecutionAdapter({ haveFn: async () => true, clock, spawnFn: () => result });
const state = await adapter.launch(await adapter.prepare({ worker: worker('claude'), cwd: process.cwd() }));
await adapter.cleanup(state);
assert.equal(signal, 'SIGTERM');
});
Loading