feat(autopilot): clamp fan-out to worker concurrency + doctor warning (#2194)

Fan-out resolved to 4 (Postgres) regardless of worker --concurrency, so surplus
cycles queued behind the worker and raced the stalled-sweeper. Two fixes for the
same mismatch:
- resolveEffectiveFanoutMax clamps to max(1, concurrency-1) (reserve a slot),
  gated on a LIVE DB-lock holder so a stale started-audit row can't shrink
  throughput (codex #9/D5); no live holder → unknown → unclamped base. Escape
  hatch autopilot.fanout_clamp_to_concurrency.
- doctor's autopilot_fanout_concurrency check warns when fan-out exceeds
  effective slots — the misconfig was silent before. Advisory (started-event
  concurrency), wired into both doctor surfaces.

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

src/commands/autopilot-fanout.ts

@@ -83,6 +83,62 @@ export async function resolveFanoutMax(engine: BrainEngine): Promise<number> {
   return engine.kind === 'pglite' ? 1 : 4;
 }
 
+/**
+ * Read the worker concurrency the supervisor most recently STARTED with, from
+ * its `started` audit event (the lowest-coupling source — no extra lock-row
+ * column). Filesystem read; returns null when no supervisor has ever started
+ * (or the event lacks concurrency). Filtered by queue so a `shell`-queue
+ * supervisor's concurrency doesn't leak into the `default`-queue decision.
+ *
+ * ADVISORY use only (doctor warning). Behavior-changing callers (the fanout
+ * clamp) must additionally gate on a LIVE supervisor — see
+ * resolveEffectiveFanoutMax — because a stale `started` row can otherwise
+ * shrink fan-out for a supervisor that isn't running that config (codex #9/D5).
+ */
+export async function readSupervisorConcurrency(queue = 'default'): Promise<number | null> {
+  try {
+    const { readSupervisorEvents } = await import('../core/minions/handlers/supervisor-audit.ts');
+    const events = readSupervisorEvents({ sinceMs: 24 * 60 * 60 * 1000 });
+    const started = events
+      .filter((e) => e.event === 'started' && (e.queue === undefined || e.queue === queue))
+      .pop();
+    const c = started?.concurrency;
+    return typeof c === 'number' && Number.isFinite(c) ? c : null;
+  } catch {
+    return null;
+  }
+}
+
+/**
+ * Resolve fanoutMax CLAMPED to the worker's effective concurrency (#2194 fix #1).
+ *
+ * Fanning out more cycles than the worker can run guarantees waiters that then
+ * race the stalled-sweeper. Clamp to `max(1, concurrency - 1)` — reserving ≥1
+ * slot for targeted sync/embed jobs that share the `default` queue.
+ *
+ * codex #9 / D5: the clamp is BEHAVIOR-changing, so it trusts only a
+ * proven-alive supervisor (live DB-lock holder, `ttl_expires_at`-gated). With
+ * no live holder the concurrency is UNKNOWN and we fall back to the unclamped
+ * default (4 pg / 1 pglite) — the safe direction (never starve on stale data).
+ * Operators can disable the clamp via `autopilot.fanout_clamp_to_concurrency`.
+ */
+export async function resolveEffectiveFanoutMax(engine: BrainEngine, queue = 'default'): Promise<number> {
+  const base = await resolveFanoutMax(engine);
+  const clampCfg = await engine.getConfig('autopilot.fanout_clamp_to_concurrency');
+  if (clampCfg === 'false' || clampCfg === '0') return base; // operator opt-out
+  try {
+    const { inspectLock, isLockHolderLive } = await import('../core/db-lock.ts');
+    const { supervisorLockId, SUPERVISOR_LOCK_TTL_MIN } = await import('../core/minions/supervisor.ts');
+    const snap = await inspectLock(engine, supervisorLockId(queue));
+    if (!snap || !isLockHolderLive(snap, SUPERVISOR_LOCK_TTL_MIN)) return base; // no live holder → unknown → no clamp
+    const concurrency = await readSupervisorConcurrency(queue);
+    if (concurrency === null) return base;
+    return Math.max(1, Math.min(base, concurrency - 1));
+  } catch {
+    return base;
+  }
+}
+
 /**
  * Read `last_full_cycle_at` ISO string from a source's config JSONB.
  * Returns null when missing or unparseable. Pure function over the row

src/commands/autopilot.ts

@@ -871,8 +871,12 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
           // codex P1-3). Fresh-install brains with no sources rows fall
           // back to the legacy single autopilot-cycle so existing
           // behavior is preserved.
-          const { dispatchPerSource, resolveFanoutMax } = await import('./autopilot-fanout.ts');
-          const fanoutMax = await resolveFanoutMax(engine);
+          const { dispatchPerSource, resolveEffectiveFanoutMax } = await import('./autopilot-fanout.ts');
+          // #2194 fix #1: clamp fan-out to the worker's effective concurrency
+          // (reserve ≥1 slot), gated on a LIVE supervisor so a stale audit row
+          // can't shrink throughput (codex #9/D5). autopilot-cycle jobs run on
+          // the 'default' queue, so that's the concurrency we compare against.
+          const fanoutMax = await resolveEffectiveFanoutMax(engine, 'default');
           const result = await dispatchPerSource(engine, queue, {
             repoPath,
             slot,

src/commands/doctor.ts

@@ -695,6 +695,9 @@ export async function doctorReportRemote(engine: BrainEngine): Promise<DoctorRep
   // issue #1801 — wedged_queue (cross-surface parity with buildChecks).
   checks.push(await computeWedgedQueueCheck(engine));
 
+  // #2194 fix #5 — warn when autopilot fan-out exceeds worker concurrency.
+  checks.push(await computeAutopilotFanoutConcurrencyCheck(engine));
+
   // v0.41 Bug 2 / Eng D8 — subagent_health surfaces rate-lease pressure to the operator.
   checks.push(await checkSubagentHealth(engine));
 
@@ -1563,6 +1566,49 @@ export async function computeWedgedQueueCheck(engine: BrainEngine): Promise<Chec
   }
 }
 
+/**
+ * #2194 fix #5: warn when autopilot's per-tick fan-out exceeds the worker's
+ * effective concurrency. Fanning out more cycles than there are worker slots
+ * guarantees waiters that race the stalled-sweeper — a silent misconfig today.
+ * Advisory (started-event concurrency is fine here; the behavior-changing clamp
+ * in resolveEffectiveFanoutMax is the one that gates on liveness). Surfaces only
+ * when a supervisor has actually started (no noise on never-supervised brains).
+ */
+export async function computeAutopilotFanoutConcurrencyCheck(engine: BrainEngine): Promise<Check> {
+  if (engine.kind !== 'postgres') {
+    return { name: 'autopilot_fanout_concurrency', status: 'ok', message: 'PGLite — single-writer, fan-out is 1' };
+  }
+  try {
+    const { resolveFanoutMax, readSupervisorConcurrency } = await import('./autopilot-fanout.ts');
+    const concurrency = await readSupervisorConcurrency('default');
+    if (concurrency === null) {
+      return { name: 'autopilot_fanout_concurrency', status: 'ok', message: 'No supervisor observed — skipping fan-out/concurrency check' };
+    }
+    const fanoutMax = await resolveFanoutMax(engine);
+    const effectiveSlots = Math.max(1, concurrency - 1);
+    if (fanoutMax > effectiveSlots) {
+      return {
+        name: 'autopilot_fanout_concurrency',
+        status: 'warn',
+        message:
+          `autopilot fan-out (${fanoutMax}/tick) exceeds worker concurrency (${concurrency}). ` +
+          `Surplus cycles queue behind the worker and race the stalled-sweeper. ` +
+          `Lower fan-out: \`gbrain config set autopilot.fanout_max_per_tick ${effectiveSlots}\`, ` +
+          `or raise the supervisor's \`--concurrency\` to ${fanoutMax + 1}. ` +
+          `(The clamp in autopilot does this automatically unless disabled.)`,
+        details: { fanout_max: fanoutMax, concurrency, effective_slots: effectiveSlots },
+      };
+    }
+    return {
+      name: 'autopilot_fanout_concurrency',
+      status: 'ok',
+      message: `fan-out ${fanoutMax}/tick within worker concurrency ${concurrency}`,
+    };
+  } catch (e) {
+    return { name: 'autopilot_fanout_concurrency', status: 'ok', message: `Skipped (${e instanceof Error ? e.message : String(e)})` };
+  }
+}
+
 export async function checkBatchRetryHealth(_engine: BrainEngine): Promise<Check> {
   try {
     // Codex M-10: surface bad env config at doctor time.
@@ -7186,6 +7232,9 @@ export async function buildChecks(
     // waiting, zero live-lock active, stale completions) as a health error.
     progress.heartbeat('wedged_queue');
     checks.push(await computeWedgedQueueCheck(engine));
+    // #2194 fix #5 — autopilot fan-out vs worker concurrency mismatch.
+    progress.heartbeat('autopilot_fanout_concurrency');
+    checks.push(await computeAutopilotFanoutConcurrencyCheck(engine));
     // v0.40.4 graph_signals_coverage — global inbound-link density when
     // graph_signals is enabled in the active mode bundle.
     progress.heartbeat('graph_signals_coverage');

test/autopilot-fanout-clamp.test.ts

@@ -0,0 +1,110 @@
+/**
+ * #2194 fix #1 (codex #9 / D5): resolveEffectiveFanoutMax clamps the per-tick
+ * fan-out to the worker's effective concurrency (max(1, concurrency-1),
+ * reserving ≥1 slot) — but ONLY when a LIVE supervisor holds the queue lock.
+ * A stale `started` audit row must not shrink throughput for a supervisor that
+ * isn't running that config, so with no live holder the clamp is skipped and
+ * the unclamped base is used.
+ */
+
+import { describe, test, expect, beforeAll, afterAll, beforeEach, afterEach } from 'bun:test';
+import { mkdtempSync, writeFileSync, rmSync } from 'fs';
+import { join } from 'path';
+import { tmpdir } from 'os';
+import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import { tryAcquireDbLock } from '../src/core/db-lock.ts';
+import { supervisorLockId, SUPERVISOR_LOCK_TTL_MIN } from '../src/core/minions/supervisor.ts';
+import { computeSupervisorAuditFilename } from '../src/core/minions/handlers/supervisor-audit.ts';
+import { resolveEffectiveFanoutMax } from '../src/commands/autopilot-fanout.ts';
+
+let engine: PGLiteEngine;
+let auditDir: string;
+const prevAuditDir = process.env.GBRAIN_AUDIT_DIR;
+
+beforeAll(async () => {
+  engine = new PGLiteEngine();
+  await engine.connect({});
+  await engine.initSchema();
+}, 30000);
+
+afterAll(async () => {
+  await engine.disconnect();
+});
+
+beforeEach(async () => {
+  auditDir = mkdtempSync(join(tmpdir(), 'gbrain-clamp-'));
+  process.env.GBRAIN_AUDIT_DIR = auditDir;
+  await engine.executeRaw(`DELETE FROM gbrain_cycle_locks WHERE id LIKE 'gbrain-supervisor:%'`);
+  // base fan-out: override to 8 so the clamp's effect is visible on PGLite
+  // (whose natural default is 1). The clamp logic is engine-agnostic.
+  await engine.setConfig('autopilot.fanout_max_per_tick', '8');
+  await engine.setConfig('autopilot.fanout_clamp_to_concurrency', 'true');
+});
+
+afterEach(() => {
+  if (prevAuditDir === undefined) delete process.env.GBRAIN_AUDIT_DIR;
+  else process.env.GBRAIN_AUDIT_DIR = prevAuditDir;
+  try { rmSync(auditDir, { recursive: true, force: true }); } catch { /* noop */ }
+});
+
+function writeStarted(concurrency: number): void {
+  const file = join(auditDir, computeSupervisorAuditFilename());
+  writeFileSync(file, JSON.stringify({
+    event: 'started', ts: new Date().toISOString(), supervisor_pid: 4242,
+    queue: 'default', concurrency,
+  }) + '\n', 'utf8');
+}
+
+describe('resolveEffectiveFanoutMax — clamp gated on live supervisor (#2194/codex #9)', () => {
+  test('NO live holder → no clamp (stale audit row cannot shrink throughput)', async () => {
+    writeStarted(3); // audit says concurrency 3, but no live lock holder
+    const n = await resolveEffectiveFanoutMax(engine, 'default');
+    expect(n).toBe(8); // unclamped base
+  });
+
+  test('live holder + concurrency 3 → clamp to max(1, 3-1) = 2', async () => {
+    writeStarted(3);
+    const holder = await tryAcquireDbLock(engine, supervisorLockId('default'), SUPERVISOR_LOCK_TTL_MIN);
+    expect(holder).not.toBeNull();
+    try {
+      const n = await resolveEffectiveFanoutMax(engine, 'default');
+      expect(n).toBe(2);
+    } finally {
+      await holder!.release();
+    }
+  });
+
+  test('live holder but clamp disabled → unclamped base', async () => {
+    await engine.setConfig('autopilot.fanout_clamp_to_concurrency', 'false');
+    writeStarted(3);
+    const holder = await tryAcquireDbLock(engine, supervisorLockId('default'), SUPERVISOR_LOCK_TTL_MIN);
+    try {
+      const n = await resolveEffectiveFanoutMax(engine, 'default');
+      expect(n).toBe(8);
+    } finally {
+      await holder!.release();
+    }
+  });
+
+  test('live holder + concurrency 1 → floor at 1 (never below 1)', async () => {
+    writeStarted(1);
+    const holder = await tryAcquireDbLock(engine, supervisorLockId('default'), SUPERVISOR_LOCK_TTL_MIN);
+    try {
+      const n = await resolveEffectiveFanoutMax(engine, 'default');
+      expect(n).toBe(1);
+    } finally {
+      await holder!.release();
+    }
+  });
+
+  test('live holder but no started event (concurrency unknown) → no clamp', async () => {
+    // lock row exists but audit has no concurrency → fall back to base.
+    const holder = await tryAcquireDbLock(engine, supervisorLockId('default'), SUPERVISOR_LOCK_TTL_MIN);
+    try {
+      const n = await resolveEffectiveFanoutMax(engine, 'default');
+      expect(n).toBe(8);
+    } finally {
+      await holder!.release();
+    }
+  });
+});

test/autopilot-fanout-wiring.test.ts

@@ -28,8 +28,11 @@ describe('autopilot.ts ↔ dispatchPerSource wiring', () => {
     );
   });
 
-  test('imports resolveFanoutMax (so PGLite gets fanoutMax=1 per codex P1-3)', () => {
-    expect(AUTOPILOT_SRC).toMatch(/resolveFanoutMax/);
+  test('imports resolveEffectiveFanoutMax (clamps to worker concurrency; PGLite base still 1)', () => {
+    // #2194 fix #1: autopilot now resolves the CLAMPED fan-out (gated on a live
+    // supervisor) instead of the raw resolveFanoutMax. The clamp wraps
+    // resolveFanoutMax, so PGLite's base-1 still holds (codex P1-3).
+    expect(AUTOPILOT_SRC).toMatch(/resolveEffectiveFanoutMax/);
   });
 
   test('calls dispatchPerSource within the shouldFullCycle branch', () => {

test/doctor-autopilot-fanout-concurrency.test.ts

@@ -0,0 +1,84 @@
+/**
+ * #2194 fix #5: `gbrain doctor` warns when autopilot's per-tick fan-out exceeds
+ * the worker's effective concurrency. Fanning out more cycles than there are
+ * worker slots guarantees waiters that race the stalled-sweeper — a silent
+ * misconfig the operator never saw before this check.
+ *
+ * Drives computeAutopilotFanoutConcurrencyCheck directly with a fake engine so
+ * the fan-out (config) and concurrency (audit) inputs are controllable without
+ * spawning a supervisor. The audit read is stubbed via GBRAIN_AUDIT_DIR + a
+ * hand-written started event.
+ */
+
+import { describe, test, expect, beforeEach, afterEach } from 'bun:test';
+import { mkdtempSync, writeFileSync, rmSync } from 'fs';
+import { join } from 'path';
+import { tmpdir } from 'os';
+import { computeAutopilotFanoutConcurrencyCheck } from '../src/commands/doctor.ts';
+import { computeSupervisorAuditFilename } from '../src/core/minions/handlers/supervisor-audit.ts';
+
+// Minimal fake engine: postgres-kind + a config map for fanout override.
+function fakeEngine(config: Record<string, string> = {}) {
+  return {
+    kind: 'postgres' as const,
+    getConfig: async (k: string) => config[k] ?? null,
+  } as any;
+}
+
+let auditDir: string;
+const prevAuditDir = process.env.GBRAIN_AUDIT_DIR;
+
+beforeEach(() => {
+  auditDir = mkdtempSync(join(tmpdir(), 'gbrain-fanout-doctor-'));
+  process.env.GBRAIN_AUDIT_DIR = auditDir;
+});
+
+afterEach(() => {
+  if (prevAuditDir === undefined) delete process.env.GBRAIN_AUDIT_DIR;
+  else process.env.GBRAIN_AUDIT_DIR = prevAuditDir;
+  try { rmSync(auditDir, { recursive: true, force: true }); } catch { /* noop */ }
+});
+
+/** Write a `started` audit event with the given concurrency for queue 'default'. */
+function writeStarted(concurrency: number): void {
+  const file = join(auditDir, computeSupervisorAuditFilename());
+  const line = JSON.stringify({
+    event: 'started',
+    ts: new Date().toISOString(),
+    supervisor_pid: 4242,
+    queue: 'default',
+    concurrency,
+  });
+  writeFileSync(file, line + '\n', 'utf8');
+}
+
+describe('computeAutopilotFanoutConcurrencyCheck (#2194 fix #5)', () => {
+  test('warns when fan-out (4) exceeds effective slots (concurrency 2 → 1)', async () => {
+    writeStarted(2);
+    const check = await computeAutopilotFanoutConcurrencyCheck(fakeEngine());
+    expect(check.status).toBe('warn');
+    expect(check.message).toContain('exceeds worker concurrency');
+    expect(check.details).toMatchObject({ fanout_max: 4, concurrency: 2, effective_slots: 1 });
+  });
+
+  test('ok when fan-out fits (override 1, concurrency 4)', async () => {
+    writeStarted(4);
+    const check = await computeAutopilotFanoutConcurrencyCheck(
+      fakeEngine({ 'autopilot.fanout_max_per_tick': '1' }),
+    );
+    expect(check.status).toBe('ok');
+  });
+
+  test('ok/skip when no supervisor has ever started (no noise on unsupervised brains)', async () => {
+    // No started event written.
+    const check = await computeAutopilotFanoutConcurrencyCheck(fakeEngine());
+    expect(check.status).toBe('ok');
+    expect(check.message).toContain('No supervisor observed');
+  });
+
+  test('PGLite short-circuits (single-writer, fan-out is 1)', async () => {
+    const check = await computeAutopilotFanoutConcurrencyCheck({ kind: 'pglite', getConfig: async () => null } as any);
+    expect(check.status).toBe('ok');
+    expect(check.message).toContain('PGLite');
+  });
+});