fix(extract,ingest,cycle): source-provenance wave — thread source identity through ingest_capture, fs-walk links, and cycle extract (#1522 #1747 #1503)

Three fixes in the same invariant class (source identity silently dropped
on the write path, collapsing to the 'default' source):

- #1522: the ingest_capture Minion handler validated IngestionEvent
  provenance (source_id/source_kind/source_uri) then dropped it on the
  importFromContent call. Now threads source_kind/source_uri +
  ingested_via='ingest_capture' into the page write, and routes the write
  under event.source_id when it names a registered source AND the event
  is trusted (fail-closed: untrusted webhook payloads carry a
  caller-controlled x-gbrain-source-id header and must not choose their
  write source; unregistered emitter ids keep default routing so the
  webhook path can't FK-fail).

- #1747: the fs-walk extractors (extractLinksFromDir /
  extractTimelineFromDir / extractForSlugs) built batch rows with no
  source_id, so addLinksBatch/addTimelineEntriesBatch mapped missing →
  'default' and the pages JOIN dropped every row on a non-default source
  ("Links: created 0 from N pages", no error). ExtractOpts gains
  sourceId; the CLI fs path resolves it via resolveSourceId
  (--source-id > env > dotfile > registered path > sole-non-default)
  and rows are stamped from/to/origin_source_id + timeline source_id.

- #1503: the cycle's extract phase (runPhaseExtract) never passed a
  sourceId, so federated-brain dream/autopilot cycles persisted nothing
  every night. It now threads cycleSourceId (explicit --source or
  resolveSourceForDir(brainDir) — the same seam runPhaseSync uses) into
  runExtractCore for both the incremental and full-walk paths.

Regression tests: handler provenance write-through (registered/
unregistered/untrusted routing), fs-walk + CLI resolution + incremental
cycle route landing edges/timeline in the right source (all red on
unfixed master), and a negative control pinning the pre-fix JOIN-drop
shape.

Reuses the ExtractOpts.sourceId threading approach from PR #1719,
rebased onto the current signal-aware signatures and extended to the
cycle + timeline paths.

Fixes #1522
Fixes #1747
Fixes #1503

Co-authored-by: seungsu <kss530c@gmail.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/commands/extract.ts

@@ -568,6 +568,20 @@ export interface ExtractOpts {
    * paths: extractForSlugs, extractLinksFromDir, extractTimelineFromDir.
    */
   signal?: AbortSignal;
+  /**
+   * Brain source id to stamp on extracted fs-walk rows (#1747 / #1503).
+   *
+   * The fs-walk extractors build LinkBatchInput / TimelineBatchInput rows
+   * with no source_id, so addLinksBatch / addTimelineEntriesBatch map
+   * missing → literal 'default'. On a brain whose content lives in a
+   * non-'default' source (e.g. 'wiki'), the batch INSERT's
+   * `JOIN pages ON (slug, source_id='default')` drops EVERY row → 0
+   * inserted, no error (the "created 0 from N pages" silent no-op).
+   * Threading the resolved source id here stamps from/to/origin_source_id
+   * so the JOIN matches. When undefined, rows fall back to 'default' as
+   * before (single-'default'-source brains unaffected).
+   */
+  sourceId?: string;
 }
 
 /**
@@ -606,7 +620,7 @@ export async function runExtractCore(engine: BrainEngine, opts: ExtractOpts): Pr
       // Nothing changed — skip entirely.
       return result;
     }
-    const r = await extractForSlugs(engine, opts.dir, opts.slugs, opts.mode, dryRun, jsonMode, workers, opts.signal);
+    const r = await extractForSlugs(engine, opts.dir, opts.slugs, opts.mode, dryRun, jsonMode, workers, opts.signal, opts.sourceId);
     result.links_created = r.links_created;
     result.timeline_entries_created = r.timeline_created;
     result.pages_processed = r.pages;
@@ -615,12 +629,12 @@ export async function runExtractCore(engine: BrainEngine, opts: ExtractOpts): Pr
 
   // Full walk path: CLI `gbrain extract` or first-run.
   if (opts.mode === 'links' || opts.mode === 'all') {
-    const r = await extractLinksFromDir(engine, opts.dir, dryRun, jsonMode, workers, opts.signal);
+    const r = await extractLinksFromDir(engine, opts.dir, dryRun, jsonMode, workers, opts.signal, opts.sourceId);
     result.links_created = r.created;
     result.pages_processed = r.pages;
   }
   if (opts.mode === 'timeline' || opts.mode === 'all') {
-    const r = await extractTimelineFromDir(engine, opts.dir, dryRun, jsonMode, workers, opts.signal);
+    const r = await extractTimelineFromDir(engine, opts.dir, dryRun, jsonMode, workers, opts.signal, opts.sourceId);
     result.timeline_entries_created = r.created;
     result.pages_processed = Math.max(result.pages_processed, r.pages);
   }
@@ -941,11 +955,21 @@ Status (v0.42):
         }
       }
     } else {
+      // #1747: resolve the brain source id and thread it into the fs-walk
+      // extractors so batch rows carry from/to_source_id. Without this they
+      // default to 'default' and addLinksBatch's JOIN drops every row on a
+      // non-'default' brain → silent "created 0 from N pages". Resolution
+      // honors --source-id, then GBRAIN_SOURCE / .gbrain-source /
+      // registered-path / sole-non-default, mirroring the source-aware
+      // inline hooks (extractLinksForSlugs) that #1204 confirmed correct.
+      const { resolveSourceId } = await import('../core/source-resolver.ts');
+      const resolvedSourceId = await resolveSourceId(engine, sourceIdFilter, brainDir);
       result = await runExtractCore(engine, {
         mode: subcommand as 'links' | 'timeline' | 'all',
         dir: brainDir,
         dryRun,
         jsonMode,
+        sourceId: resolvedSourceId,
         workers,
       });
     }
@@ -985,6 +1009,8 @@ async function extractForSlugs(
   // shared counter increments atomic.
   workers: number = 1,
   signal?: AbortSignal,
+  // #1747/#1503: stamp resolved brain source id on batch rows (see ExtractOpts.sourceId).
+  sourceId?: string,
 ): Promise<{ links_created: number; timeline_created: number; pages: number }> {
   // Build the full slug set for link resolution (fast: just readdir, no file reads)
   const allFiles = walkMarkdownFiles(brainDir);
@@ -1065,7 +1091,9 @@ async function extractForSlugs(
               if (!jsonMode) console.log(`  ${link.from_slug} → ${link.to_slug} (${link.link_type})`);
               linksCreated++;
             } else {
-              linkBatch.push(link);
+              linkBatch.push(sourceId
+                ? { ...link, from_source_id: sourceId, to_source_id: sourceId, origin_source_id: sourceId }
+                : link);
               if (linkBatch.length >= BATCH_SIZE) await flushLinks();
             }
           }
@@ -1078,7 +1106,7 @@ async function extractForSlugs(
               if (!jsonMode) console.log(`  ${entry.slug}: ${entry.date} — ${entry.summary}`);
               timelineCreated++;
             } else {
-              timelineBatch.push({ slug: entry.slug, date: entry.date, source: entry.source, summary: entry.summary, detail: entry.detail });
+              timelineBatch.push({ slug: entry.slug, date: entry.date, source: entry.source, summary: entry.summary, detail: entry.detail, ...(sourceId ? { source_id: sourceId } : {}) });
               if (timelineBatch.length >= BATCH_SIZE) await flushTimeline();
             }
           }
@@ -1107,6 +1135,9 @@ async function extractLinksFromDir(
   // v0.41.15.0 (T7): in-process worker count. Default 1.
   workers: number = 1,
   signal?: AbortSignal,
+  // #1747/#1503: stamp resolved brain source id on batch rows so the
+  // addLinksBatch JOIN matches non-'default' source pages.
+  sourceId?: string,
 ): Promise<{ created: number; pages: number }> {
   const files = walkMarkdownFiles(brainDir);
   const allSlugs = new Set(files.map(f => pathToSlug(f.relPath)));
@@ -1163,7 +1194,9 @@ async function extractLinksFromDir(
             if (!jsonMode) console.log(`  ${link.from_slug} → ${link.to_slug} (${link.link_type})`);
             created++;
           } else {
-            batch.push(link);
+            batch.push(sourceId
+              ? { ...link, from_source_id: sourceId, to_source_id: sourceId, origin_source_id: sourceId }
+              : link);
             if (batch.length >= BATCH_SIZE) await flush();
           }
         }
@@ -1186,6 +1219,9 @@ async function extractTimelineFromDir(
   // v0.41.15.0 (T7): in-process worker count. Default 1.
   workers: number = 1,
   signal?: AbortSignal,
+  // #1747/#1503: stamp resolved brain source id so addTimelineEntriesBatch
+  // matches non-'default' source pages.
+  sourceId?: string,
 ): Promise<{ created: number; pages: number }> {
   const files = walkMarkdownFiles(brainDir);
 
@@ -1232,7 +1268,7 @@ async function extractTimelineFromDir(
             if (!jsonMode) console.log(`  ${entry.slug}: ${entry.date} — ${entry.summary}`);
             created++;
           } else {
-            batch.push({ slug: entry.slug, date: entry.date, source: entry.source, summary: entry.summary, detail: entry.detail });
+            batch.push({ slug: entry.slug, date: entry.date, source: entry.source, summary: entry.summary, detail: entry.detail, ...(sourceId ? { source_id: sourceId } : {}) });
             if (batch.length >= BATCH_SIZE) await flush();
           }
         }

src/core/cycle.ts

@@ -956,6 +956,12 @@ async function runPhaseExtract(
   dryRun: boolean,
   changedSlugs?: string[],
   signal?: AbortSignal,
+  // #1503: the brain source the cycle is scoped to (cycleSourceId — explicit
+  // --source or resolved from brainDir). Threaded to runExtractCore so
+  // fs-walk link/timeline rows carry source_id; without it addLinksBatch maps
+  // missing → 'default' and its pages JOIN drops every row on a federated
+  // brain ("Links: created 0 from N pages" every cycle).
+  sourceId?: string,
 ): Promise<PhaseResult> {
   try {
     const { runExtractCore } = await import('../commands/extract.ts');
@@ -978,6 +984,7 @@ async function runPhaseExtract(
       dir: brainDir,
       slugs: changedSlugs,  // undefined = full walk (first run / manual)
       signal,
+      sourceId,
     });
     const linksCreated = result?.links_created ?? 0;
     const timelineCreated = result?.timeline_entries_created ?? 0;
@@ -1706,7 +1713,7 @@ export async function runCycle(
         // If sync didn't run (phases exclude it) or failed, syncPagesAffected
         // is undefined → extract falls back to full walk (safe default).
         progress.start('cycle.extract');
-        const { result, duration_ms } = await timePhase(() => runPhaseExtract(engine, brainDir, dryRun, syncPagesAffected, opts.signal));
+        const { result, duration_ms } = await timePhase(() => runPhaseExtract(engine, brainDir, dryRun, syncPagesAffected, opts.signal, cycleSourceId));
         result.duration_ms = duration_ms;
         phaseResults.push(result);
         progress.finish();

src/core/minions/handlers/ingest-capture.ts

@@ -113,7 +113,36 @@ export function makeIngestCaptureHandler(engine: BrainEngine) {
     // by passing { noEmbed: false } in job.data.
     const noEmbed = (data as { noEmbed?: unknown }).noEmbed !== false;
 
-    const result = await importFromContent(engine, slug, event.content, { noEmbed });
+    // #1522: thread the validated event's provenance into the page write
+    // instead of dropping it on the floor. source_kind / source_uri are
+    // pure provenance strings (no scoping power) and persist
+    // unconditionally via importFromContent's putPage write-through.
+    //
+    // event.source_id is the emitter's IngestionSource instance id, NOT
+    // necessarily a registered brain source (the webhook path fabricates
+    // `webhook-<clientId>`, which pages.source_id's FK would reject). It
+    // routes the page write only when BOTH hold:
+    //   - the event is trusted (fail-closed: an untrusted webhook payload
+    //     carries a caller-controlled x-gbrain-source-id header and must
+    //     not get to choose its write source), AND
+    //   - the id names a registered source row.
+    // Otherwise the write keeps the pre-fix default-source routing.
+    let sourceId: string | undefined;
+    if (!untrustedPayload) {
+      const rows = await engine.executeRaw<{ id: string }>(
+        `SELECT id FROM sources WHERE id = $1`,
+        [event.source_id],
+      );
+      if (rows.length > 0) sourceId = event.source_id;
+    }
+
+    const result = await importFromContent(engine, slug, event.content, {
+      noEmbed,
+      sourceId,
+      source_kind: event.source_kind,
+      source_uri: event.source_uri,
+      ingested_via: 'ingest_capture',
+    });
 
     return {
       slug,

test/cycle-extract-source.test.ts

@@ -0,0 +1,114 @@
+/**
+ * #1503 — the cycle's extract phase threads the resolved per-source id.
+ *
+ * Pre-fix, runPhaseExtract called runExtractCore with no sourceId, so on a
+ * federated brain (content in a non-'default' source) the fs-walk batch rows
+ * mapped to source_id='default', the pages JOIN dropped every row, and every
+ * dream/autopilot cycle logged "Links: created 0 from N pages" while
+ * persisting nothing. This pins that the cycle resolves the source from the
+ * brain dir (resolveSourceForDir — same seam runPhaseSync uses) and that the
+ * extracted link/timeline rows land in that source.
+ *
+ * GBRAIN_HOME is isolated per test because the PGLite cycle path takes a
+ * file lock at ~/.gbrain/cycle.lock (see cycle-last-full-cycle-at.test.ts).
+ */
+import { describe, test, expect, beforeAll, afterAll, beforeEach, afterEach } from 'bun:test';
+import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from 'fs';
+import { tmpdir } from 'os';
+import { join } from 'path';
+import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import { resetPgliteState } from './helpers/reset-pglite.ts';
+import { withEnv } from './helpers/with-env.ts';
+import { runCycle } from '../src/core/cycle.ts';
+
+let engine: PGLiteEngine;
+let brainDir: string;
+let gbrainHome: string;
+
+beforeAll(async () => {
+  engine = new PGLiteEngine();
+  await engine.connect({});
+  await engine.initSchema();
+}, 30_000);
+
+afterAll(async () => {
+  await engine.disconnect();
+}, 30_000);
+
+beforeEach(async () => {
+  await resetPgliteState(engine);
+  brainDir = mkdtempSync(join(tmpdir(), 'gbrain-cycle-extract-src-'));
+  gbrainHome = mkdtempSync(join(tmpdir(), 'gbrain-cycle-extract-src-home-'));
+  mkdirSync(join(brainDir, 'people'), { recursive: true });
+  writeFileSync(
+    join(brainDir, 'people', 'alice.md'),
+    '# Alice\n\nMet [[people/bob]] today.\n\n## Timeline\n\n- **2026-01-05** | meeting — Discussed the wiki\n',
+  );
+  writeFileSync(join(brainDir, 'people', 'bob.md'), '# Bob\n\nFriend of [[people/alice]].\n');
+
+  // Federated shape: the brain checkout is a registered non-default source
+  // and its pages live ONLY under that source.
+  await engine.executeRaw(
+    `INSERT INTO sources (id, name, local_path) VALUES ('wiki', 'wiki', $1)
+     ON CONFLICT (id) DO UPDATE SET local_path = EXCLUDED.local_path`,
+    [brainDir],
+  );
+  await engine.executeRaw(
+    `INSERT INTO pages (slug, source_id, type, title, compiled_truth, timeline)
+     VALUES
+       ('people/alice', 'wiki', 'person', 'Alice', '', ''),
+       ('people/bob', 'wiki', 'person', 'Bob', '', '')`,
+  );
+});
+
+afterEach(() => {
+  rmSync(brainDir, { recursive: true, force: true });
+  rmSync(gbrainHome, { recursive: true, force: true });
+});
+
+describe('cycle extract phase on a federated brain (#1503)', () => {
+  test('extract phase resolves the source from brainDir and writes links + timeline there', async () => {
+    await withEnv({ GBRAIN_HOME: gbrainHome }, async () => {
+      const report = await runCycle(engine, {
+        brainDir,
+        phases: ['extract'],
+      });
+      const extractPhase = report.phases.find(p => p.phase === 'extract');
+      expect(extractPhase?.status).toBe('ok');
+      // The #1503 symptom was exactly linksCreated: 0 on federated brains.
+      expect(Number(extractPhase?.details?.linksCreated ?? 0)).toBeGreaterThanOrEqual(2);
+      expect(Number(extractPhase?.details?.timelineCreated ?? 0)).toBeGreaterThanOrEqual(1);
+    });
+
+    const links = await engine.executeRaw<{ from_src: string; to_src: string }>(
+      `SELECT pf.source_id AS from_src, pt.source_id AS to_src
+       FROM links l
+       JOIN pages pf ON pf.id = l.from_page_id
+       JOIN pages pt ON pt.id = l.to_page_id`,
+    );
+    expect(links.length).toBeGreaterThanOrEqual(2);
+    for (const l of links) {
+      expect(l.from_src).toBe('wiki');
+      expect(l.to_src).toBe('wiki');
+    }
+
+    const tl = await engine.executeRaw<{ n: string }>(
+      `SELECT COUNT(*)::text AS n FROM timeline_entries t
+       JOIN pages p ON p.id = t.page_id AND p.source_id = 'wiki'`,
+    );
+    expect(Number(tl[0]?.n ?? 0)).toBeGreaterThanOrEqual(1);
+  });
+
+  test('explicit opts.sourceId wins (checkout-less --source shape)', async () => {
+    await withEnv({ GBRAIN_HOME: gbrainHome }, async () => {
+      const report = await runCycle(engine, {
+        brainDir,
+        sourceId: 'wiki',
+        phases: ['extract'],
+      });
+      const extractPhase = report.phases.find(p => p.phase === 'extract');
+      expect(extractPhase?.status).toBe('ok');
+      expect(Number(extractPhase?.details?.linksCreated ?? 0)).toBeGreaterThanOrEqual(2);
+    });
+  });
+});

test/extract-fs-source-id.test.ts

@@ -0,0 +1,149 @@
+/**
+ * #1747 — fs-walk extract on a non-default source.
+ *
+ * `gbrain import --source-id wiki` puts pages under source 'wiki', but the
+ * fs-walk extractors (extractLinksFromDir / extractTimelineFromDir /
+ * extractForSlugs) built batch rows with no source_id. addLinksBatch /
+ * addTimelineEntriesBatch map missing → literal 'default', and their
+ * `JOIN pages ON (slug, source_id)` dropped every row → "Links: created 0
+ * from N pages" with no error. This file pins that the resolved source id
+ * threads through both the CLI fs path (--source-id) and the library
+ * runExtractCore path (incremental slugs — the cycle's route, #1503).
+ *
+ * Hermetic via PGLite in-memory. Canonical shared-engine block per
+ * scripts/check-test-isolation.sh.
+ */
+
+import { describe, test, expect, beforeAll, afterAll, beforeEach, afterEach } from 'bun:test';
+import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from 'fs';
+import { tmpdir } from 'os';
+import { join } from 'path';
+import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import { resetPgliteState } from './helpers/reset-pglite.ts';
+import { runExtract, runExtractCore } from '../src/commands/extract.ts';
+
+let engine: PGLiteEngine;
+let brainDir: string;
+
+beforeAll(async () => {
+  engine = new PGLiteEngine();
+  await engine.connect({});
+  await engine.initSchema();
+}, 30_000);
+
+afterAll(async () => {
+  await engine.disconnect();
+}, 30_000);
+
+beforeEach(async () => {
+  await resetPgliteState(engine);
+  brainDir = mkdtempSync(join(tmpdir(), 'gbrain-extract-fs-src-'));
+  mkdirSync(join(brainDir, 'people'), { recursive: true });
+  // alice links to bob and carries a timeline bullet; bob links back.
+  writeFileSync(
+    join(brainDir, 'people', 'alice.md'),
+    '# Alice\n\nMet [[people/bob]] today.\n\n## Timeline\n\n- **2026-01-05** | meeting — Discussed the wiki\n',
+  );
+  writeFileSync(join(brainDir, 'people', 'bob.md'), '# Bob\n\nFriend of [[people/alice]].\n');
+
+  // Pages live ONLY in source 'wiki' (the `import --source-id wiki` shape).
+  await engine.executeRaw(
+    `INSERT INTO sources (id, name, local_path) VALUES ('wiki', 'wiki', $1)
+     ON CONFLICT (id) DO UPDATE SET local_path = EXCLUDED.local_path`,
+    [brainDir],
+  );
+  await engine.executeRaw(
+    `INSERT INTO pages (slug, source_id, type, title, compiled_truth, timeline)
+     VALUES
+       ('people/alice', 'wiki', 'person', 'Alice', '', ''),
+       ('people/bob', 'wiki', 'person', 'Bob', '', '')`,
+  );
+});
+
+async function linkSourceIds(): Promise<Array<{ from_src: string; to_src: string }>> {
+  return engine.executeRaw<{ from_src: string; to_src: string }>(
+    `SELECT pf.source_id AS from_src, pt.source_id AS to_src
+     FROM links l
+     JOIN pages pf ON pf.id = l.from_page_id
+     JOIN pages pt ON pt.id = l.to_page_id`,
+  );
+}
+
+async function timelineCount(): Promise<number> {
+  const rows = await engine.executeRaw<{ n: string }>(
+    `SELECT COUNT(*)::text AS n FROM timeline_entries t JOIN pages p ON p.id = t.page_id AND p.source_id = 'wiki'`,
+  );
+  return Number(rows[0]?.n ?? 0);
+}
+
+describe('fs-walk extract on a non-default source (#1747)', () => {
+  test('CLI `extract all --source-id wiki` creates edges + timeline in the wiki source', async () => {
+    const origLog = console.log;
+    console.log = () => {};
+    try {
+      await runExtract(engine, ['all', '--dir', brainDir, '--source-id', 'wiki', '--json']);
+    } finally {
+      console.log = origLog;
+    }
+
+    const links = await linkSourceIds();
+    expect(links.length).toBeGreaterThanOrEqual(2); // alice→bob, bob→alice
+    for (const l of links) {
+      expect(l.from_src).toBe('wiki');
+      expect(l.to_src).toBe('wiki');
+    }
+    expect(await timelineCount()).toBeGreaterThanOrEqual(1);
+  });
+
+  test('CLI fs path auto-resolves the source from the registered dir (no --source-id)', async () => {
+    // brainDir is the registered local_path of 'wiki' (and 'wiki' is the sole
+    // non-default source) — the resolver chain must land on it without a flag.
+    const origLog = console.log;
+    console.log = () => {};
+    try {
+      await runExtract(engine, ['links', '--dir', brainDir, '--json']);
+    } finally {
+      console.log = origLog;
+    }
+
+    const links = await linkSourceIds();
+    expect(links.length).toBeGreaterThanOrEqual(2);
+    for (const l of links) expect(l.from_src).toBe('wiki');
+  });
+
+  test('runExtractCore incremental slugs path (the cycle route, #1503) stamps sourceId', async () => {
+    const result = await runExtractCore(engine, {
+      mode: 'all',
+      dir: brainDir,
+      slugs: ['people/alice', 'people/bob'],
+      jsonMode: true,
+      sourceId: 'wiki',
+    });
+
+    expect(result.links_created).toBeGreaterThanOrEqual(2);
+    expect(result.timeline_entries_created).toBeGreaterThanOrEqual(1);
+    const links = await linkSourceIds();
+    for (const l of links) {
+      expect(l.from_src).toBe('wiki');
+      expect(l.to_src).toBe('wiki');
+    }
+    expect(await timelineCount()).toBeGreaterThanOrEqual(1);
+  });
+
+  test('regression shape: without a sourceId the batch JOIN drops every row (created 0)', async () => {
+    // Pre-#1747 behavior, kept as the negative control: unstamped rows map to
+    // 'default' where these pages don't exist, so nothing is inserted.
+    const result = await runExtractCore(engine, {
+      mode: 'all',
+      dir: brainDir,
+      slugs: ['people/alice', 'people/bob'],
+      jsonMode: true,
+    });
+    expect(result.links_created).toBe(0);
+    expect(result.timeline_entries_created).toBe(0);
+  });
+});
+
+afterEach(() => {
+  rmSync(brainDir, { recursive: true, force: true });
+});

test/ingestion/ingest-capture.test.ts

@@ -162,6 +162,70 @@ describe('ingest_capture handler — validation + routing', () => {
   });
 });
 
