feat(orchestrator): executor drain-loop scheduler

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: gewenyu99

src/lib/programs/orchestrator/__tests__/executor.test.ts

@@ -0,0 +1,127 @@
+import * as fs from 'fs';
+import * as os from 'os';
+import * as path from 'path';
+import {
+  QueueStore,
+  type QueuedTask,
+  type TaskHandoff,
+} from '@lib/programs/orchestrator/queue';
+import { drainQueue, type RunTask } from '@lib/programs/orchestrator/executor';
+
+const HANDOFF: TaskHandoff = { goals: 'g', did: 'd', forNextAgent: 'n' };
+
+function tmpDir(): string {
+  return fs.mkdtempSync(path.join(os.tmpdir(), 'executor-test-'));
+}
+
+describe('drainQueue', () => {
+  let dir: string;
+  let q: QueueStore;
+
+  beforeEach(() => {
+    dir = tmpDir();
+    q = new QueueStore(dir, 'run-1');
+  });
+
+  afterEach(() => fs.rmSync(dir, { recursive: true, force: true }));
+
+  const completing: RunTask = async (task) => {
+    await q.complete(task.id, HANDOFF);
+  };
+
+  it('runs a single task to done and drains', async () => {
+    const a = await q.enqueue({ type: 'install' });
+    await drainQueue(q, completing, { cap: 1, maxIterations: 50 });
+    expect(q.get(a.id)?.status).toBe('done');
+    expect(q.isDrained()).toBe(true);
+  });
+
+  it('runs a dependent task only after its dependency completes', async () => {
+    const order: string[] = [];
+    const a = await q.enqueue({ type: 'install' });
+    const b = await q.enqueue({ type: 'init', dependsOn: [a.id] });
+    const runner: RunTask = async (task) => {
+      order.push(task.type);
+      await q.complete(task.id, HANDOFF);
+    };
+    await drainQueue(q, runner, { cap: 1, maxIterations: 50 });
+    expect(order).toEqual(['install', 'init']);
+    expect(q.get(b.id)?.status).toBe('done');
+  });
+
+  it('retries a task that ends without reporting, then fails it', async () => {
+    const a = await q.enqueue({ type: 'install', maxAttempts: 2 });
+    const noReport: RunTask = async () => {
+      /* agent never calls complete_task */
+    };
+    await drainQueue(q, noReport, { cap: 1, maxIterations: 50 });
+    expect(q.get(a.id)?.status).toBe('failed');
+    expect(q.get(a.id)?.attempts).toBe(2);
+  });
+
+  it('succeeds on a retry within the attempt budget', async () => {
+    let calls = 0;
+    const a = await q.enqueue({ type: 'install', maxAttempts: 3 });
+    const flaky: RunTask = async (task: QueuedTask) => {
+      calls += 1;
+      if (calls >= 2) await q.complete(task.id, HANDOFF);
+    };
+    await drainQueue(q, flaky, { cap: 1, maxIterations: 50 });
+    expect(q.get(a.id)?.status).toBe('done');
+    expect(calls).toBe(2);
+  });
+
+  it('does not run a task whose dependency failed', async () => {
+    const a = await q.enqueue({ type: 'install', maxAttempts: 1 });
+    const b = await q.enqueue({ type: 'init', dependsOn: [a.id] });
+    const runner: RunTask = async (task) => {
+      if (task.type === 'init') await q.complete(task.id, HANDOFF);
+      // install never reports, so it fails after its single attempt.
+    };
+    await drainQueue(q, runner, { cap: 1, maxIterations: 50 });
+    expect(q.get(a.id)?.status).toBe('failed');
+    expect(q.get(b.id)?.status).toBe('pending');
+    expect(q.isDrained()).toBe(true);
+  });
+
+  it('respects the concurrency cap', async () => {
+    let active = 0;
+    let maxActive = 0;
+    const runner: RunTask = async (task) => {
+      active += 1;
+      maxActive = Math.max(maxActive, active);
+      await new Promise((r) => setTimeout(r, 5));
+      await q.complete(task.id, HANDOFF);
+      active -= 1;
+    };
+    await q.enqueue({ type: 'install' });
+    await q.enqueue({ type: 'init' });
+    await drainQueue(q, runner, { cap: 2, maxIterations: 50 });
+    expect(maxActive).toBe(2);
+  });
+
+  it('runs independent tasks one at a time at cap 1', async () => {
+    let active = 0;
+    let maxActive = 0;
+    const runner: RunTask = async (task) => {
+      active += 1;
+      maxActive = Math.max(maxActive, active);
+      await new Promise((r) => setTimeout(r, 5));
+      await q.complete(task.id, HANDOFF);
+      active -= 1;
+    };
+    await q.enqueue({ type: 'install' });
+    await q.enqueue({ type: 'init' });
+    await drainQueue(q, runner, { cap: 1, maxIterations: 50 });
+    expect(maxActive).toBe(1);
+  });
+
+  it('terminates via the iteration backstop instead of looping forever', async () => {
+    const a = await q.enqueue({ type: 'install', maxAttempts: 999 });
+    const neverReports: RunTask = async () => {
+      /* would retry forever without the backstop */
+    };
+    await drainQueue(q, neverReports, { cap: 1, maxIterations: 3 });
+    expect(q.get(a.id)?.attempts).toBeLessThanOrEqual(3);
+  });
+});

