v0.42.16.0 feat(doctor): brain health as a solved problem — cause-ranked doctor + OOM-loop line + auto-drain + pool-reap (#1685) (#1802)

* feat(minions): pool-recovery audit + reconnect reason-threading + shared drain helper (#1685 GAP B, 5A)

- pool-recovery-audit.ts: reap_detected (CONNECTION_ENDED) vs reconnect_other; recovered/failed split
- postgres-engine reconnect(ctx?) classifies the triggering error so only true pooler reaps are tagged (CODEX #8)
- retry.ts reconnect callback widened to thread the error; retry-matcher isConnectionEndedError
- runExtractAtomsDrainForSource shared helper (cycleLockIdFor + withRefreshingLock) — one drain path (5A)
- supervisor-audit readRecentSupervisorEvents (current+prev ISO week, CODEX #7)
- extract-atoms-drain PROTECTED; autopilot.auto_drain.* config keys

* feat(doctor): worker_oom_loop + pool_reap_health checks + cause-ranked top_issues (#1685 GAP A/B/C)

- computeWorkerOomLoopCheck: unions supervisor rss_watchdog + minion_jobs watchdog-abort (CODEX #5), cap fallback to resolveDefaultMaxRssMb (CODEX #6)
- computePoolReapHealthCheck: reaps-not-recovering fail, thrash warn
- doctor-cause-rank rankIssues: tier ordering + grounded downstream_of (CODEX #9) + drift guard (4A)
- supervisor causeStr + queue_health cross-reference worker_oom_loop (DRY 1C)
- register both checks in doctor-categories ops

* feat(autopilot): per-source extract_atoms auto-drain + handler + dream --drain refactor (#1685 GAP D)

- autopilot per-source gate: enabled + !packDeclares + backlog>threshold + daily cap; time-sloted idempotency key (CODEX #2)
- extract-atoms-drain Minion handler (thin wrapper, LockUnavailableError -> deferred)
- dream --drain routes through the shared helper (5A)

* chore: bump version and changelog (v0.42.12.0)

#1685 brain-health-as-solved-problem: cause-ranked doctor, worker_oom_loop
line, per-source auto-drain, pool-reap health. Layers on #1678/#1735.

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

* docs(todos): file #1685 GAP E + remote-path follow-ups (v0.42.12.0)

* fix(#1685): pre-landing review — multi-source auto-drain, honest pool-reap signal, lock-renewal reap labeling

- autopilot: drop maxWaiting (coalesces by name+queue not source → only one source drained + cap over-count); pre-check idempotency key so only genuinely-new sources submit+count
- pool_reap_health: fail on reconnect FAILURES (the real signal), not reaps>0&&failures>0 (false causality when a recovered reap + unrelated failure co-occur)
- lock-renewal-tick threads its triggering error to reconnect() so a CONNECTION_ENDED pooler reap is labeled reap_detected not reconnect_other (pool_reap_health now fires for the #1678 incident path)

* chore: re-version v0.42.12.0 → v0.42.16.0 (#1685)

Slot collision avoidance per queue.

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

* docs: restore slim CLAUDE.md + move #1685 entries to KEY_FILES.md (fix check:doc-history)

The master merge wrongly kept the pre-restructure 577KB CLAUDE.md; the
check:doc-history guard caps it at 60KB. Take master's slim CLAUDE.md and
record the #1685 files (doctor-cause-rank, pool-recovery-audit, worker_oom_loop
+ pool_reap_health checks, auto-drain, 5A helper) as current-state prose in
docs/architecture/KEY_FILES.md (no release markers). llms regenerated.

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

---------

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

test/audit/pool-recovery-audit.test.ts

@@ -0,0 +1,86 @@
+// #1685 GAP B — pool-recovery audit JSONL primitive.
+//
+// Hermetic: GBRAIN_AUDIT_DIR override via withEnv; fresh tempdir per test.
+
+import { describe, expect, test, beforeEach, afterEach } from 'bun:test';
+import * as fs from 'fs';
+import * as os from 'os';
+import * as path from 'path';
+import { withEnv } from '../helpers/with-env.ts';
+import {
+  logPoolRecovery,
+  readRecentPoolRecoveries,
+  _poolRecoveryAuditFeatureName,
+} from '../../src/core/audit/pool-recovery-audit.ts';
+
+let tmpDir: string;
+beforeEach(() => {
+  tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'pool-recovery-audit-'));
+});
+afterEach(() => {
+  try { fs.rmSync(tmpDir, { recursive: true, force: true }); } catch { /* best-effort */ }
+});
+
+describe('logPoolRecovery + readRecentPoolRecoveries', () => {
+  test('round-trips a reap → recovered pair into the right counters', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      logPoolRecovery('reap_detected', Object.assign(new Error('write CONNECTION_ENDED'), { code: 'CONNECTION_ENDED' }));
+      logPoolRecovery('reconnect_succeeded');
+      const r = readRecentPoolRecoveries(1);
+      expect(r.reaps).toBe(1);
+      expect(r.recoveries).toBe(1);
+      expect(r.failures).toBe(0);
+      expect(r.others).toBe(0);
+      expect(r.events).toHaveLength(2);
+    });
+  });
+
+  test('counts a reap that failed to recover (the not-auto-recovering signal)', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      logPoolRecovery('reap_detected');
+      logPoolRecovery('reconnect_failed', new Error('EHOSTUNREACH'));
+      const r = readRecentPoolRecoveries(1);
+      expect(r.reaps).toBe(1);
+      expect(r.failures).toBe(1);
+      expect(r.recoveries).toBe(0);
+    });
+  });
+
+  test('reconnect_other is tracked separately from reaps', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      logPoolRecovery('reconnect_other', new Error('network blip'));
+      logPoolRecovery('reconnect_succeeded');
+      const r = readRecentPoolRecoveries(1);
+      expect(r.reaps).toBe(0);
+      expect(r.others).toBe(1);
+      expect(r.recoveries).toBe(1);
+    });
+  });
+
+  test('redacts connection info from the error summary', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      logPoolRecovery(
+        'reconnect_failed',
+        new Error('could not connect to postgres://user:secret@db.example.com:5432/app (192.168.1.42)'),
+      );
+      const r = readRecentPoolRecoveries(1);
+      const summary = r.events[0].error_summary ?? '';
+      expect(summary).not.toContain('secret');
+      expect(summary).not.toContain('192.168.1.42');
+      expect(summary).toContain('<REDACTED');
+    });
+  });
+
+  test('empty dir → all-zero counters, no throw', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      const r = readRecentPoolRecoveries(1);
+      expect(r.reaps).toBe(0);
+      expect(r.events).toEqual([]);
+      expect(r.most_recent_ts).toBeNull();
+    });
+  });
+
+  test('stable feature name', () => {
+    expect(_poolRecoveryAuditFeatureName()).toBe('pool-recovery');
+  });
+});