+describe('ingest_capture handler — provenance write-through (#1522)', () => {
+  async function pageRow(slug: string): Promise<{ source_id: string; source_kind: string | null; source_uri: string | null; ingested_via: string | null } | undefined> {
+    const rows = await engine.executeRaw<{ source_id: string; source_kind: string | null; source_uri: string | null; ingested_via: string | null }>(
+      `SELECT source_id, source_kind, source_uri, ingested_via FROM pages WHERE slug = $1`,
+      [slug],
+    );
+    return rows[0];
+  }
+
+  test('trusted event with a registered source id routes the page write there and persists source_kind/source_uri', async () => {
+    await engine.executeRaw(
+      `INSERT INTO sources (id, name) VALUES ('m365-example', 'm365-example') ON CONFLICT (id) DO NOTHING`,
+    );
+    const handler = makeIngestCaptureHandler(engine);
+    const ev = makeEvent({
+      content: '# calendar event',
+      source_id: 'm365-example',
+      source_kind: 'm365-calendar',
+      source_uri: 'm365:event/abc-123',
+    });
+    const result = await handler(makeJob({ event: ev, slug: 'calendar/evt-1' }));
+    expect(result.status).toBe('imported');
+
+    const row = await pageRow('calendar/evt-1');
+    expect(row?.source_id).toBe('m365-example');
+    expect(row?.source_kind).toBe('m365-calendar');
+    expect(row?.source_uri).toBe('m365:event/abc-123');
+    expect(row?.ingested_via).toBe('ingest_capture');
+  });
+
+  test('unregistered emitter source_id (webhook-<clientId>) keeps default-source routing but still persists provenance', async () => {
+    const handler = makeIngestCaptureHandler(engine);
+    // makeEvent's source_id 'webhook-test' is NOT a registered source.
+    const ev = makeEvent({ content: '# webhook capture' });
+    const result = await handler(makeJob({ event: ev, slug: 'inbox/webhook-1' }));
+    expect(result.status).toBe('imported');
+
+    const row = await pageRow('inbox/webhook-1');
+    expect(row?.source_id).toBe('default');
+    expect(row?.source_kind).toBe('webhook');
+    expect(row?.source_uri).toBe('mcp-webhook:client-x:1234');
+    expect(row?.ingested_via).toBe('ingest_capture');
+  });
+
+  test('untrusted event cannot choose its write source even when the id is registered (fail-closed)', async () => {
+    await engine.executeRaw(
+      `INSERT INTO sources (id, name) VALUES ('wiki', 'wiki') ON CONFLICT (id) DO NOTHING`,
+    );
+    const handler = makeIngestCaptureHandler(engine);
+    const ev = makeEvent({
+      content: '# untrusted payload',
+      source_id: 'wiki',
+      untrusted_payload: true,
+    });
+    const result = await handler(makeJob({ event: ev, slug: 'inbox/untrusted-1' }));
+    expect(result.status).toBe('imported');
+
+    const row = await pageRow('inbox/untrusted-1');
+    expect(row?.source_id).toBe('default');
+    // Provenance strings (no scoping power) still persist.
+    expect(row?.source_kind).toBe('webhook');
+  });
+});
+
 describe('ingest_capture handler — integration with importFromContent', () => {
   test('imported event lands as a page in the DB', async () => {
     const handler = makeIngestCaptureHandler(engine);