fix(extract-conversation-facts): fan out --background per source when --source-id is omitted (#3227)

The Minion handler is single-source fail-closed (requires
data.sourceId), but the CLI --background path submitted one job with
sourceId undefined — so the exact invocation doctor prescribes
(--background --max-cost-usd 5) always produced an instantly-dead job.
Mirror the foreground multi-source loop: enumerate eligible sources and
submit one job per source, printing every job_id. Explicit --source-id,
PGLite (inline degrade), and parse errors fall through unchanged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/commands/autopilot.ts

@@ -825,7 +825,10 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
                         {
                           queue: 'default',
                           idempotency_key: idemKey,
-                          max_attempts: 1,
+                          // issue #3218: the handler now throws on an
+                          // all-provider-failed batch, so give the queue's
+                          // backoff a chance (was 1 — dead-lettered instantly).
+                          max_attempts: 3,
                           timeout_ms: timeoutMs,
                         },
                         { allowProtectedSubmit: true },

src/commands/extract-conversation-facts.ts

@@ -1318,7 +1318,7 @@ in the facts table (source='${TERMINAL_AUDIT_SOURCE}'). gbrain doctor's
 conversation_facts_backlog check counts pages without this row.
 `;
 
-function buildJobParams(args: string[]): Record<string, unknown> {
+export function buildJobParams(args: string[]): Record<string, unknown> {
   const parsed = parseArgs(args);
   return {
     sourceId: parsed.sourceId,
@@ -1340,6 +1340,56 @@ function buildJobParams(args: string[]): Record<string, unknown> {
   };
 }
 
+/**
+ * issue #3227: `--background` with no `--source-id` used to submit ONE job
+ * with `sourceId: undefined`, which the Minion handler rejects fail-closed
+ * ("requires data.sourceId") — the exact invocation `gbrain doctor`
+ * prescribes always died instantly. Fan out one job per eligible source
+ * (job_id is per-call by design), printing every job_id. The per-job
+ * `--max-cost-usd` cap matches foreground semantics ("per-source budget cap
+ * defaults to --max-cost-usd").
+ *
+ * Returns true when it handled submission (caller exits). Returns false to
+ * fall through to the plain single-job path when: no --background, PGLite
+ * (maybeBackground degrades to inline; the foreground loop already
+ * multi-sources), explicit --source-id, or a parse error (let the
+ * foreground path report it).
+ *
+ * `submit` is injectable for tests; production routes each per-source arg
+ * vector through the shared maybeBackground helper.
+ */
+export async function fanOutBackgroundBySource(
+  engine: BrainEngine,
+  args: string[],
+  submit: (perSourceArgs: string[]) => Promise<boolean> = (perSourceArgs) =>
+    maybeBackground({
+      engine,
+      args: perSourceArgs,
+      jobName: 'extract-conversation-facts',
+      paramBuilder: buildJobParams,
+    }),
+): Promise<boolean> {
+  if (!args.includes('--background') || engine.kind === 'pglite') return false;
+  const parsed = parseArgs(args.filter((a) => a !== '--background' && a !== '--follow'));
+  if (parsed.error || parsed.sourceId) return false;
+
+  if (args.includes('--follow')) {
+    process.stderr.write(
+      '[--background] --follow is skipped for multi-source fan-out; use `gbrain jobs follow <id>` per job.\n',
+    );
+  }
+  const base = args.filter((a) => a !== '--follow');
+  const sourceIds = (await listSources(engine)).map((s) => s.id);
+  if (sourceIds.length === 0) {
+    process.stderr.write('[--background] no sources found; nothing to submit.\n');
+    return true;
+  }
+  for (const id of sourceIds) {
+    await submit([...base, '--source-id', id]);
+  }
+  return true;
+}
+
 export async function runExtractConversationFacts(
   engine: BrainEngine,
   args: string[],
@@ -1350,7 +1400,13 @@ export async function runExtractConversationFacts(
     return;
   }
 
-  // --background path.
+  // --background path. issue #3227: the Minion handler is single-source
+  // fail-closed (`data.sourceId` required), but doctor prescribes
+  // `--background` with no --source-id — that submission produced an
+  // instantly-dead job. Mirror the foreground multi-source loop by fanning
+  // out one job per source before the plain single-job path runs.
+  const fannedOut = await fanOutBackgroundBySource(engine, args);
+  if (fannedOut) return;
   const backgrounded = await maybeBackground({
     engine,
     args,

src/commands/jobs.ts

@@ -2061,11 +2061,26 @@ export async function registerBuiltinHandlers(
         ? job.data.repoPath
         : ((await engine.getConfig('sync.repo_path')) ?? undefined);
     try {
-      return await runExtractAtomsDrainForSource(engine, {
+      const result = await runExtractAtomsDrainForSource(engine, {
         sourceId,
         windowSeconds,
         brainDir: repoPath,
       });
+      // issue #3218: every item the drain attempted failed (0 succeeded, >=1
+      // provider error) — completing this job normally would mark the
+      // durable job done while the backlog sits untouched, and no retry
+      // policy would ever fire on it again. Throw so the worker's ordinary
+      // failJob path (attempt+backoff, or dead-letter once exhausted) takes
+      // over instead — matching the existing behavior for every other
+      // handler failure. Partial success (>=1 item extracted) keeps
+      // completing normally, unchanged.
+      if (result.status === 'provider_failure') {
+        throw new Error(
+          `extract-atoms-drain: all provider calls failed this batch ` +
+          `(batches=${result.batches}, remaining=${result.remaining ?? '?'}) — retrying`,
+        );
+      }
+      return result;
     } catch (e) {
       if (e instanceof LockUnavailableError) {
         return { phase: 'extract_atoms', status: 'skipped', deferred: true, reason: 'cycle_already_running' };

src/core/brain-writer.ts

@@ -139,8 +139,21 @@ export function autoFixFrontmatter(
     fixes.push({ code: 'NULL_BYTES', description: 'Stripped null bytes' });
   }
 
-  // 2. MISSING_CLOSE — if there's an opener but no closer before a heading,
-  //    insert `---` immediately before the heading. Walk lines once.
+  // 2. MISSING_CLOSE — if there's an opener but no closer at all, insert
+  //    `---` immediately before the first heading-shaped line (best-effort
+  //    guess at where the frontmatter was meant to end).
+  //
+  //    Find the closer FIRST, scanning the full zone — do not stop at the
+  //    first `#`-prefixed line. A `#` line between the opening and closing
+  //    `---` is a YAML comment (comments are valid anywhere in a YAML
+  //    document), not a markdown heading; only the genuine absence of a
+  //    closing `---` counts as MISSING_CLOSE. Mirrors the fix applied to
+  //    the parseMarkdown validator in #2153 — this is the sibling
+  //    reimplementation in the auto-fixer and had the same bug (it broke
+  //    out of the scan on the first heading-shaped line, so a `#` comment
+  //    appearing before a real closing fence was misdetected as
+  //    MISSING_CLOSE and the fix inserted a spurious `---` that split
+  //    valid frontmatter in two, pushing the real keys into the body).
   {
     const lines = working.split('\n');
     let firstNonEmpty = -1;
@@ -149,24 +162,27 @@ export function autoFixFrontmatter(
     }
     if (firstNonEmpty >= 0 && lines[firstNonEmpty].trim() === '---') {
       let closeIdx = -1;
-      let headingIdx = -1;
       for (let i = firstNonEmpty + 1; i < lines.length; i++) {
-        const t = lines[i].trim();
-        if (t === '---') { closeIdx = i; break; }
-        if (/^#{1,6}\s/.test(t)) { headingIdx = i; break; }
+        if (lines[i].trim() === '---') { closeIdx = i; break; }
       }
-      if (closeIdx === -1 && headingIdx >= 0) {
-        const fixed = [
-          ...lines.slice(0, headingIdx),
-          '---',
-          '',
-          ...lines.slice(headingIdx),
-        ];
-        working = fixed.join('\n');
-        fixes.push({
-          code: 'MISSING_CLOSE',
-          description: `Inserted closing --- before heading at line ${headingIdx + 1}`,
-        });
+      if (closeIdx === -1) {
+        let headingIdx = -1;
+        for (let i = firstNonEmpty + 1; i < lines.length; i++) {
+          if (/^#{1,6}\s/.test(lines[i].trim())) { headingIdx = i; break; }
+        }
+        if (headingIdx >= 0) {
+          const fixed = [
+            ...lines.slice(0, headingIdx),
+            '---',
+            '',
+            ...lines.slice(headingIdx),
+          ];
+          working = fixed.join('\n');
+          fixes.push({
+            code: 'MISSING_CLOSE',
+            description: `Inserted closing --- before heading at line ${headingIdx + 1}`,
+          });
+        }
       }
     }
   }

src/core/cycle/extract-atoms-drain.ts

@@ -33,8 +33,14 @@ export interface ExtractAtomsDrainDeps {
    * routine cycle's skip contract.
    */
   withLock: <T>(work: () => Promise<T>) => Promise<T>;
-  /** Process one bounded batch (rediscovers eligibility). Returns counts. */
-  runBatch: () => Promise<{ extracted: number; skipped: number }>;
+  /**
+   * Process one bounded batch (rediscovers eligibility). Returns counts, plus
+   * `providerFailure` (issue #3218) when EVERY item the batch attempted threw
+   * (zero items succeeded, at least one failure) — i.e. the batch's warning
+   * result was actually a total provider outage, not a partial/no-op batch.
+   * Omit/false for the ordinary partial-success or nothing-to-do cases.
+   */
+  runBatch: () => Promise<{ extracted: number; skipped: number; providerFailure?: boolean }>;
   /** Count remaining eligible-but-unextracted pages, or null on query error. */
   countRemaining: () => Promise<number | null>;
   /** Injectable clock. Production: Date.now. */
@@ -52,15 +58,22 @@ export interface ExtractAtomsDrainOpts {
 
 export interface ExtractAtomsDrainResult {
   phase: 'extract_atoms';
-  status: 'ok';
+  /**
+   * issue #3218: 'provider_failure' when any batch reported `providerFailure`
+   * (every item it attempted errored). The Minion handler throws on this
+   * status so the durable job retries instead of completing over a backlog
+   * that made zero forward progress. Partial-success batches (>=1 item
+   * succeeded) always report 'ok', unchanged from before.
+   */
+  status: 'ok' | 'provider_failure';
   extracted: number;
   skipped: number;
   /** Eligible pages still pending after the window. null if the count errored. */
   remaining: number | null;
   /** Batches actually processed. */
   batches: number;
-  /** Why the loop stopped: drained | window | no_progress | max_batches. */
-  stopped: 'drained' | 'window' | 'no_progress' | 'max_batches';
+  /** Why the loop stopped: drained | window | no_progress | max_batches | provider_failure. */
+  stopped: 'drained' | 'window' | 'no_progress' | 'max_batches' | 'provider_failure';
 }
 
 export async function runExtractAtomsDrain(
@@ -74,6 +87,10 @@ export async function runExtractAtomsDrain(
     let skipped = 0;
     let batches = 0;
     let stopped: ExtractAtomsDrainResult['stopped'] = 'window';
+    // issue #3218: latched once any batch reports providerFailure — drives
+    // the returned `status`, independent of how `stopped` reads after the
+    // final (possibly overriding) remaining-count check below.
+    let providerFailure = false;
 
     while (deps.now() < deadline) {
       if (batches >= maxBatches) { stopped = 'max_batches'; break; }
@@ -87,6 +104,17 @@ export async function runExtractAtomsDrain(
       batches++;
       deps.onBatch?.({ batch: batches, extracted: r.extracted, remaining: before });
 
+      // issue #3218: every item this batch attempted failed (0 succeeded, >=1
+      // error) — a total provider outage, not ordinary no-op/partial progress.
+      // Stop immediately (same hot-loop guard as no_progress below) and flag
+      // it so the caller can retry via its own policy instead of treating the
+      // drain as a clean completion.
+      if (r.providerFailure) {
+        providerFailure = true;
+        stopped = 'provider_failure';
+        break;
+      }
+
       // Stop if a batch made zero forward progress — extraction is failing or
       // everything left is ineligible (e.g. all skipped). Prevents a hot loop
       // that spends budget without draining.
@@ -94,8 +122,22 @@ export async function runExtractAtomsDrain(
     }
 
     const remaining = await deps.countRemaining();
-    if (remaining === 0) stopped = 'drained';
-    return { phase: 'extract_atoms', status: 'ok', extracted, skipped, remaining, batches, stopped };
+    // issue #3218 (codex P2): don't let a final remaining===0 recount
+    // overwrite 'provider_failure' back to 'drained' — that would report the
+    // contradictory {status: 'provider_failure', stopped: 'drained'} and
+    // mislead the CLI/JSON consumer (dream.ts prints both fields verbatim).
+    // status already takes precedence for the Minion handler's retry
+    // decision; keep `stopped` consistent with it once a failure latched.
+    if (!providerFailure && remaining === 0) stopped = 'drained';
+    return {
+      phase: 'extract_atoms',
+      status: providerFailure ? 'provider_failure' : 'ok',
+      extracted,
+      skipped,
+      remaining,
+      batches,
+      stopped,
+    };
   });
 }
 
@@ -157,9 +199,22 @@ export async function runExtractAtomsDrainForSource(
           brainDir: opts.brainDir,
         });
         const d = (r.details ?? {}) as Record<string, unknown>;
+        // issue #3218: `r.status` collapses to 'warn' whether ONE item failed
+        // (partial success — leave the drain's existing ok/no_progress path
+        // alone) or EVERY item failed (a total provider outage the drain
+        // adapter was silently swallowing). Re-derive the total-failure case
+        // from the per-item counts `runPhaseExtractAtoms` already returns:
+        // >=1 failure AND zero items successfully processed (transcripts_processed
+        // + pages_processed both 0 means every attempted `chat()` call threw —
+        // items that succeed with 0 atoms still count as processed, so this
+        // does not fire on "provider fine, nothing extractable").
+        const failures = Array.isArray(d.failures) ? d.failures : [];
+        const itemsSucceeded =
+          Number(d.transcripts_processed ?? 0) + Number(d.pages_processed ?? 0);
         return {
           extracted: Number(d.atoms_extracted ?? 0),
           skipped: Number(d.duplicates_skipped ?? 0),
+          providerFailure: failures.length > 0 && itemsSucceeded === 0,
         };
       },
       countRemaining: () => countExtractAtomsBacklog(engine, extractionSourceId),

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

@@ -43,6 +43,19 @@ describe('autopilot auto-drain wiring', () => {
     expect(SRC).toMatch(/engine\.kind === 'postgres'[\s\S]{0,400}auto_drain/);
   });
 
+  // issue #3218 (codex P1): with the handler now throwing on an
+  // all-provider-failed batch, max_attempts:1 made the queue's retry policy
+  // "dead-letter on the first failure, no backoff attempt" — regression-guard
+  // against silently reverting to 1.
+  test('issue #3218: submits with max_attempts 3 (not 1) so a retry can backoff before dead-lettering', () => {
+    // lastIndexOf: the queue.add(...) call site itself (the earlier occurrence
+    // is the unrelated created_at count query above it in the same function).
+    const callSite = SRC.lastIndexOf("'extract-atoms-drain'");
+    const drainBlock = SRC.slice(callSite, callSite + 900);
+    expect(drainBlock).toContain('max_attempts: 3');
+    expect(drainBlock).not.toContain('max_attempts: 1');
+  });
+
   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

test/brain-writer.test.ts

@@ -32,6 +32,43 @@ describe('autoFixFrontmatter', () => {
     expect(idxClose).toBeLessThan(idxHeading);
   });
 
+  // Regression for #3225: a `#`-prefixed line inside an already-closed
+  // frontmatter fence is a YAML comment, not a markdown heading. The old
+  // MISSING_CLOSE scan broke out on the first heading-shaped line without
+  // continuing to look for the real closer, so it inserted a spurious
+  // `---` before the comment and split valid frontmatter in two — pushing
+  // the real keys (title, pubDate, ...) into the document body.
+  test('does not corrupt closed frontmatter containing a YAML comment line', () => {
+    const input = `${fence}\n# a YAML comment inside the frontmatter block\ntitle: "Real Title"\npubDate: 2026-06-29\n${fence}\nBody...`;
+    const { content, fixes } = autoFixFrontmatter(input);
+    expect(content).toBe(input);
+    expect(fixes).toEqual([]);
+  });
+
+  test('does not corrupt closed frontmatter that is comment-only', () => {
+    const input = `${fence}\n# just a comment\n# another comment\n${fence}\nBody`;
+    const { content, fixes } = autoFixFrontmatter(input);
+    expect(content).toBe(input);
+    expect(fixes).toEqual([]);
+  });
+
+  test('does not corrupt closed frontmatter with an indented `#` line inside a YAML block scalar', () => {
+    const input = `${fence}\ndescription: |\n  # not a heading, just literal block-scalar text\ntitle: ok\n${fence}\nBody`;
+    const { content, fixes } = autoFixFrontmatter(input);
+    expect(content).toBe(input);
+    expect(fixes).toEqual([]);
+  });
+
+  test('YAML comment before close does not suppress an unrelated real fix (SLUG_MISMATCH)', () => {
+    const input = `${fence}\n# a YAML comment\ntitle: hi\nslug: wrong-slug\n${fence}\nBody`;
+    const { content, fixes } = autoFixFrontmatter(input, { filePath: 'people/jane-doe.md' });
+    expect(fixes.some(f => f.code === 'MISSING_CLOSE')).toBe(false);
+    expect(fixes.some(f => f.code === 'SLUG_MISMATCH')).toBe(true);
+    // The frontmatter fence itself must stay intact — only the slug line
+    // is removed, the comment/title/close survive unchanged.
+    expect(content).toBe(`${fence}\n# a YAML comment\ntitle: hi\n\n${fence}\nBody`);
+  });
+
   test('rewrites nested-quote title to single-quoted', () => {
     const input = `${fence}\ntype: concept\ntitle: "Phil "Nick" Last"\n${fence}\n\nbody`;
     const { content, fixes } = autoFixFrontmatter(input);

test/cycle/extract-atoms-synthesize-concepts.test.ts

@@ -177,6 +177,31 @@ describe('v0.41 T5: runPhaseExtractAtoms via stubbed chat', () => {
     expect((result.details?.failures as unknown[]).length).toBe(1);
   });
 
+  // issue #3218 — when EVERY item's chat() call throws (all-provider-failed),
+  // `transcripts_processed`/`pages_processed` must stay 0 while `failures`
+  // records one entry per item. This is the exact shape the
+  // extract-atoms-drain wiring (`runExtractAtomsDrainForSource`) uses to
+  // derive `providerFailure` (failures.length > 0 && itemsSucceeded === 0),
+  // distinguishing a total outage from the partial-success case above.
+  test('all items fail: transcripts_processed/pages_processed stay 0, every item recorded in failures', async () => {
+    const chat = async (_o: ChatOpts): Promise<never> => {
+      throw new Error('provider unavailable');
+    };
+    const result = await runPhaseExtractAtoms(engine, {
+      _transcripts: [
+        { filePath: '/a.txt', content: 'a', contentHash: 'ha' },
+        { filePath: '/b.txt', content: 'b', contentHash: 'hb' },
+      ],
+      _pages: [],
+      _chat: chat as typeof import('../../src/core/ai/gateway.ts').chat,
+    });
+    expect(result.status).toBe('warn');
+    expect(result.details?.atoms_extracted).toBe(0);
+    expect(result.details?.transcripts_processed).toBe(0);
+    expect(result.details?.pages_processed).toBe(0);
+    expect((result.details?.failures as unknown[]).length).toBe(2);
+  });
+
   // v0.41.2.1 regression case (D9 #14 wording): with _pages:[] and same
   // _transcripts, all PRE-EXISTING PhaseResult.details fields match
   // pre-fix values byte-for-byte. The new fields (pages_processed,

test/extract-atoms-drain.test.ts

@@ -77,6 +77,62 @@ describe('runExtractAtomsDrain (issue #1678)', () => {
     expect(result.stopped).toBe('no_progress');
     expect(batches).toBe(1);
     expect(result.remaining).toBe(5);
+    expect(result.status).toBe('ok');
+  });
+
+  // issue #3218 — a batch where every attempted item errored (providerFailure)
+  // must surface distinctly from an ordinary no_progress/drained/window stop,
+  // so the Minion handler can retry instead of completing the durable job.
+  it('stops with status=provider_failure when a batch reports providerFailure', async () => {
+    let batches = 0;
+    const result = await runExtractAtomsDrain(
+      {
+        withLock: passThroughLock,
+        countRemaining: async () => 5,
+        runBatch: async () => {
+          batches++;
+          return { extracted: 0, skipped: 0, providerFailure: true };
+        },
+        now: () => 0,
+      },
+      { windowMs: 1_000_000 },
+    );
+    expect(result.status).toBe('provider_failure');
+    expect(result.stopped).toBe('provider_failure');
+    expect(batches).toBe(1);
+    expect(result.remaining).toBe(5);
+  });
+
+  // issue #3218 (codex P2) — a final recount of 0 must NOT overwrite
+  // stopped='provider_failure' back to 'drained'. Otherwise the caller sees
+  // the contradictory {status: 'provider_failure', stopped: 'drained'}.
+  it('preserves stopped=provider_failure even when the final recount is 0', async () => {
+    const result = await runExtractAtomsDrain(
+      {
+        withLock: passThroughLock,
+        countRemaining: seq([3, 0]), // before-check: 3; final post-loop recount: 0
+        runBatch: async () => ({ extracted: 0, skipped: 0, providerFailure: true }),
+        now: () => 0,
+      },
+      { windowMs: 1_000_000 },
+    );
+    expect(result.status).toBe('provider_failure');
+    expect(result.stopped).toBe('provider_failure');
+    expect(result.remaining).toBe(0);
+  });
+
+  it('does not flag provider_failure for an ordinary partial-success batch', async () => {
+    const result = await runExtractAtomsDrain(
+      {
+        withLock: passThroughLock,
+        countRemaining: seq([3, 0, 0]),
+        runBatch: async () => ({ extracted: 1, skipped: 0, providerFailure: false }),
+        now: () => 0,
+      },
+      { windowMs: 1_000_000 },
+    );
+    expect(result.status).toBe('ok');
+    expect(result.stopped).toBe('drained');
   });
 
   it('propagates a busy-lock error (caller reports cycle_already_running)', async () => {
@@ -133,4 +189,42 @@ describe('shared wiring helper holds the cycle lock (5A)', () => {
     expect(src).toContain('cycleLockIdFor(opts.sourceId)');
     expect(src).toContain('withRefreshingLock(engine, lockId');
   });
+
+  // issue #3218 — the wiring's `runBatch` must derive `providerFailure` from
+  // the SAME per-item counts pinned by
+  // `extract-atoms-synthesize-concepts.test.ts`'s "all items fail" case
+  // (failures.length > 0 && transcripts_processed + pages_processed === 0),
+  // not from `r.status` (which collapses partial and total failure into the
+  // same 'warn' value — the exact discard the issue reports).
+  it('runBatch derives providerFailure from failures.length + zero processed items, not r.status', () => {
+    const runBatchBlock = src.slice(src.indexOf('runBatch: async () => {'));
+    expect(runBatchBlock).toContain('d.failures');
+    expect(runBatchBlock).toContain('transcripts_processed');
+    expect(runBatchBlock).toContain('pages_processed');
+    expect(runBatchBlock).toContain('providerFailure: failures.length > 0 && itemsSucceeded === 0');
+  });
+});
+
+// issue #3218 — the Minion handler must throw (not complete) when the drain
+// reports status='provider_failure', so the worker's ordinary failJob path
+// (attempt+backoff / dead-letter) retries the durable job instead of the
+// backlog silently completing untouched.
+describe('extract-atoms-drain Minion handler retries on provider_failure (issue #3218)', () => {
+  const jobsSrc = readFileSync(join(import.meta.dir, '../src/commands/jobs.ts'), 'utf8');
+  const handlerBlock = jobsSrc.slice(
+    jobsSrc.indexOf("registerBuiltinJob(worker, engine, 'extract-atoms-drain'"),
+    jobsSrc.indexOf("registerBuiltinJob(worker, engine, 'extract-atoms-drain'") + 2200,
+  );
+
+  it("throws when result.status === 'provider_failure' instead of returning it", () => {
+    expect(handlerBlock).toMatch(/result\.status === 'provider_failure'/);
+    expect(handlerBlock).toMatch(/if \(result\.status === 'provider_failure'\) \{\s*throw new Error/);
+  });
+
+  it('still returns the deferred/skipped shape on LockUnavailableError (unchanged)', () => {
+    expect(handlerBlock).toContain('e instanceof LockUnavailableError');
+    expect(handlerBlock).toContain(
+      "{ phase: 'extract_atoms', status: 'skipped', deferred: true, reason: 'cycle_already_running' }",
+    );
+  });
 });

test/extract-conversation-facts.test.ts

@@ -37,6 +37,8 @@ import {
   TERMINAL_AUDIT_SOURCE,
   PER_SEGMENT_SOURCE_PREFIX,
   ALLOWED_TYPES,
+  fanOutBackgroundBySource,
+  buildJobParams,
 } from '../src/commands/extract-conversation-facts.ts';
 
 // ---------------------------------------------------------------------------
@@ -554,3 +556,81 @@ describe('body cap constant (Eng A2)', () => {
     expect(MAX_PAGE_BODY_BYTES).toBe(25 * 1024 * 1024);
   });
 });
+
+// issue #3227 — `--background` with no --source-id used to submit a single
+// job with sourceId undefined, which the Minion handler rejects fail-closed
+// ("requires data.sourceId"): the exact invocation doctor prescribes
+// (`gbrain extract-conversation-facts --background --max-cost-usd 5`)
+// always produced an instantly-dead job. The CLI now fans out one job per
+// source. Hermetic: stub engine + injectable submit.
+describe('fanOutBackgroundBySource (issue #3227)', () => {
+  const stubEngine = (sourceIds: string[]) =>
+    ({
+      kind: 'postgres',
+      executeRaw: async (sql: string) =>
+        sql.includes('FROM sources')
+          ? sourceIds.map((id) => ({
+              id,
+              name: id,
+              local_path: null,
+              last_sync_at: null,
+              config: {},
+            }))
+          : [{ n: 0 }],
+    }) as unknown as import('../src/core/engine.ts').BrainEngine;
+
+  test('fans out one submission per source, each pinned with --source-id', async () => {
+    const submitted: string[][] = [];
+    const handled = await fanOutBackgroundBySource(
+      stubEngine(['default', 'media']),
+      ['--background', '--max-cost-usd', '5'],
+      async (a) => { submitted.push(a); return true; },
+    );
+    expect(handled).toBe(true);
+    expect(submitted.length).toBe(2);
+    for (const a of submitted) expect(a).toContain('--background');
+    expect(submitted[0].slice(-2)).toEqual(['--source-id', 'default']);
+    expect(submitted[1].slice(-2)).toEqual(['--source-id', 'media']);
+  });
+
+  test('every fanned-out submission carries a defined sourceId in the job params', async () => {
+    // The load-bearing contract: the handler throws on missing data.sourceId.
+    const ids: unknown[] = [];
+    await fanOutBackgroundBySource(
+      stubEngine(['default', 'media']),
+      ['--background', '--max-cost-usd', '5'],
+      async (a) => {
+        ids.push(buildJobParams(a.filter((x) => x !== '--background')).sourceId);
+        return true;
+      },
+    );
+    expect(ids).toEqual(['default', 'media']);
+  });
+
+  test('explicit --source-id falls through to the plain single-job path', async () => {
+    const submitted: string[][] = [];
+    const handled = await fanOutBackgroundBySource(
+      stubEngine(['default', 'media']),
+      ['--background', '--source-id', 'media'],
+      async (a) => { submitted.push(a); return true; },
+    );
+    expect(handled).toBe(false);
+    expect(submitted.length).toBe(0);
+  });
+
+  test('no --background falls through', async () => {
+    const handled = await fanOutBackgroundBySource(
+      stubEngine(['default']),
+      ['--max-cost-usd', '5'],
+      async () => true,
+    );
+    expect(handled).toBe(false);
+  });
+
+  test('pglite falls through (maybeBackground degrades to inline multi-source foreground)', async () => {
+    const eng = stubEngine(['default']);
+    (eng as { kind: string }).kind = 'pglite';
+    const handled = await fanOutBackgroundBySource(eng, ['--background'], async () => true);
+    expect(handled).toBe(false);
+  });
+});