test/autopilot-auto-drain-wiring.test.ts

@@ -0,0 +1,54 @@
+/**
+ * #1685 GAP D — autopilot auto-drain wiring regression guards.
+ *
+ * The submission is inline in the autopilot tick body, so these are
+ * source-shape assertions (the proven `autopilot-*-wiring.test.ts` pattern).
+ * The load-bearing one is CODEX #2: the idempotency key MUST carry a time slot,
+ * else queue.add returns the first completed job forever and the source never
+ * drains again.
+ */
+import { describe, test, expect } from 'bun:test';
+import { readFileSync } from 'fs';
+import { join } from 'path';
+
+const SRC = readFileSync(join(import.meta.dir, '../src/commands/autopilot.ts'), 'utf8');
+
+describe('autopilot auto-drain wiring', () => {
+  test('CODEX #2: idempotency key includes a UTC-day time slot (not static)', () => {
+    expect(SRC).toContain('autopilot-extract-atoms-drain:${src.id}:${utcDay}');
+    // A static key would be the regression — guard against the bare form.
+    expect(SRC).not.toContain('`autopilot-extract-atoms-drain:${src.id}`');
+  });
+
+  test('CODEX #1: submits with allowProtectedSubmit', () => {
+    expect(SRC).toMatch(/extract-atoms-drain[\s\S]{0,800}allowProtectedSubmit: true/);
+  });
+
+  test('CODEX #3: enumerates sources and counts backlog per source', () => {
+    expect(SRC).toContain('loadAllSources(engine)');
+    expect(SRC).toContain('countExtractAtomsBacklog(engine, src.id)');
+  });
+
+  test('gates on pack NOT declaring extract_atoms (the silent-backlog condition)', () => {
+    expect(SRC).toContain("packDeclaresPhase(engine, 'extract_atoms')");
+  });
+
+  test('gates on the enabled flag and a daily spend cap (DECISION 3C)', () => {
+    expect(SRC).toContain('autopilot.auto_drain.enabled');
+    expect(SRC).toContain('autopilot.auto_drain.max_usd_per_day');
+    expect(SRC).toContain('maxJobsToday');
+  });
+
+  test('is Postgres-gated (PGLite has no worker surface)', () => {
+    expect(SRC).toMatch(/engine\.kind === 'postgres'[\s\S]{0,400}auto_drain/);
+  });
+
+  test('CODEX impl #4: no maxWaiting (it coalesces by name+queue, not source)', () => {
+    // maxWaiting would return source A's waiting job for source B's submit,
+    // never queuing B and over-counting the cap. The per-source idempotency key
+    // is the dedup; a pre-check on it avoids counting idempotency-hit re-submits.
+    const drainBlock = SRC.slice(SRC.indexOf("'extract-atoms-drain'"));
+    expect(drainBlock.slice(0, 900)).not.toContain('maxWaiting');
+    expect(SRC).toContain('WHERE idempotency_key = $1 LIMIT 1');
+  });
+});