src/lib/programs/orchestrator/executor.ts

@@ -0,0 +1,92 @@
+/**
+ * The executor drains the queue. It is the scheduler: it picks runnable tasks
+ * (dependencies satisfied), respects the concurrency cap, runs each through a
+ * `runTask` function, and reads the outcome the task reported through
+ * `complete_task`. A task that ends without reporting is retried while attempts
+ * remain, then failed. A `maxIterations` backstop guarantees termination.
+ *
+ * The drain loop is independent of how a task actually runs. `runTask` is
+ * injected: the real one spins up a fresh agent, the tests use a fake.
+ */
+import type { QueueStore, QueuedTask } from './queue';
+
+/** Per-task agent configuration the resolver produces from a task's type. */
+export interface ResolvedTask {
+  model: string;
+  allowedTools: readonly string[];
+  disallowedTools: readonly string[];
+  prompt: string;
+}
+
+/** Resolves a queued task to what the agent needs. The real one is markdown-backed. */
+export type TaskResolver = (
+  task: QueuedTask,
+  store: QueueStore,
+) => ResolvedTask;
+
+/** Runs one task's agent. It is expected to drive the task to a terminal state
+ *  (via the task agent calling complete_task). */
+export type RunTask = (task: QueuedTask) => Promise<void>;
+
+export interface DrainOptions {
+  /** Max tasks running at once. Default 1 (sequential) for the prototype. */
+  cap: number;
+  /** Backstop against a pathological always-one-more-pending loop. */
+  maxIterations: number;
+}
+
+export const DEFAULT_DRAIN_OPTIONS: DrainOptions = {
+  cap: 1,
+  maxIterations: 200,
+};
+
+async function runOne(
+  store: QueueStore,
+  runTask: RunTask,
+  task: QueuedTask,
+): Promise<void> {
+  await store.start(task.id);
+  try {
+    await runTask(task);
+  } catch {
+    // The task threw rather than reporting. The outcome check below handles it.
+  }
+
+  const after = store.get(task.id);
+  if (!after) return;
+
+  if (after.status === 'in_progress') {
+    // The agent ended without calling complete_task. Retry or fail.
+    if (after.attempts < after.maxAttempts) {
+      await store.requeue(task.id);
+    } else {
+      await store.fail(task.id, {
+        type: 'no-report',
+        message: 'Task ended without calling complete_task.',
+      });
+    }
+    return;
+  }
+
+  if (after.status === 'failed' && after.attempts < after.maxAttempts) {
+    await store.requeue(task.id);
+  }
+}
+
+/**
+ * Drain the queue to a terminal state. Returns when every task is done, failed,
+ * or blocked by a failed dependency, or when the iteration backstop trips.
+ */
+export async function drainQueue(
+  store: QueueStore,
+  runTask: RunTask,
+  opts: DrainOptions = DEFAULT_DRAIN_OPTIONS,
+): Promise<void> {
+  let iterations = 0;
+  while (!store.isDrained()) {
+    if (++iterations > opts.maxIterations) break;
+    const runnable = store.nextRunnable(opts.cap);
+    if (runnable.length === 0) break;
+    await Promise.all(runnable.map((t) => runOne(store, runTask, t)));
+  }
+}