From 119f30acff8ca97bdb8c626b5ac3696d4608103c Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Tue, 28 Jul 2026 16:49:01 -0700 Subject: [PATCH] fix(autopilot): stop the drain-worker self-deadlocking at concurrency=1 (#2050) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit autopilot spawns its drain worker as `jobs work --max-rss N` with no --concurrency, which resolveWorkerConcurrency resolves to 1. A patterns/synthesize parent job then submits its subagent child to the 'default' queue and blocks in waitForCompletion — but the parent holds the worker's only slot, so the child is never claimed: structural self-deadlock that burns the whole subagent_wait_timeout window (35 min default), cancels the child, and repeats every cycle. The PR #2128 mitigation was PGLite-only (runPgliteSubagentsInline early-returned unless engine.kind === 'pglite'). Fix — the child must not need a worker slot at all: - runPgliteSubagentsInline -> runSubagentsInline: the inline claim -> run -> complete/fail drain now runs on BOTH engines, from the slot the parent already holds. - patterns/synthesize children go to a private per-run queue on both engines (previously 'default' on Postgres), so a 'default'-queue worker can never race the parent for its own child. - New: the drain heartbeats the child's claim lock at lockMs/3 (worker cadence parity). Required on Postgres because a concurrent worker sweeps handleStalled() across ALL queues; without renewal any child running longer than 30s would be requeued mid-run and stall-churned to dead. A false renew (row cancelled/reclaimed) aborts the handler; errors are swallowed (best-effort, never an unhandledRejection). Why not raise the spawn concurrency: N slots deadlock identically once N blocking parents are claimed (autopilot fans out up to 4 cycle jobs per tick, all submitted before any child exists), so no fixed number is structurally safe. Inline draining is: it adds zero worker slots and zero DB-pool concurrency — the child's work replaces the parent's idle 5s waitForCompletion polling in the same slot — so it cannot add pooler pressure or starve supervisor/job lock renewals. The spawn args stay untouched, which also preserves GBRAIN_WORKER_CONCURRENCY as an operator escape hatch (an explicit --concurrency flag would out-rank the env). Test: test/autopilot-drain-deadlock-2050.test.ts runs the real patterns phase and the real drain against a PGLite engine masked as kind='postgres'. On unmodified master both tests fail behaviorally (child_outcome 'timeout', child stuck 'waiting' on 'default'); with the fix the child is drained inline on a dream-inline-* queue and the stalled-sweep never touches the heartbeated lock. Co-Authored-By: Claude Opus 5 --- src/core/cycle/patterns.ts | 39 +++--- src/core/cycle/synthesize.ts | 72 ++++++---- test/autopilot-drain-deadlock-2050.test.ts | 150 +++++++++++++++++++++ test/cycle-patterns-child-outcome.test.ts | 2 +- test/e2e/dream-synthesize-pglite.test.ts | 6 +- 5 files changed, 221 insertions(+), 48 deletions(-) create mode 100644 test/autopilot-drain-deadlock-2050.test.ts diff --git a/src/core/cycle/patterns.ts b/src/core/cycle/patterns.ts index 0673f718e..f9b3b7134 100644 --- a/src/core/cycle/patterns.ts +++ b/src/core/cycle/patterns.ts @@ -30,14 +30,13 @@ import { serializeMarkdown } from '../markdown.ts'; import type { Page, PageType } from '../types.ts'; // #2415: allow-list + output-root resolution shared with the synthesize // phase — both phases must agree on the configured namespace. -// runPgliteSubagentsInline is shared too: PGLite has no separate Minions -// worker process (the embedded data-dir holds an exclusive file lock), so a -// job submitted via queue.add() sits in 'waiting' forever unless something -// drives the claim -> run -> complete loop inline. synthesize.ts already -// does this for its own children; patterns.ts previously submitted and -// waited without ever draining, so every real (non-dry-run) invocation on a -// PGLite brain hung until subagentWaitTimeoutMs (default 35 min). -import { loadAllowedSlugPrefixes, loadOutputRoot, runPgliteSubagentsInline } from './synthesize.ts'; +// runSubagentsInline is shared too: a job submitted via queue.add() sits in +// 'waiting' forever unless something drives the claim -> run -> complete +// loop — on PGLite because no separate worker can open the embedded +// data-dir, on Postgres because the parent phase itself occupies a worker +// slot and can deadlock a fully-occupied worker (#2050). synthesize.ts +// drains its own children the same way. +import { loadAllowedSlugPrefixes, loadOutputRoot, runSubagentsInline } from './synthesize.ts'; import { probeChatModel } from '../ai/gateway.ts'; import { normalizeModelId } from '../model-id.ts'; @@ -185,13 +184,12 @@ export async function runPhasePatterns( } const queue = new MinionQueue(engine); - // PGLite children drain inline (no separate worker can open the embedded - // data-dir), so give this job a private per-run queue: the inline drain - // must never claim unrelated 'default'-queue jobs a Postgres worker owns. - // Mirrors synthesize.ts's childQueueName derivation exactly. - const childQueueName = engine.kind === 'pglite' - ? `dream-inline-${Date.now()}-${randomUUID().slice(0, 8)}` - : 'default'; + // #2050: children drain inline on BOTH engines (see runSubagentsInline), + // so give this job a private per-run queue: the inline drain must never + // claim unrelated 'default'-queue jobs, and a 'default'-queue worker must + // never claim a child this parent is about to run itself. Mirrors + // synthesize.ts's childQueueName derivation exactly. + const childQueueName = `dream-inline-${Date.now()}-${randomUUID().slice(0, 8)}`; const data: SubagentHandlerData = { prompt: buildPatternsPrompt(reflections, config.minEvidence, config.sourceSlugPrefix, config.outputSlugPrefix), model: config.model, @@ -207,12 +205,11 @@ export async function runPhasePatterns( allowProtectedSubmit: true, }); - // PGLite cannot run a separate Minions worker because the embedded DB - // holds an exclusive file lock. Drain this phase's private child queue - // inline so the parent observes the terminal state instead of polling - // waitForCompletion until subagentWaitTimeoutMs expires. No-op on - // Postgres (a real worker process claims the job there). - await runPgliteSubagentsInline(engine, queue, childQueueName, opts.yieldDuringPhase); + // Drain this phase's private child queue inline so the parent observes + // the terminal state instead of polling waitForCompletion until + // subagentWaitTimeoutMs expires. Runs on BOTH engines — on Postgres the + // parent job otherwise deadlocks a fully-occupied worker (#2050). + await runSubagentsInline(engine, queue, childQueueName, opts.yieldDuringPhase); let outcome: string; try { diff --git a/src/core/cycle/synthesize.ts b/src/core/cycle/synthesize.ts index f786294a3..ab43524f1 100644 --- a/src/core/cycle/synthesize.ts +++ b/src/core/cycle/synthesize.ts @@ -265,27 +265,41 @@ export interface SynthesizePhaseOpts { once?: boolean; } -const INLINE_PGLITE_LOCK_MS = 30_000; +const INLINE_LOCK_MS = 30_000; /** - * PGLite cannot be served by a separate Minions worker process: the embedded - * data-dir holds an exclusive file lock, so subagent children enqueued by the - * synth parent would sit in 'waiting' until waitForCompletion times out. - * Drive the same claim → run → complete/fail loop a worker would perform, - * inline, against this phase's private child queue. + * Drain this phase's private child queue inline: drive the same claim → run → + * complete/fail loop a worker would perform, from the parent's own slot. + * + * Why inline on BOTH engines: + * - PGLite: no separate Minions worker can run at all (the embedded + * data-dir holds an exclusive file lock), so children would sit in + * 'waiting' until waitForCompletion times out. + * - Postgres (#2050): the parent phase itself runs as a job inside a + * `jobs work` process. A worker whose slots are all occupied by such + * parents (autopilot spawns its drain worker at the default + * concurrency=1) can never claim the child the parent is blocking on — + * a structural self-deadlock. Running children inline means a child + * never needs a worker slot, so the deadlock is impossible at ANY + * concurrency, and no extra DB-pool pressure is added: the child's work + * replaces the parent's idle waitForCompletion polling in the slot the + * parent already holds. * * `yieldDuringPhase` is ticked on a 60s interval while a child runs so the * 5-min cycle lock TTL keeps refreshing during long (up to 30-min) children. + * The child's own claim lock is heartbeated at lockMs/3 (worker cadence + * parity) — on Postgres a concurrent worker sweeps handleStalled() across + * ALL queues, so without renewal any child running longer than lockMs would + * be requeued mid-run and stall-churned to dead. */ -export async function runPgliteSubagentsInline( +export async function runSubagentsInline( engine: BrainEngine, queue: MinionQueue, queueName: string, yieldDuringPhase?: () => Promise, handler: MinionHandler = makeSubagentHandler({ engine }), + lockMs: number = INLINE_LOCK_MS, ): Promise { - if (engine.kind !== 'pglite') return; - while (true) { // Housekeeping a worker would normally perform, so child rows can reach // terminal states (delayed retries promoted, timeouts dead-lettered) @@ -293,10 +307,10 @@ export async function runPgliteSubagentsInline( await queue.promoteDelayed(); await queue.handleStalled(); await queue.handleTimeouts(); - await queue.handleWallClockTimeouts(INLINE_PGLITE_LOCK_MS); + await queue.handleWallClockTimeouts(lockMs); const lockToken = randomUUID(); - const job = await queue.claim(lockToken, INLINE_PGLITE_LOCK_MS, queueName, ['subagent']); + const job = await queue.claim(lockToken, lockMs, queueName, ['subagent']); if (!job) return; const abort = new AbortController(); @@ -353,6 +367,18 @@ export async function runPgliteSubagentsInline( const keepalive = yieldDuringPhase ? setInterval(() => { yieldDuringPhase().catch(() => { /* best-effort */ }); }, 60_000) : null; + // #2050: heartbeat the child's claim lock while the handler runs so a + // concurrent Postgres worker's handleStalled() sweep (all queues, not + // just its own) can't requeue a live child. A false return means the row + // was cancelled or reclaimed — abort the handler. Errors are swallowed + // (best-effort; the next tick retries), never an unhandledRejection. + const renewTimer = setInterval(() => { + queue.renewLock(job.id, lockToken, lockMs) + .then((ok) => { + if (!ok && !abort.signal.aborted) abort.abort(new Error('lock-renewal-failed')); + }) + .catch(() => { /* best-effort; next tick retries */ }); + }, Math.max(50, Math.floor(lockMs / 3))); try { const result = await handler(context); await queue.completeJob( @@ -376,6 +402,7 @@ export async function runPgliteSubagentsInline( } finally { if (timeoutTimer) clearTimeout(timeoutTimer); if (keepalive) clearInterval(keepalive); + clearInterval(renewTimer); } } } @@ -546,12 +573,11 @@ export async function runPhaseSynthesize( } const queue = new MinionQueue(engine); - // PGLite children drain inline (no separate worker can open the embedded - // data-dir), so give them a private per-run queue: the inline drain must - // never claim unrelated 'default'-queue jobs a Postgres worker owns. - const childQueueName = engine.kind === 'pglite' - ? `dream-inline-${Date.now()}-${randomUUID().slice(0, 8)}` - : 'default'; + // #2050: children drain inline on BOTH engines (see runSubagentsInline), + // so give them a private per-run queue: the inline drain must never claim + // unrelated 'default'-queue jobs, and a 'default'-queue worker must never + // claim a child this parent is about to run itself. + const childQueueName = `dream-inline-${Date.now()}-${randomUUID().slice(0, 8)}`; const childIds: number[] = []; /** Map child job_id → chunk metadata for D6 orchestrator-side slug rewrite. */ const chunkInfo = new Map(); @@ -659,11 +685,11 @@ export async function runPhaseSynthesize( } } - // PGLite cannot run a separate Minions worker because the embedded DB - // holds an exclusive file lock. Drain this phase's private child queue - // inline so the parent observes terminal child states instead of polling - // waiters until subagentWaitTimeoutMs expires. No-op on Postgres. - await runPgliteSubagentsInline(engine, queue, childQueueName, opts.yieldDuringPhase); + // Drain this phase's private child queue inline so the parent observes + // terminal child states instead of polling waiters until + // subagentWaitTimeoutMs expires. Runs on BOTH engines — on Postgres the + // parent job otherwise deadlocks a fully-occupied worker (#2050). + await runSubagentsInline(engine, queue, childQueueName, opts.yieldDuringPhase); // Wait for every child to reach a terminal state. Tick yieldDuringPhase // every 5 min so the cycle lock TTL refreshes. @@ -1598,5 +1624,5 @@ export const __testing = { buildSynthesisPrompt, stampDreamProvenance, reverseWriteRefs, - runPgliteSubagentsInline, + runSubagentsInline, }; diff --git a/test/autopilot-drain-deadlock-2050.test.ts b/test/autopilot-drain-deadlock-2050.test.ts new file mode 100644 index 000000000..8ac6c3d69 --- /dev/null +++ b/test/autopilot-drain-deadlock-2050.test.ts @@ -0,0 +1,150 @@ +/** + * #2050 — a patterns/synthesize parent job that blocks on its own subagent + * child must not deadlock a fully-occupied Postgres drain worker. + * + * Pre-fix, the inline child drain was PGLite-only (runPgliteSubagentsInline + * returned early unless engine.kind === 'pglite') and Postgres children were + * submitted to the 'default' queue. Autopilot spawns its drain worker with no + * --concurrency (resolves to 1), so the parent occupied the only slot, the + * child was never claimed, and the phase burned the whole + * subagent_wait_timeout_ms window before cancelling its own child — every + * cycle, forever. + * + * These tests run the REAL phase/drain code against a PGLite engine masked as + * kind='postgres' (only the `kind` property is intercepted; every method call + * hits the real engine), which is exactly the branch the production Postgres + * environment takes. All assertions are behavioral: on unmodified master they + * fail with outcome 'timeout' / a never-claimed 'waiting' child on the + * 'default' queue. + */ + +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import type { BrainEngine } from '../src/core/engine.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; +import { runPhasePatterns } from '../src/core/cycle/patterns.ts'; +import { MinionQueue } from '../src/core/minions/queue.ts'; +import { withEnv } from './helpers/with-env.ts'; + +let engine: PGLiteEngine; +let schemaVersion: string; + +/** Mask a PGLite engine as Postgres: only `kind` is intercepted, every + * method executes on the real engine (bound to the target so internal + * state is untouched). */ +function maskAsPostgres(target: PGLiteEngine): BrainEngine { + return new Proxy(target, { + get(t, prop) { + if (prop === 'kind') return 'postgres'; + const v = Reflect.get(t, prop); + return typeof v === 'function' ? v.bind(t) : v; + }, + }) as unknown as BrainEngine; +} + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({ database_url: '' }); + await engine.initSchema(); + schemaVersion = (await engine.getConfig('version')) ?? '7'; +}, 60_000); + +afterAll(async () => { + await engine.disconnect(); +}); + +beforeEach(async () => { + await resetPgliteState(engine); + await engine.setConfig('version', schemaVersion); +}); + +async function seedReflections(): Promise { + for (let i = 0; i < 3; i++) { + await engine.executeRaw( + `INSERT INTO pages (slug, type, title, compiled_truth) + VALUES ($1, 'note', $2, $3)`, + [ + `wiki/personal/reflections/2026-07-0${i + 1}-reflection`, + `Reflection ${i + 1}`, + `Recurring theme fixture number ${i + 1}.`, + ], + ); + } +} + +describe('#2050 — patterns parent must not deadlock on its own child (postgres path)', () => { + test('child is drained inline on a private queue instead of waiting out the parent', async () => { + const brainDir = mkdtempSync(join(tmpdir(), 'gbrain-2050-patterns-')); + try { + await seedReflections(); + // Small wait window: on master the child is never claimed (no worker in + // this process, parent would hold the only slot in production), so the + // phase burns this whole window and reports outcome 'timeout'. + await engine.setConfig('dream.patterns.subagent_wait_timeout_ms', '2000'); + + const pgAlike = maskAsPostgres(engine); + const result = await withEnv({ ANTHROPIC_API_KEY: 'sk-ant-test' }, () => + runPhasePatterns(pgAlike, { brainDir, dryRun: false }), + ); + + // Master: child_outcome 'timeout' + PATTERNS_CHILD_TIMEOUT (self-deadlock, + // then the phase cancels its own child). Fixed: the inline drain actually + // ran the child (the fake API key fails fast → 'dead'), no deadlock. + expect(result.details.child_outcome).toBe('dead'); + expect(result.error?.code).toBe('PATTERNS_CHILD_DEAD'); + + const jobs = await engine.executeRaw<{ queue: string; status: string }>( + `SELECT queue, status FROM minion_jobs WHERE name = 'subagent' ORDER BY id DESC LIMIT 1`, + ); + expect(jobs).toHaveLength(1); + // Master: queue 'default' (claimable by — and deadlocked behind — the + // same worker running the parent), status 'cancelled'. + expect(jobs[0].queue).toStartWith('dream-inline-'); + expect(jobs[0].status).toBe('dead'); + } finally { + rmSync(brainDir, { recursive: true, force: true }); + } + }, 60_000); + + test('inline drain runs children on a non-pglite engine and heartbeats their lock', async () => { + const { __testing } = await import('../src/core/cycle/synthesize.ts'); + const drain = (__testing as Record).runSubagentsInline + ?? (__testing as Record).runPgliteSubagentsInline; + + const pgAlike = maskAsPostgres(engine); + const queue = new MinionQueue(pgAlike); + const job = await queue.add( + 'subagent', + { prompt: 'noop', model: 'anthropic:claude-sonnet-4-5', max_turns: 1 }, + { queue: 'inline-2050-test', max_stalled: 3 }, + { allowProtectedSubmit: true }, + ); + + // Stub handler outliving the (shortened) claim lock: sleeps past lockMs, + // then runs the same handleStalled() sweep a concurrent Postgres worker + // fires on a timer. With the heartbeat the lock is fresh and the sweep + // must NOT touch the running child; without it the child would be + // requeued mid-run (stall churn → dead after max_stalled). + let sweptWhileRunning = 0; + const stub = async () => { + await new Promise((r) => setTimeout(r, 1400)); + const { requeued, dead } = await queue.handleStalled(); + sweptWhileRunning = [...requeued, ...dead].filter((j) => j.id === job.id).length; + return { ok: true }; + }; + + // Master behavioral failure: runPgliteSubagentsInline no-ops on a + // kind='postgres' engine, the stub never runs, the job stays 'waiting'. + await (drain as ( + e: BrainEngine, q: MinionQueue, name: string, + y?: () => Promise, h?: unknown, lockMs?: number, + ) => Promise)(pgAlike, queue, 'inline-2050-test', undefined, stub, 1000); + + const after = await queue.getJob(job.id); + expect(after?.status).toBe('completed'); + expect(sweptWhileRunning).toBe(0); + }, 30_000); +}); diff --git a/test/cycle-patterns-child-outcome.test.ts b/test/cycle-patterns-child-outcome.test.ts index 60c94f9f5..c976c1666 100644 --- a/test/cycle-patterns-child-outcome.test.ts +++ b/test/cycle-patterns-child-outcome.test.ts @@ -5,7 +5,7 @@ * zero pattern pages written (e.g. when no subagent-capable worker slot was * free for the whole wait window) — a silent no-op for days. * - * A later fix added runPgliteSubagentsInline to this phase (patterns.ts + * A later fix added the shared inline drain (now runSubagentsInline) to this phase (patterns.ts * previously submitted a job and waited without anything ever claiming it on * PGLite — synthesize.ts already had this inline drain, patterns.ts didn't). * So a fake ANTHROPIC_API_KEY here now gets claimed and actually attempted; diff --git a/test/e2e/dream-synthesize-pglite.test.ts b/test/e2e/dream-synthesize-pglite.test.ts index a859f5af2..b60cbe48a 100644 --- a/test/e2e/dream-synthesize-pglite.test.ts +++ b/test/e2e/dream-synthesize-pglite.test.ts @@ -532,7 +532,7 @@ describe('E2E synthesize — PGLite inline subagent drain (takeover of #2699)', ); let ticks = 0; - await synthTesting.runPgliteSubagentsInline( + await synthTesting.runSubagentsInline( rig.engine, queue, queueName, @@ -573,7 +573,7 @@ describe('E2E synthesize — PGLite inline subagent drain (takeover of #2699)', { allowProtectedSubmit: true }, ); - await synthTesting.runPgliteSubagentsInline( + await synthTesting.runSubagentsInline( rig.engine, queue, queueName, @@ -607,7 +607,7 @@ describe('E2E synthesize — PGLite inline subagent drain (takeover of #2699)', // Handler only ends when ctx.signal fires — like the real subagent // handler mid-LLM-call. Without the inline timeout timer this awaits // forever and the drain (and the whole cycle) wedges. - await synthTesting.runPgliteSubagentsInline( + await synthTesting.runSubagentsInline( rig.engine, queue, queueName,