test/connection-resilience.test.ts

@@ -310,7 +310,10 @@ describe('Eng-review D3 — executeRaw has no per-call retry wrapper', () => {
 
   it('PostgresEngine.reconnect() still exists for supervisor-driven recovery', () => {
     const src = readFileSync(resolve('src/core/postgres-engine.ts'), 'utf-8');
-    expect(src).toContain('async reconnect()');
+    // v0.42.10.0 (#1685 GAP B): reconnect() gained an optional ctx param so it
+    // can classify the triggering error for the pool-recovery audit. Match the
+    // prefix so both `reconnect()` and `reconnect(ctx?)` satisfy the contract.
+    expect(src).toContain('async reconnect(');
     expect(src).toContain('await this.disconnect()');
   });
 

test/doctor-cause-rank.test.ts

@@ -0,0 +1,87 @@
+/**
+ * #1685 GAP C — cause-ranked doctor issues. Pure unit tests.
+ *
+ * Covers: fail-before-warn + root-before-symptom ordering, evidence-gated
+ * downstream_of (NEVER from co-occurrence alone — CODEX #9), fix-hint
+ * preference, and the DECISION 4A drift guard (every cause-graph name still
+ * exists in doctor-categories).
+ */
+import { describe, it, expect } from 'bun:test';
+import {
+  rankIssues,
+  ROOT_CAUSE_CHECKS,
+  SYMPTOM_CHECKS,
+  CAUSE_GRAPH_NAMES,
+  allKnownCheckNames,
+  type RankableCheck,
+} from '../src/core/doctor-cause-rank.ts';
+
+const ok = (name: string): RankableCheck => ({ name, status: 'ok', message: 'fine' });
+const warn = (name: string, msg = 'warned'): RankableCheck => ({ name, status: 'warn', message: msg });
+const fail = (name: string, msg = 'failed'): RankableCheck => ({ name, status: 'fail', message: msg });
+
+describe('rankIssues', () => {
+  it('drops ok checks, returns [] when everything is healthy', () => {
+    expect(rankIssues([ok('connection'), ok('queue_health')])).toEqual([]);
+  });
+
+  it('orders fail before warn', () => {
+    const out = rankIssues([warn('stale_locks'), fail('schema_version')]);
+    expect(out.map((i) => i.name)).toEqual(['schema_version', 'stale_locks']);
+  });
+
+  it('orders root before symptom within the same status', () => {
+    // queue_health (symptom) + worker_oom_loop (root), both fail.
+    const out = rankIssues([fail('queue_health'), fail('worker_oom_loop')]);
+    expect(out.map((i) => i.name)).toEqual(['worker_oom_loop', 'queue_health']);
+    expect(out[0].tier).toBe('root');
+    expect(out[1].tier).toBe('symptom');
+  });
+
+  it('tags downstream_of ONLY when the named root is itself failing', () => {
+    const both = rankIssues([fail('worker_oom_loop'), fail('queue_health')]);
+    const q1 = both.find((i) => i.name === 'queue_health')!;
+    expect(q1.downstream_of).toBe('worker_oom_loop');
+  });
+
+  it('does NOT tag downstream_of when the root is absent (no co-occurrence guess)', () => {
+    const out = rankIssues([fail('queue_health')]); // worker_oom_loop not failing
+    const q = out.find((i) => i.name === 'queue_health')!;
+    expect(q.downstream_of).toBeUndefined();
+  });
+
+  it('does NOT tag downstream_of from a generic root×symptom cartesian', () => {
+    // schema_version is a root and stale_locks is a symptom, but there is NO
+    // declared causal edge between them — co-occurrence must not invent one.
+    const out = rankIssues([fail('schema_version'), fail('stale_locks')]);
+    const s = out.find((i) => i.name === 'stale_locks')!;
+    expect(s.downstream_of).toBeUndefined();
+  });
+
+  it('uses details.fix_hint when present, else the message', () => {
+    const out = rankIssues([
+      { name: 'worker_oom_loop', status: 'fail', message: 'long message', details: { fix_hint: 'raise --max-rss' } },
+      warn('orphan_ratio', 'too many orphans'),
+    ]);
+    expect(out.find((i) => i.name === 'worker_oom_loop')!.fix).toBe('raise --max-rss');
+    expect(out.find((i) => i.name === 'orphan_ratio')!.fix).toBe('too many orphans');
+  });
+
+  it('is deterministic (name tiebreak) for same status+tier', () => {
+    const out = rankIssues([warn('zeta_unknown'), warn('alpha_unknown')]);
+    expect(out.map((i) => i.name)).toEqual(['alpha_unknown', 'zeta_unknown']);
+  });
+});
+
+describe('DECISION 4A drift guard', () => {
+  it('every cause-graph name exists in doctor-categories known names', () => {
+    const known = allKnownCheckNames();
+    const missing = [...CAUSE_GRAPH_NAMES].filter((n) => !known.has(n));
+    expect(missing).toEqual([]);
+  });
+
+  it('root and symptom sets are disjoint', () => {
+    const overlap = [...ROOT_CAUSE_CHECKS].filter((n) => SYMPTOM_CHECKS.has(n));
+    expect(overlap).toEqual([]);
+  });
+});

test/doctor-pool-reap-health.test.ts

@@ -0,0 +1,77 @@
+// #1685 GAP B — pool_reap_health doctor check.
+//
+// computePoolReapHealthCheck only touches engine.kind + the pool-recovery audit
+// (filesystem), so a minimal `{ kind: 'postgres' }` stub drives it hermetically.
+
+import { describe, expect, test, beforeEach, afterEach } from 'bun:test';
+import * as fs from 'fs';
+import * as os from 'os';
+import * as path from 'path';
+import { withEnv } from './helpers/with-env.ts';
+import { logPoolRecovery } from '../src/core/audit/pool-recovery-audit.ts';
+import { computePoolReapHealthCheck } from '../src/commands/doctor.ts';
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+const pg = { kind: 'postgres' } as any;
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+const pglite = { kind: 'pglite' } as any;
+
+let tmpDir: string;
+beforeEach(() => {
+  tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'pool-reap-health-'));
+});
+afterEach(() => {
+  try { fs.rmSync(tmpDir, { recursive: true, force: true }); } catch { /* best-effort */ }
+});
+
+describe('computePoolReapHealthCheck', () => {
+  test('null on PGLite (no pool) and on null engine', async () => {
+    expect(await computePoolReapHealthCheck(pglite)).toBeNull();
+    expect(await computePoolReapHealthCheck(null)).toBeNull();
+  });
+
+  test('fail when reconnect failed (reconnect is throwing)', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      logPoolRecovery('reap_detected');
+      logPoolRecovery('reconnect_failed', new Error('EHOSTUNREACH'));
+      const c = await computePoolReapHealthCheck(pg);
+      expect(c?.status).toBe('fail');
+      expect(c?.message).toContain('reconnect is throwing');
+      expect(c?.name).toBe('pool_reap_health');
+    });
+  });
+
+  // CODEX impl review #3: the fail trigger is the reconnect FAILURES themselves
+  // (reconnect throwing is the real, actionable problem), NOT a fabricated
+  // reap→failure causal link. A reconnect_failed with zero reaps still fails.
+  test('fail on reconnect failure even with zero reaps (no false causality)', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      logPoolRecovery('reconnect_failed', new Error('password authentication failed'));
+      const c = await computePoolReapHealthCheck(pg);
+      expect(c?.status).toBe('fail');
+      expect(c?.message).toContain('0 pooler reap(s) detected');
+      expect(c?.message).not.toContain('not auto-recovering');
+    });
+  });
+
+  test('warn on pooler thrash (>=10 reaps all recovered)', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      for (let i = 0; i < 12; i++) {
+        logPoolRecovery('reap_detected');
+        logPoolRecovery('reconnect_succeeded');
+      }
+      const c = await computePoolReapHealthCheck(pg);
+      expect(c?.status).toBe('warn');
+      expect(c?.message).toContain('12×');
+    });
+  });
+
+  test('null (quiet) when a few reaps all recovered', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      logPoolRecovery('reap_detected');
+      logPoolRecovery('reconnect_succeeded');
+      const c = await computePoolReapHealthCheck(pg);
+      expect(c).toBeNull();
+    });
+  });
+});

