feat(jobs): autopilot-cycle handler honors source_id + pull + archive recheck

Threads the v0.38 per-source dispatch payload through the autopilot-cycle
handler at src/commands/jobs.ts:1146. Closes three codex round-1 findings:

- P0-2 / P1-B: validates job.data.source_id at handler entry via the
  canonical source-id.ts isValidSourceId boolean check. Malformed
  source_id from a queue replay dead-letters with a clear error
  instead of reaching cycle code.

- P1-2: job.data.pull explicit boolean overrides the legacy hardcoded
  `true`, so per-source dispatch for local-only sources can pass
  pull: false (no git network round-trip for sources without remote_url).
  Missing/undefined preserves the legacy true for back-compat with cron/
  launchd callers that don't know about the new field.

- P1-5: archived-source recheck happens BEFORE runCycle is invoked
  (cheap SELECT archived FROM sources WHERE id = $1). If the source was
  archived between fan-out and worker claim, handler returns
  { status: 'skipped', reason: 'source_archived' } cleanly — no lock
  acquired, no phases run, no last_full_cycle_at touched. Same skip
  shape for source_not_found (deleted between dispatch and claim).

7 PGLite integration tests cover all five paths (legacy / valid /
not-found / archived / malformed-source_id / non-string source_id /
pull: false override).

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

src/commands/jobs.ts

