From 180a776f1356dc32e31f17f183ff54df2ff646b4 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Tue, 16 Jun 2026 22:47:00 -0700 Subject: [PATCH] feat(autopilot): clamp fan-out to worker concurrency + doctor warning (#2194) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- src/commands/autopilot-fanout.ts | 56 +++++++++ src/commands/autopilot.ts | 8 +- src/commands/doctor.ts | 49 ++++++++ test/autopilot-fanout-clamp.test.ts | 110 ++++++++++++++++++ test/autopilot-fanout-wiring.test.ts | 7 +- ...octor-autopilot-fanout-concurrency.test.ts | 84 +++++++++++++ 6 files changed, 310 insertions(+), 4 deletions(-) create mode 100644 test/autopilot-fanout-clamp.test.ts create mode 100644 test/doctor-autopilot-fanout-concurrency.test.ts diff --git a/src/commands/autopilot-fanout.ts b/src/commands/autopilot-fanout.ts index 364ac7d61..5ba58523a 100644 --- a/src/commands/autopilot-fanout.ts +++ b/src/commands/autopilot-fanout.ts @@ -83,6 +83,62 @@ export async function resolveFanoutMax(engine: BrainEngine): Promise { 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 { + 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 { + 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 diff --git a/src/commands/autopilot.ts b/src/commands/autopilot.ts index fd232eb6f..09d95557c 100644 --- a/src/commands/autopilot.ts +++ b/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, diff --git a/src/commands/doctor.ts b/src/commands/doctor.ts index aa85b7bac..fe98bc920 100644 --- a/src/commands/doctor.ts +++ b/src/commands/doctor.ts @@ -695,6 +695,9 @@ export async function doctorReportRemote(engine: BrainEngine): Promise { + 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 { 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'); diff --git a/test/autopilot-fanout-clamp.test.ts b/test/autopilot-fanout-clamp.test.ts new file mode 100644 index 000000000..f4b1df3e5 --- /dev/null +++ b/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(); + } + }); +}); diff --git a/test/autopilot-fanout-wiring.test.ts b/test/autopilot-fanout-wiring.test.ts index cb783aefe..addf8dbe3 100644 --- a/test/autopilot-fanout-wiring.test.ts +++ b/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', () => { diff --git a/test/doctor-autopilot-fanout-concurrency.test.ts b/test/doctor-autopilot-fanout-concurrency.test.ts new file mode 100644 index 000000000..a52571a9c --- /dev/null +++ b/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 = {}) { + 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'); + }); +});