Compare commits

...
Author SHA1 Message Date
Garry TanandClaude Opus 5 119f30acff fix(autopilot): stop the drain-worker self-deadlocking at concurrency=1 (#2050)
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 <noreply@anthropic.com>
2026-07-28 16:49:01 -07:00
5 changed files with 221 additions and 48 deletions
+18 -21
View File
@@ -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 {
+49 -23
View File
@@ -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<void>,
handler: MinionHandler = makeSubagentHandler({ engine }),
lockMs: number = INLINE_LOCK_MS,
): Promise<void> {
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<number, { idx: number; hash6: string }>();
@@ -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,
};
+150
View File
@@ -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<void> {
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<string, unknown>).runSubagentsInline
?? (__testing as Record<string, unknown>).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<void>, h?: unknown, lockMs?: number,
) => Promise<void>)(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);
});
+1 -1
View File
@@ -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;
+3 -3
View File
@@ -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,