@@ -1149,6 +1149,53 @@ export async function registerBuiltinHandlers(worker: MinionWorker, engine: Brai
       ? job.data.repoPath
       : (await engine.getConfig('sync.repo_path')) ?? '.';
 
+    // v0.38 (codex r1 P1-2 + P1-5): per-source dispatch threading.
+    //   - source_id: when set, runCycle uses the per-source lock ID and
+    //     writes last_full_cycle_at on success. Validated at handler entry
+    //     so queue replays with malformed source_id dead-letter instead of
+    //     reaching cycle code.
+    //   - pull: when set, overrides the legacy hardcoded `true` so
+    //     per-source dispatch can disable pull for local-only sources.
+    //     Missing/undefined keeps the legacy `true` for back-compat.
+    //   - Archive recheck: if source_id is set but the source was
+    //     archived between fan-out and worker claim, skip cleanly.
+    const rawSourceId = job.data.source_id;
+    let sourceId: string | undefined;
+    if (rawSourceId !== undefined && rawSourceId !== null) {
+      if (typeof rawSourceId !== 'string') {
+        throw new Error(`autopilot-cycle: invalid source_id (not a string): ${JSON.stringify(rawSourceId)}`);
+      }
+      const { isValidSourceId } = await import('../core/source-id.ts');
+      if (!isValidSourceId(rawSourceId)) {
+        // Dead-letter early — malformed source_id from queue replay shouldn't
+        // reach cycle code. TS narrowing via isValidSourceId boolean shape
+        // (assertValidSourceId would require static-import per TS2775).
+        throw new Error(`autopilot-cycle: invalid source_id (regex): ${JSON.stringify(rawSourceId)}`);
+      }
+      // Archive recheck (codex r1 P1-5): cheap pre-cycle lookup. Returns
+      // immediately if source is gone or archived; runCycle never even
+      // acquires a lock.
+      const rows = await engine.executeRaw<{ archived: boolean | null }>(
+        `SELECT archived FROM sources WHERE id = $1`,
+        [rawSourceId],
+      );
+      if (rows.length === 0) {
+        return {
+          partial: false,
+          status: 'skipped',
+          report: { reason: 'source_not_found', source_id: rawSourceId },
+        };
+      }
+      if (rows[0].archived === true) {
+        return {
+          partial: false,
+          status: 'skipped',
+          report: { reason: 'source_archived', source_id: rawSourceId },
+        };
+      }
+      sourceId = rawSourceId;
+    }
+
     // Allow callers to select phases via job data (e.g. skip embed for
     // fast cycles). Validates against ALL_PHASES to prevent injection.
     const { ALL_PHASES } = await import('../core/cycle.ts');
@@ -1157,10 +1204,14 @@ export async function registerBuiltinHandlers(worker: MinionWorker, engine: Brai
       ? (job.data.phases as string[]).filter(p => validPhases.has(p as any))
       : undefined;
 
+    // Pull default: legacy `true` for back-compat; explicit boolean wins.
+    const pull = typeof job.data.pull === 'boolean' ? job.data.pull : true;
+
     const report = await runCycle(engine, {
       brainDir: repoPath,
-      pull: true, // autopilot daemon opts into git pull
+      pull,
       signal: job.signal, // propagate abort so cycle bails on timeout/cancel
+      ...(sourceId ? { sourceId } : {}),
       ...(requestedPhases && requestedPhases.length > 0 ? { phases: requestedPhases as any } : {}),
       yieldBetweenPhases: async () => {
         // Yield to the event loop so worker lock-renewal can fire.

test/autopilot-cycle-handler.test.ts

@@ -0,0 +1,112 @@
+/**
+ * v0.38 autopilot-cycle handler — source_id validation + archive recheck.
+ *
+ * Covers the codex r1 P1-5 finding: archived-source recheck must happen
+ * before lock acquisition (handler entry, not deep in runPhaseSync).
+ * Also covers codex r2 P1-B (source_id validation at primitive layer).
+ */
+import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
+import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import { MinionWorker } from '../src/core/minions/worker.ts';
+import { registerBuiltinHandlers } from '../src/commands/jobs.ts';
+import { mkdtempSync } from 'fs';
+import { tmpdir } from 'os';
+import { join } from 'path';
+
+let engine: PGLiteEngine;
+let brainDir: string;
+
+beforeAll(async () => {
+  engine = new PGLiteEngine();
+  await engine.connect({ database_url: '' }); // in-memory
+  await engine.initSchema();
+}, 60_000);
+
+afterAll(async () => {
+  await engine.disconnect();
+});
+
+beforeEach(async () => {
+  // Targeted DELETE preserves the `config.version` key that
+  // MinionQueue.ensureSchema requires (full resetPgliteState wipes it).
+  // Same pattern as test/minions.test.ts.
+  await engine.executeRaw('DELETE FROM minion_jobs').catch(() => {});
+  await engine.executeRaw('DELETE FROM gbrain_cycle_locks').catch(() => {});
+  await engine.executeRaw(`DELETE FROM sources WHERE id <> 'default'`).catch(() => {});
+  brainDir = mkdtempSync(join(tmpdir(), 'gbrain-autopilot-handler-'));
+});
+
+async function seedSource(id: string, opts: { archived?: boolean } = {}): Promise<void> {
+  await engine.executeRaw(
+    `INSERT INTO sources (id, name, local_path, config, archived, created_at)
+     VALUES ($1, $2, $3, '{}'::jsonb, $4, NOW())
+     ON CONFLICT (id) DO UPDATE SET archived = EXCLUDED.archived`,
+    [id, id, brainDir, opts.archived === true],
+  );
+}
+
+/**
+ * Invoke the autopilot-cycle handler directly by reaching into the worker's
+ * handler registry. Bypasses the queue lifecycle entirely — we're testing
+ * the handler's logic (source_id validation + archive recheck + pull
+ * threading), not queue mechanics.
+ */
+async function runHandlerOnce(jobData: Record<string, unknown>): Promise<{ partial: boolean; status: string; report: any }> {
+  const worker = new MinionWorker(engine, { concurrency: 1 });
+  await registerBuiltinHandlers(worker, engine);
+  const handler = (worker as unknown as { handlers: Map<string, (j: any) => Promise<any>> }).handlers.get('autopilot-cycle');
+  if (!handler) throw new Error('autopilot-cycle handler not registered');
+  return handler({
+    id: 1,
+    name: 'autopilot-cycle',
+    data: jobData,
+    signal: new AbortController().signal,
+  }) as Promise<{ partial: boolean; status: string; report: any }>;
+}
+
+describe('autopilot-cycle handler source_id validation + archive recheck', () => {
+  test('missing source_id (legacy caller) runs cycle normally', async () => {
+    const result = await runHandlerOnce({ repoPath: brainDir, phases: ['lint'] });
+    // status is whatever runCycle decided; just ensure handler didn't reject
+    expect(['ok', 'clean', 'partial', 'failed', 'skipped']).toContain(result.status);
+  });
+
+  test('valid source_id + existing source runs cycle', async () => {
+    await seedSource('alpha');
+    const result = await runHandlerOnce({ repoPath: brainDir, source_id: 'alpha', phases: ['lint'] });
+    expect(['ok', 'clean']).toContain(result.status);
+  });
+
+  test('source_id pointing at non-existent source returns skipped', async () => {
+    const result = await runHandlerOnce({ repoPath: brainDir, source_id: 'no-such-source', phases: ['lint'] });
+    expect(result.status).toBe('skipped');
+    expect(result.report.reason).toBe('source_not_found');
+  });
+
+  test('source_id pointing at archived source returns skipped (codex P1-5)', async () => {
+    await seedSource('archived-src', { archived: true });
+    const result = await runHandlerOnce({ repoPath: brainDir, source_id: 'archived-src', phases: ['lint'] });
+    expect(result.status).toBe('skipped');
+    expect(result.report.reason).toBe('source_archived');
+  });
+
+  test('malformed source_id (regex fail) throws (codex P1-B)', async () => {
+    await expect(
+      runHandlerOnce({ repoPath: brainDir, source_id: 'BAD ID', phases: ['lint'] }),
+    ).rejects.toThrow(/invalid source_id/);
+  });
+
+  test('non-string source_id throws', async () => {
+    await expect(
+      runHandlerOnce({ repoPath: brainDir, source_id: 42, phases: ['lint'] }),
+    ).rejects.toThrow(/not a string/);
+  });
+
+  test('explicit pull: false overrides default pull: true', async () => {
+    // Behavior check via lack-of-throw + return shape — no actual pull
+    // is invoked because the phase set is just ['lint'].
+    await seedSource('echo');
+    const result = await runHandlerOnce({ repoPath: brainDir, source_id: 'echo', pull: false, phases: ['lint'] });
+    expect(['ok', 'clean']).toContain(result.status);
+  });
+});