feat(orchestrator): in-memory queue + disk persistence (QueueStore)

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

Author: gewenyu99

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

@@ -0,0 +1,146 @@
+import * as fs from 'fs';
+import * as os from 'os';
+import * as path from 'path';
+import {
+  QueueStore,
+  type QueueFile,
+  type TaskHandoff,
+} from '@lib/programs/orchestrator/queue';
+
+function tmpDir(): string {
+  return fs.mkdtempSync(path.join(os.tmpdir(), 'queue-test-'));
+}
+
+describe('QueueStore', () => {
+  let dir: string;
+  let q: QueueStore;
+
+  beforeEach(() => {
+    dir = tmpDir();
+    q = new QueueStore(dir, 'run-1');
+  });
+
+  afterEach(() => {
+    fs.rmSync(dir, { recursive: true, force: true });
+  });
+
+  it('enqueues a pending task with defaults', async () => {
+    const t = await q.enqueue({ type: 'install' });
+    expect(t.status).toBe('pending');
+    expect(t.attempts).toBe(0);
+    expect(t.maxAttempts).toBe(2);
+    expect(t.enqueuedBy).toBe('orchestrator');
+    expect(t.dependsOn).toEqual([]);
+    expect(q.list()).toHaveLength(1);
+  });
+
+  it('only marks a task runnable once its dependencies are done', async () => {
+    const a = await q.enqueue({ type: 'install' });
+    const b = await q.enqueue({ type: 'init', dependsOn: [a.id] });
+
+    expect(q.nextRunnable(10).map((t) => t.id)).toEqual([a.id]);
+
+    await q.start(a.id);
+    await q.complete(a.id);
+    expect(q.nextRunnable(10).map((t) => t.id)).toEqual([b.id]);
+  });
+
+  it('runs independent tasks in parallel, bounded by the cap and in-progress count', async () => {
+    const a = await q.enqueue({ type: 'install' });
+    const b = await q.enqueue({ type: 'init' });
+
+    // Both independent, so both are runnable; the cap bounds how many.
+    expect(
+      q
+        .nextRunnable(10)
+        .map((t) => t.id)
+        .sort(),
+    ).toEqual([a.id, b.id].sort());
+    expect(q.nextRunnable(1)).toHaveLength(1);
+
+    await q.start(a.id);
+    // One slot used, so cap 1 yields nothing more until a finishes.
+    expect(q.nextRunnable(1)).toHaveLength(0);
+    // Cap 2 still has a free slot for b.
+    expect(q.nextRunnable(2).map((t) => t.id)).toEqual([b.id]);
+  });
+
+  it('start increments attempts and supports within-run retry while attempts remain', async () => {
+    const t = await q.enqueue({ type: 'install', maxAttempts: 2 });
+    await q.start(t.id);
+    expect(q.get(t.id)?.attempts).toBe(1);
+
+    await q.fail(t.id, { type: 'API_ERROR', message: 'boom' });
+    expect(q.get(t.id)?.status).toBe('failed');
+
+    // Retry: attempts (1) < maxAttempts (2), so requeue and run again.
+    await q.requeue(t.id);
+    expect(q.get(t.id)?.status).toBe('pending');
+    await q.start(t.id);
+    expect(q.get(t.id)?.attempts).toBe(2);
+  });
+
+  it('completing a task writes and reads back a structured handoff', async () => {
+    const t = await q.enqueue({ type: 'install' });
+    const handoff: TaskHandoff = {
+      goals: 'install the sdk',
+      did: 'added posthog-js',
+      forNextAgent: 'env vars not set yet',
+      filesTouched: ['package.json'],
+    };
+    await q.start(t.id);
+    await q.complete(t.id, handoff);
+
+    expect(q.get(t.id)?.status).toBe('done');
+    expect(q.readHandoff(t.id)).toEqual(handoff);
+    expect(q.readHandoffsByType('install')).toEqual([handoff]);
+  });
+
+  it('is drained when a pending task is blocked by a failed dependency', async () => {
+    const a = await q.enqueue({ type: 'install' });
+    await q.enqueue({ type: 'init', dependsOn: [a.id] });
+
+    expect(q.isDrained()).toBe(false);
+    await q.start(a.id);
+    await q.fail(a.id, { type: 'API_ERROR', message: 'boom' });
+
+    // init can never run now, and nothing is in progress.
+    expect(q.nextRunnable(10)).toHaveLength(0);
+    expect(q.isDrained()).toBe(true);
+  });
+
+  it('reflects the in-memory queue to queue.json on flush', async () => {
+    const a = await q.enqueue({ type: 'install' });
+    await q.start(a.id);
+    await q.complete(a.id);
+    q.flushNow();
+
+    const file = JSON.parse(fs.readFileSync(q.queuePath, 'utf8')) as QueueFile;
+    expect(file.version).toBe(1);
+    expect(file.runId).toBe('run-1');
+    expect(file.tasks).toHaveLength(1);
+    expect(file.tasks[0].status).toBe('done');
+  });
+
+  it('appends a line to audit.jsonl for each transition', async () => {
+    const a = await q.enqueue({ type: 'install' });
+    await q.start(a.id);
+    await q.complete(a.id);
+
+    const lines = fs
+      .readFileSync(q.auditPath, 'utf8')
+      .trim()
+      .split('\n')
+      .map((l) => JSON.parse(l) as { event: string });
+    expect(lines.map((l) => l.event)).toEqual(['enqueue', 'start', 'complete']);
+  });
+
+  it('serializes concurrent enqueues with no lost updates', async () => {
+    await Promise.all(
+      Array.from({ length: 25 }, (_, i) => q.enqueue({ type: `t${i}` })),
+    );
+    expect(q.list()).toHaveLength(25);
+    const ids = new Set(q.list().map((t) => t.id));
+    expect(ids.size).toBe(25);
+  });
+});