test/doctor-worker-oom-loop.test.ts

@@ -0,0 +1,101 @@
+// #1685 GAP A — worker_oom_loop doctor check.
+//
+// Hermetic: writes synthetic supervisor audit JSONL into a GBRAIN_AUDIT_DIR
+// tempdir and drives computeWorkerOomLoopCheck with engine=null (supervised
+// half + cap logic + thresholds; the minion_jobs bare-worker branch is
+// Postgres-only and mirrors the queue_health subcheck-3 query covered by E2E).
+// Also pins the cross-week supervisor reader (CODEX #7).
+
+import { describe, expect, test, beforeEach, afterEach } from 'bun:test';
+import * as fs from 'fs';
+import * as os from 'os';
+import * as path from 'path';
+import { withEnv } from './helpers/with-env.ts';
+import { computeWorkerOomLoopCheck } from '../src/commands/doctor.ts';
+import {
+  computeSupervisorAuditFilename,
+  readRecentSupervisorEvents,
+} from '../src/core/minions/handlers/supervisor-audit.ts';
+
+let tmpDir: string;
+beforeEach(() => {
+  tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'worker-oom-loop-'));
+});
+afterEach(() => {
+  try { fs.rmSync(tmpDir, { recursive: true, force: true }); } catch { /* best-effort */ }
+});
+
+function writeSupervisorRows(rows: object[], fileDate = new Date()): void {
+  const file = path.join(tmpDir, computeSupervisorAuditFilename(fileDate));
+  fs.writeFileSync(file, rows.map((r) => JSON.stringify(r)).join('\n') + '\n', 'utf8');
+}
+
+const nowIso = () => new Date().toISOString();
+const rssKill = () => ({ event: 'worker_exited', ts: nowIso(), likely_cause: 'rss_watchdog', code: 12 });
+const breaker = (cap: number) => ({ event: 'health_warn', ts: nowIso(), reason: 'rss_watchdog_loop', max_rss_mb: cap });
+
+describe('computeWorkerOomLoopCheck', () => {
+  test('fail with breaker-stamped cap when the breaker tripped', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      writeSupervisorRows([rssKill(), rssKill(), rssKill(), rssKill(), rssKill(), rssKill(), breaker(2048)]);
+      const c = await computeWorkerOomLoopCheck(null);
+      expect(c?.status).toBe('fail');
+      expect(c?.name).toBe('worker_oom_loop');
+      expect(c?.message).toContain('cap=2048MB');
+      expect(c?.message).toContain('raise --max-rss');
+      expect(c?.details?.oom_kills).toBe(6);
+      expect(c?.details?.supervisor_kills).toBe(6);
+      expect(c?.details?.bare_worker_kills).toBe(0);
+      expect(c?.details?.cap_source).toBe('breaker');
+      expect(c?.details?.breaker_tripped).toBe(true);
+    });
+  });
+
+  test('warn with auto-sized cap fallback when no breaker event stamped a cap (CODEX #6)', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      writeSupervisorRows([rssKill(), rssKill()]);
+      const c = await computeWorkerOomLoopCheck(null);
+      expect(c?.status).toBe('warn');
+      expect(c?.details?.cap_source).toBe('default');
+      expect(c?.message).toContain('auto-sized default');
+      expect(c?.details?.oom_kills).toBe(2);
+    });
+  });
+
+  test('fail at >=5 kills even without a breaker event', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      writeSupervisorRows([rssKill(), rssKill(), rssKill(), rssKill(), rssKill()]);
+      const c = await computeWorkerOomLoopCheck(null);
+      expect(c?.status).toBe('fail');
+      expect(c?.details?.breaker_tripped).toBe(false);
+    });
+  });
+
+  test('null when the worker never OOM-looped', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      writeSupervisorRows([{ event: 'started', ts: nowIso() }]);
+      expect(await computeWorkerOomLoopCheck(null)).toBeNull();
+    });
+  });
+
+  test('null on an empty audit dir', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      expect(await computeWorkerOomLoopCheck(null)).toBeNull();
+    });
+  });
+});
+
+describe('readRecentSupervisorEvents — cross-week (CODEX #7)', () => {
+  test('reads a within-24h event from the PREVIOUS ISO-week file', async () => {
+    await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => {
+      // Event timestamped within the 24h window but written into last week's
+      // file (the Monday-reads-Sunday case). The single-file reader would miss
+      // it; the cross-week reader must find it.
+      const prevWeekFileDate = new Date(Date.now() - 7 * 86400000);
+      writeSupervisorRows([rssKill()], prevWeekFileDate);
+      const events = readRecentSupervisorEvents(24);
+      expect(events.length).toBe(1);
+      expect(events[0].event).toBe('worker_exited');
+    });
+  });
+});

