From b6e078c97a0fd18a4af67814d3d59490aade08f1 Mon Sep 17 00:00:00 2001
From: Chris Phillipson <christopher.a.phillipson@gmail.com>
Date: Wed, 29 Jul 2026 00:09:21 -0700
Subject: [PATCH] feat: add generalized host run command

---
 bin/agentic-kit.mjs                     |   2 +
 src/commands/run.mjs                    | 103 ++++++++++++++++++++++
 src/lib/execution/adapters.mjs          |   4 +
 src/lib/execution/claude.mjs            |  17 ++++
 src/lib/execution/codex.mjs             |  17 ++++
 src/lib/execution/subprocess.mjs        | 110 ++++++++++++++++++++++++
 tests/kit/cli-help.test.mjs             |   2 +-
 tests/kit/execution-runner.test.mjs     |  10 +--
 tests/kit/run-command.test.mjs          |  19 ++++
 tests/kit/subprocess-execution.test.mjs |  60 +++++++++++++
 10 files changed, 338 insertions(+), 6 deletions(-)
 create mode 100644 src/commands/run.mjs
 create mode 100644 src/lib/execution/claude.mjs
 create mode 100644 src/lib/execution/codex.mjs
 create mode 100644 src/lib/execution/subprocess.mjs
 create mode 100644 tests/kit/run-command.test.mjs
 create mode 100644 tests/kit/subprocess-execution.test.mjs

diff --git a/bin/agentic-kit.mjs b/bin/agentic-kit.mjs
index f2ab65bd..4ff068ff 100755
--- a/bin/agentic-kit.mjs
+++ b/bin/agentic-kit.mjs
@@ -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'),
@@ -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
diff --git a/src/commands/run.mjs b/src/commands/run.mjs
new file mode 100644
index 00000000..fa9f824c
--- /dev/null
+++ b/src/commands/run.mjs
@@ -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;
+}
diff --git a/src/lib/execution/adapters.mjs b/src/lib/execution/adapters.mjs
index ae4365f8..1c442320 100644
--- a/src/lib/execution/adapters.mjs
+++ b/src/lib/execution/adapters.mjs
@@ -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],
 ]));
diff --git a/src/lib/execution/claude.mjs b/src/lib/execution/claude.mjs
new file mode 100644
index 00000000..4a2454b9
--- /dev/null
+++ b/src/lib/execution/claude.mjs
@@ -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();
diff --git a/src/lib/execution/codex.mjs b/src/lib/execution/codex.mjs
new file mode 100644
index 00000000..d0138e08
--- /dev/null
+++ b/src/lib/execution/codex.mjs
@@ -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();
diff --git a/src/lib/execution/subprocess.mjs b/src/lib/execution/subprocess.mjs
new file mode 100644
index 00000000..66efe9ea
--- /dev/null
+++ b/src/lib/execution/subprocess.mjs
@@ -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);
+}
diff --git a/tests/kit/cli-help.test.mjs b/tests/kit/cli-help.test.mjs
index ed5a159f..6bf47e7e 100644
--- a/tests/kit/cli-help.test.mjs
+++ b/tests/kit/cli-help.test.mjs
@@ -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');
diff --git a/tests/kit/execution-runner.test.mjs b/tests/kit/execution-runner.test.mjs
index 044ec72c..09c1f83f 100644
--- a/tests/kit/execution-runner.test.mjs
+++ b/tests/kit/execution-runner.test.mjs
@@ -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 = [] } = {}) {
@@ -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/);
 });
diff --git a/tests/kit/run-command.test.mjs b/tests/kit/run-command.test.mjs
new file mode 100644
index 00000000..2147182a
--- /dev/null
+++ b/tests/kit/run-command.test.mjs
@@ -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/);
+});
diff --git a/tests/kit/subprocess-execution.test.mjs b/tests/kit/subprocess-execution.test.mjs
new file mode 100644
index 00000000..de53c7c3
--- /dev/null
+++ b/tests/kit/subprocess-execution.test.mjs
@@ -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');
+});