test/dream-cli-flags.test.ts

@@ -119,9 +119,15 @@ describe('dream CLI flag wiring', () => {
       expect(dreamSrc).toContain('--drain currently supports only --phase extract_atoms');
     });
 
-    test('drain holds the same cycle lock id (contends with the routine cycle)', () => {
-      expect(dreamSrc).toContain('cycleLockIdFor(resolvedSourceId)');
-      expect(dreamSrc).toContain('withRefreshingLock');
+    test('drain routes through the shared helper with the resolved source (5A)', () => {
+      // v0.42.10.0 (#1685 GAP D / 5A): the lock+batch+count wiring moved into
+      // runExtractAtomsDrainForSource so the CLI, the Minion handler, and
+      // autopilot share ONE drain path. dream threads resolvedSourceId so the
+      // helper picks cycleLockIdFor(resolvedSourceId) — the same lock the routine
+      // cycle holds for that source. The lock-id contract is now pinned in
+      // test/extract-atoms-drain.test.ts ("shared wiring helper holds the cycle lock").
+      expect(dreamSrc).toContain('runExtractAtomsDrainForSource');
+      expect(dreamSrc).toContain('sourceId: resolvedSourceId');
     });
 
     test('drain reports remaining + exits non-zero when incomplete', () => {

test/extract-atoms-drain-handler.test.ts

@@ -0,0 +1,52 @@
+/**
+ * #1685 GAP D — extract-atoms-drain Minion handler: registration + protected
+ * gate. Canonical PGLite block (CLAUDE.md R3+R4).
+ */
+import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
+import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import { MinionQueue } from '../src/core/minions/queue.ts';
+import { MinionWorker } from '../src/core/minions/worker.ts';
+import { registerBuiltinHandlers } from '../src/commands/jobs.ts';
+
+let engine: PGLiteEngine;
+let queue: MinionQueue;
+
+beforeAll(async () => {
+  engine = new PGLiteEngine();
+  await engine.connect({ database_url: '' });
+  await engine.initSchema();
+  queue = new MinionQueue(engine);
+});
+
+afterAll(async () => {
+  await engine.disconnect();
+});
+
+beforeEach(async () => {
+  await engine.executeRaw('DELETE FROM minion_jobs');
+});
+
+describe('extract-atoms-drain handler', () => {
+  test('registerBuiltinHandlers registers the handler', async () => {
+    const worker = new MinionWorker(engine);
+    await registerBuiltinHandlers(worker, engine);
+    expect(worker.registeredNames).toContain('extract-atoms-drain');
+  });
+
+  test('queue.add rejects an untrusted submission (PROTECTED, CODEX #1)', async () => {
+    await expect(queue.add('extract-atoms-drain', { sourceId: 'default' })).rejects.toThrow(
+      /protected job name/i,
+    );
+  });
+
+  test('queue.add accepts a trusted submission (allowProtectedSubmit)', async () => {
+    const job = await queue.add(
+      'extract-atoms-drain',
+      { sourceId: 'default', window: 120 },
+      { queue: 'default' },
+      { allowProtectedSubmit: true },
+    );
+    expect(job.id).toBeGreaterThan(0);
+    expect(job.name).toBe('extract-atoms-drain');
+  });
+});

test/extract-atoms-drain.test.ts

@@ -9,10 +9,13 @@
  */
 
 import { describe, it, expect } from 'bun:test';
+import { readFileSync } from 'fs';
+import { join } from 'path';
 import {
   runExtractAtomsDrain,
   type ExtractAtomsDrainDeps,
 } from '../src/core/cycle/extract-atoms-drain.ts';
+import { isProtectedJobName, PROTECTED_JOB_NAMES } from '../src/core/minions/protected-names.ts';
 
 function seq(values: Array<number | null>): () => Promise<number | null> {
   let i = 0;
@@ -106,3 +109,28 @@ describe('runExtractAtomsDrain (issue #1678)', () => {
     expect(batches).toBe(4);
   });
 });
+
+// #1685 GAP D (CODEX #1) — the auto-drain Minion job burns Haiku, so it must be
+// PROTECTED: no MCP/OAuth-scoped caller can submit it; only trusted local
+// callers (autopilot, explicit CLI with --allow-protected) can.
+describe('extract-atoms-drain protected-name membership', () => {
+  it('extract-atoms-drain is PROTECTED', () => {
+    expect(isProtectedJobName('extract-atoms-drain')).toBe(true);
+    expect(PROTECTED_JOB_NAMES.has('extract-atoms-drain')).toBe(true);
+  });
+});
+
+// #1685 GAP D / 5A — the shared wiring helper is the single drain path. The
+// "drain holds the same cycle lock id as the routine cycle" contract (moved out
+// of dream.ts in the 5A refactor) lives here now.
+describe('shared wiring helper holds the cycle lock (5A)', () => {
+  const src = readFileSync(
+    join(import.meta.dir, '../src/core/cycle/extract-atoms-drain.ts'),
+    'utf8',
+  );
+  it('runExtractAtomsDrainForSource uses cycleLockIdFor(opts.sourceId) + withRefreshingLock', () => {
+    expect(src).toContain('runExtractAtomsDrainForSource');
+    expect(src).toContain('cycleLockIdFor(opts.sourceId)');
+    expect(src).toContain('withRefreshingLock(engine, lockId');
+  });
+});

test/worker-lock-renewal.test.ts

@@ -505,6 +505,26 @@ describe('runLockRenewalTick: reconnect-once dep (issue #1678)', () => {
     expect(state.consecutiveFailures).toBe(1);
   });
 
+  // CODEX impl review #2 (#1685 GAP B): the tick must thread the triggering
+  // renewLock error to reconnect, so PostgresEngine.reconnect can classify a
+  // CONNECTION_ENDED pooler reap as reap_detected (not reconnect_other) for
+  // pool_reap_health. Pin the threading.
+  test('reconnect receives the triggering renewLock error', async () => {
+    const audit = freshAudit();
+    const renewErr = new Error('write CONNECTION_ENDED');
+    let received: unknown = 'NOT_CALLED';
+    const deps: LockRenewalDeps = {
+      renewLock: async () => { throw renewErr; },
+      audit: audit.sink,
+      now: () => 1000,
+      setTimeout: makeFakeTimer().setTimeout,
+      reconnect: async (ctx?: { error?: unknown }) => { received = ctx?.error; },
+    };
+    const result = await runLockRenewalTick(deps, makeState({ lastSuccessfulRenewalAt: 0 }));
+    expect(result).toEqual({ kind: 'ok' });
+    expect(received).toBe(renewErr);
+  });
+
   test('a reconnect throw is swallowed — tick still returns ok (no unhandledRejection class)', async () => {
     const audit = freshAudit();
     const deps: LockRenewalDeps = {