fix(mcp): source-scope hardening for remote callers (#2881)

Three fixes in the same leak class (a remote caller reading or writing
outside its granted sources), for multi-source / multi-tenant brains:

1. dispatchToolCall now refuses remote calls that arrive without a
   resolved sourceId (missing_source_scope error envelope) instead of
   silently falling back to the shared 'default' source. Every shipped
   transport already passes sourceId explicitly (serve-http from the
   OAuth client row, http-transport from the legacy token grant, stdio
   from GBRAIN_SOURCE); reaching the fallback remotely always meant a
   programmatic caller skipped scope resolution — the bug class behind
   #1924 / #1371. Trusted local callers (remote === false) keep the
   historical fallback. Direct-dispatch tests updated to carry an
   explicit sourceId, matching the real transport contract.

2. log_ingest threads ctx.sourceId (same pattern as get_chunks /
   get_page), so ingest events are attributed to the caller's source
   instead of piling into 'default'. Engines already accept
   entry.source_id (v0.31.2).

3. get_ingest_log is source-scoped for remote callers via the
   linkReadScopeOpts collapse rule (scalar grant → [scalar]; federated
   grant → granted array); it previously returned the whole brain's
   ingest log to any read-scoped remote client, and ingest summaries can
   carry another source's private context. Trusted local callers keep
   the whole-brain view.

Tests: dispatch guard (refuse remote-without-source, keep local
fallback, guard ordering after op lookup) and end-to-end ingest-log
attribution + scoping over the real dispatch path, on PGLite.

src/core/engine.ts

@@ -1904,7 +1904,11 @@ export interface BrainEngine {
 
   // Ingest log
   logIngest(entry: IngestLogInput): Promise<void>;
-  getIngestLog(opts?: { limit?: number }): Promise<IngestLogEntry[]>;
+  /**
+   * `opts.sourceIds` scopes the log to those sources (federated read grant /
+   * remote caller scope). Omitted → whole brain (trusted local callers).
+   */
+  getIngestLog(opts?: { limit?: number; sourceIds?: string[] }): Promise<IngestLogEntry[]>;
 
   // Sync
   /**

src/core/operations.ts

@@ -2704,6 +2704,11 @@ const log_ingest: Operation = {
   handler: async (ctx, p) => {
     if (ctx.dryRun) return { dry_run: true, action: 'log_ingest' };
     await ctx.engine.logIngest({
+      // Thread ctx.sourceId (same pattern as get_chunks/get_page above): on a
+      // multi-source brain the ingest event must be attributed to the caller's
+      // source, not the shared 'default' bucket. Absent sourceId still falls to
+      // the engine's 'default' (single-source brains unchanged).
+      ...(ctx.sourceId ? { source_id: ctx.sourceId } : {}),
       source_type: p.source_type as string,
       source_ref: p.source_ref as string,
       pages_updated: p.pages_updated as string[],
@@ -2720,7 +2725,17 @@ const get_ingest_log: Operation = {
     limit: { type: 'number', description: 'Max entries (default 20)' },
   },
   handler: async (ctx, p) => {
-    return ctx.engine.getIngestLog({ limit: clampSearchLimit(p.limit as number | undefined, 20, 50) });
+    // Source-scope the log for remote callers (scalar grant → single-element
+    // array; federated grant → the granted array — linkReadScopeOpts collapse
+    // rule). Trusted local callers (remote === false) keep the whole-brain
+    // view, matching every other read op's local posture. Ingest summaries
+    // can carry another source's private context, so an unscoped remote read
+    // is a cross-source leak.
+    const scope = ctx.remote !== false ? linkReadScopeOpts(ctx) : {};
+    return ctx.engine.getIngestLog({
+      limit: clampSearchLimit(p.limit as number | undefined, 20, 50),
+      ...(scope.sourceIds ? { sourceIds: scope.sourceIds } : scope.sourceId ? { sourceIds: [scope.sourceId] } : {}),
+    });
   },
   scope: 'read',
 };

src/core/pglite-engine.ts

@@ -5364,11 +5364,16 @@ export class PGLiteEngine implements BrainEngine {
     );
   }
 
-  async getIngestLog(opts?: { limit?: number }): Promise<IngestLogEntry[]> {
+  async getIngestLog(opts?: { limit?: number; sourceIds?: string[] }): Promise<IngestLogEntry[]> {
     const limit = opts?.limit || 50;
+    // Source-scope for remote / federated callers; unscoped only for trusted
+    // local callers (mirrors the postgres engine).
+    const scoped = opts?.sourceIds && opts.sourceIds.length > 0;
     const { rows } = await this.db.query(
-      `SELECT * FROM ingest_log ORDER BY created_at DESC LIMIT $1`,
-      [limit]
+      scoped
+        ? `SELECT * FROM ingest_log WHERE source_id = ANY($2::text[]) ORDER BY created_at DESC LIMIT $1`
+        : `SELECT * FROM ingest_log ORDER BY created_at DESC LIMIT $1`,
+      scoped ? [limit, opts?.sourceIds] : [limit]
     );
     // Belt-and-suspenders source_id fallback for any pre-v50 row that
     // somehow survived without the backfill.

src/core/postgres-engine.ts

@@ -5462,11 +5462,16 @@ export class PostgresEngine implements BrainEngine {
     `;
   }
 
-  async getIngestLog(opts?: { limit?: number }): Promise<IngestLogEntry[]> {
+  async getIngestLog(opts?: { limit?: number; sourceIds?: string[] }): Promise<IngestLogEntry[]> {
     const sql = this.sql;
     const limit = opts?.limit || 50;
+    // Source-scope for remote / federated callers; unscoped only for trusted
+    // local callers (same posture as searchKeyword's sourceIds filter).
+    const scope = opts?.sourceIds && opts.sourceIds.length > 0
+      ? sql`WHERE source_id = ANY(${opts.sourceIds}::text[])`
+      : sql``;
     const rows = await sql`
-      SELECT * FROM ingest_log ORDER BY created_at DESC LIMIT ${limit}
+      SELECT * FROM ingest_log ${scope} ORDER BY created_at DESC LIMIT ${limit}
     `;
     // Belt-and-suspenders source_id fallback for any pre-v50 row.
     return (rows as unknown as IngestLogEntry[]).map(r => ({

src/mcp/dispatch.ts

@@ -247,6 +247,21 @@ export async function dispatchToolCall(
     };
   }
 
+  // Remote callers must arrive with a resolved source scope. Every shipped
+  // transport passes sourceId explicitly (serve-http from the OAuth client
+  // row, http-transport from the legacy token grant, stdio from
+  // GBRAIN_SOURCE); a remote call reaching the 'default' fallback means a
+  // programmatic caller skipped scope resolution, and silently landing in
+  // the shared 'default' source is the cross-source leak class behind
+  // #1924 / #1371. Trusted local callers (remote === false) keep the
+  // historical fallback via buildOperationContext.
+  if ((opts.remote ?? true) && !opts.sourceId) {
+    return {
+      content: [{ type: 'text', text: JSON.stringify({ error: 'missing_source_scope', message: `Remote tool call '${name}' carries no resolved sourceId; refusing the shared 'default' source fallback. Pass an explicit sourceId resolved from the caller's grant.` }, null, 2) }],
+      isError: true,
+    };
+  }
+
   const ctx = buildOperationContext(engine, safeParams, opts);
 
   try {

test/e2e/auth-permissions.test.ts

@@ -80,8 +80,7 @@ d('access_tokens.permissions.takes_holders end-to-end', () => {
 
     // Now dispatch with that allow-list, verify SQL filter applies
     const result = await dispatchToolCall(engine, 'takes_list', { page_slug: 'people/alice-example' }, {
-      remote: true,
-      takesHoldersAllowList: allowList,
+      remote: true, sourceId: 'default',      takesHoldersAllowList: allowList,
     });
     expect(result.isError).toBeFalsy();
     const takes = JSON.parse(result.content[0].text) as Array<{ holder: string }>;
@@ -100,8 +99,7 @@ d('access_tokens.permissions.takes_holders end-to-end', () => {
       [`tok-w-${Date.now()}`, hash, { takes_holders: ['world'] }],
     );
     const result = await dispatchToolCall(engine, 'takes_search', { query: 'founder' }, {
-      remote: true,
-      takesHoldersAllowList: ['world'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world'],
     });
     const hits = JSON.parse(result.content[0].text) as Array<{ holder: string }>;
     expect(hits.every(h => h.holder === 'world')).toBe(true);

test/e2e/takes-postgres.test.ts

@@ -213,8 +213,7 @@ d('v0.28 MCP allow-list — Postgres dispatch', () => {
   test('takes_list returns only world holders when allow-list = ["world"]', async () => {
     const engine = getEngine();
     const result = await dispatchToolCall(engine, 'takes_list', { page_slug: 'people/alice-example' }, {
-      remote: true,
-      takesHoldersAllowList: ['world'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world'],
     });
     expect(result.isError).toBeFalsy();
     const takes = JSON.parse(result.content[0].text);
@@ -236,8 +235,7 @@ d('v0.28 MCP allow-list — Postgres dispatch', () => {
   test('takes_search honors allow-list', async () => {
     const engine = getEngine();
     const result = await dispatchToolCall(engine, 'takes_search', { query: 'technical' }, {
-      remote: true,
-      takesHoldersAllowList: ['world'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world'],
     });
     const hits = JSON.parse(result.content[0].text) as Array<{ holder: string }>;
     expect(hits.every(h => h.holder === 'world')).toBe(true);
@@ -246,8 +244,7 @@ d('v0.28 MCP allow-list — Postgres dispatch', () => {
   test('think op rejects save/take from remote callers', async () => {
     const engine = getEngine();
     const result = await dispatchToolCall(engine, 'think', { question: 'q', save: true, take: true }, {
-      remote: true,
-    });
+      remote: true, sourceId: 'default',    });
     const env = JSON.parse(result.content[0].text);
     // Remote with save/take → safe path forces them off, runs gather-only
     expect(env.remote_persisted_blocked).toBe(true);
@@ -384,8 +381,7 @@ d('v0.30.0 MCP dispatch — Postgres', () => {
   test('takes_scorecard via MCP returns correct counts with allow-list', async () => {
     const engine = getEngine();
     const result = await dispatchToolCall(engine, 'takes_scorecard', { holder: 'garry' }, {
-      remote: true,
-      takesHoldersAllowList: ['garry'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['garry'],
     });
     expect(result.isError).toBeFalsy();
     const card = JSON.parse(result.content[0].text);
@@ -399,8 +395,7 @@ d('v0.30.0 MCP dispatch — Postgres', () => {
   test('takes_calibration via MCP returns bucket array with allow-list', async () => {
     const engine = getEngine();
     const result = await dispatchToolCall(engine, 'takes_calibration', { holder: 'garry', bucket_size: 0.1 }, {
-      remote: true,
-      takesHoldersAllowList: ['garry'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['garry'],
     });
     expect(result.isError).toBeFalsy();
     const buckets = JSON.parse(result.content[0].text);
@@ -419,8 +414,7 @@ d('v0.30.0 MCP dispatch — Postgres', () => {
     // 'world' has only fact-kind takes in the seed; bets are garry-only.
     // Scorecard scoped to world should report zero resolved.
     const result = await dispatchToolCall(engine, 'takes_scorecard', {}, {
-      remote: true,
-      takesHoldersAllowList: ['world'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world'],
     });
     const card = JSON.parse(result.content[0].text);
     // No resolved bets exist with holder='world' in our seed.

test/e2e/v0_29-mcp-dispatch-pglite.test.ts

@@ -73,7 +73,7 @@ describe('v0.29 E2E — dispatchToolCall for the three new ops', () => {
     const result = await dispatchToolCall(engine, 'get_recent_salience', {
       days: 7,
       limit: 10,
-    }, { remote: true });
+    }, { remote: true, sourceId: 'default' });
 
     expect(result.isError).toBeFalsy();
     expect(result.content[0].type).toBe('text');
@@ -89,7 +89,7 @@ describe('v0.29 E2E — dispatchToolCall for the three new ops', () => {
     const result = await dispatchToolCall(engine, 'find_anomalies', {
       lookback_days: 30,
       sigma: 1.5, // lower threshold so the small fixture tips the cohort
-    }, { remote: true });
+    }, { remote: true, sourceId: 'default' });
 
     expect(result.isError).toBeFalsy();
     const rows = JSON.parse(result.content[0].text);
@@ -115,7 +115,7 @@ describe('v0.29 E2E — dispatchToolCall for the three new ops', () => {
     // every MCP transport sets, so the reject must fire here.
     const result = await dispatchToolCall(engine, 'get_recent_transcripts', {
       days: 7,
-    }, { remote: true });
+    }, { remote: true, sourceId: 'default' });
 
     expect(result.isError).toBe(true);
     const err = JSON.parse(result.content[0].text);
@@ -142,8 +142,81 @@ describe('v0.29 E2E — dispatchToolCall for the three new ops', () => {
   test('unknown tool returns Unknown tool error envelope (regression guard)', async () => {
     // Generic dispatch shape contract — protects against typos in op
     // names accidentally short-circuiting elsewhere in the dispatcher.
-    const result = await dispatchToolCall(engine, 'get_recent_definitely_not_a_real_op', {}, { remote: true });
+    const result = await dispatchToolCall(engine, 'get_recent_definitely_not_a_real_op', {}, { remote: true, sourceId: 'default' });
     expect(result.isError).toBe(true);
     expect(result.content[0].text).toMatch(/Unknown tool/);
   });
 });
+
+describe('dispatch source-scope guard — remote callers must carry a resolved sourceId', () => {
+  test('remote call without sourceId is refused (missing_source_scope), never lands in default', async () => {
+    const result = await dispatchToolCall(engine, 'get_recent_salience', { days: 7 }, { remote: true });
+    expect(result.isError).toBe(true);
+    const body = JSON.parse(result.content[0].text);
+    expect(body.error).toBe('missing_source_scope');
+  });
+
+  test('remote default (opts.remote omitted) is treated as remote and refused too', async () => {
+    const result = await dispatchToolCall(engine, 'get_recent_salience', { days: 7 }, {});
+    expect(result.isError).toBe(true);
+    const body = JSON.parse(result.content[0].text);
+    expect(body.error).toBe('missing_source_scope');
+  });
+
+  test('trusted local callers (remote === false) keep the historical default fallback', async () => {
+    const result = await dispatchToolCall(engine, 'get_recent_salience', { days: 7 }, { remote: false });
+    expect(result.isError).toBeFalsy();
+  });
+
+  test('the guard runs after op lookup: unknown tool still reports Unknown tool, not scope', async () => {
+    const result = await dispatchToolCall(engine, 'not_a_real_op_scope_guard', {}, { remote: true });
+    expect(result.isError).toBe(true);
+    expect(result.content[0].text).toMatch(/Unknown tool/);
+  });
+});
+
+describe('ingest log source scoping — log_ingest threads ctx.sourceId, get_ingest_log honors the grant', () => {
+  beforeAll(async () => {
+    await dispatchToolCall(engine, 'log_ingest', {
+      source_type: 'chat', source_ref: 'alice-conv-1', pages_updated: ['people/carol'], summary: 'alice private context',
+    }, { remote: true, sourceId: 'u-alice' });
+    await dispatchToolCall(engine, 'log_ingest', {
+      source_type: 'chat', source_ref: 'bob-conv-1', pages_updated: ['people/dave'], summary: 'bob private context',
+    }, { remote: true, sourceId: 'u-bob' });
+  });
+
+  test('log_ingest attributes the entry to the caller source, not default', async () => {
+    const result = await dispatchToolCall(engine, 'get_ingest_log', {}, { remote: false });
+    const rows = JSON.parse(result.content[0].text) as Array<{ source_id: string; source_ref: string }>;
+    expect(rows.find(r => r.source_ref === 'alice-conv-1')?.source_id).toBe('u-alice');
+    expect(rows.find(r => r.source_ref === 'bob-conv-1')?.source_id).toBe('u-bob');
+  });
+
+  test('remote scalar-scoped caller only sees its own source rows', async () => {
+    const result = await dispatchToolCall(engine, 'get_ingest_log', {}, { remote: true, sourceId: 'u-alice' });
+    const rows = JSON.parse(result.content[0].text) as Array<{ source_id: string }>;
+    expect(rows.length).toBeGreaterThan(0);
+    expect(rows.every(r => r.source_id === 'u-alice')).toBe(true);
+  });
+
+  test('federated grant sees exactly the granted sources', async () => {
+    const result = await dispatchToolCall(engine, 'get_ingest_log', {}, {
+      remote: true,
+      sourceId: 'u-alice',
+      auth: { token: 't', clientId: 'c', scopes: ['read'], allowedSources: ['u-alice', 'u-bob'] },
+    });
+    const rows = JSON.parse(result.content[0].text) as Array<{ source_id: string }>;
+    const seen = new Set(rows.map(r => r.source_id));
+    expect(seen.has('u-alice')).toBe(true);
+    expect(seen.has('u-bob')).toBe(true);
+    expect(seen.has('default')).toBe(false);
+  });
+
+  test('trusted local caller keeps the whole-brain view', async () => {
+    const result = await dispatchToolCall(engine, 'get_ingest_log', {}, { remote: false });
+    const rows = JSON.parse(result.content[0].text) as Array<{ source_id: string }>;
+    const seen = new Set(rows.map(r => r.source_id));
+    expect(seen.has('u-alice')).toBe(true);
+    expect(seen.has('u-bob')).toBe(true);
+  });
+});

test/skill-catalog-transports.test.ts

@@ -56,7 +56,7 @@ async function call(
   params: Record<string, unknown>,
   opts: { remote: boolean; auth?: AuthInfo },
 ) {
-  return unpack(await dispatchToolCall(engine, name, params, opts));
+  return unpack(await dispatchToolCall(engine, name, params, { sourceId: 'default', ...opts }));
 }
 
 describe('list_skills over dispatch', () => {

test/takes-fence-read-ops.serial.test.ts

@@ -86,8 +86,7 @@ describe('C4: get_page takes-fence redaction (#728)', () => {
 
   test('MCP caller with narrow allow-list (["world"]) sees fence STRIPPED', async () => {
     const result = await dispatchToolCall(engine, 'get_page', { slug: PAGE_SLUG }, {
-      remote: true,
-      takesHoldersAllowList: ['world'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world'],
     });
     const page = parseResult(result) as { compiled_truth: string };
     expect(page.compiled_truth).not.toContain(TAKES_FENCE_BEGIN);
@@ -105,8 +104,7 @@ describe('C4: get_page takes-fence redaction (#728)', () => {
     // takes_search are the typed surfaces for take inspection. get_page is
     // not an authorized take-reading channel.
     const result = await dispatchToolCall(engine, 'get_page', { slug: PAGE_SLUG }, {
-      remote: true,
-      takesHoldersAllowList: ['world', 'garry', 'brain'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world', 'garry', 'brain'],
     });
     const page = parseResult(result) as { compiled_truth: string };
     expect(page.compiled_truth).not.toContain(TAKES_FENCE_BEGIN);
@@ -129,8 +127,7 @@ describe('C4: get_versions takes-fence redaction (#728)', () => {
     );
 
     const result = await dispatchToolCall(engine, 'get_versions', { slug: PAGE_SLUG }, {
-      remote: true,
-      takesHoldersAllowList: ['world'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world'],
     });
     const versions = parseResult(result) as Array<{ compiled_truth: string }>;
     expect(versions.length).toBeGreaterThan(0);

test/takes-mcp-allowlist.serial.test.ts

@@ -58,8 +58,7 @@ describe('per-token takes-holder allow-list — takes_list', () => {
 
   test('allow-list ["world"] (default-deny token) returns ONLY world holders', async () => {
     const result = await dispatchToolCall(engine, 'takes_list', { page_slug: 'people/alice-example' }, {
-      remote: true,
-      takesHoldersAllowList: ['world'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world'],
     });
     const takes = parseResult(result) as Array<{ holder: string; claim: string }>;
     expect(takes).toHaveLength(1);
@@ -69,8 +68,7 @@ describe('per-token takes-holder allow-list — takes_list', () => {
 
   test('allow-list ["world", "garry"] returns world + garry, hides brain hunches', async () => {
     const result = await dispatchToolCall(engine, 'takes_list', { page_slug: 'people/alice-example' }, {
-      remote: true,
-      takesHoldersAllowList: ['world', 'garry'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world', 'garry'],
     });
     const takes = parseResult(result) as Array<{ holder: string }>;
     const holders = takes.map(t => t.holder).sort();
@@ -79,8 +77,7 @@ describe('per-token takes-holder allow-list — takes_list', () => {
 
   test('allow-list with no overlap returns empty (no fallback to default)', async () => {
     const result = await dispatchToolCall(engine, 'takes_list', { page_slug: 'people/alice-example' }, {
-      remote: true,
-      takesHoldersAllowList: ['nonexistent-holder'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['nonexistent-holder'],
     });
     const takes = parseResult(result) as unknown[];
     expect(takes).toHaveLength(0);
@@ -90,8 +87,7 @@ describe('per-token takes-holder allow-list — takes_list', () => {
 describe('per-token takes-holder allow-list — takes_search', () => {
   test('allow-list ["world"] filters search hits to public claims only', async () => {
     const result = await dispatchToolCall(engine, 'takes_search', { query: 'founder' }, {
-      remote: true,
-      takesHoldersAllowList: ['world'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world'],
     });
     const hits = parseResult(result) as Array<{ holder: string; claim: string }>;
     expect(hits.every(h => h.holder === 'world')).toBe(true);
@@ -135,8 +131,7 @@ describe('per-token takes-holder allow-list — get_page body channel', () => {
 
   test('remote token with allow-list strips fence from compiled_truth', async () => {
     const result = await dispatchToolCall(engine, 'get_page', { slug: SLUG }, {
-      remote: true,
-      takesHoldersAllowList: ['world'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world'],
     });
     const page = parseResult(result) as { compiled_truth: string };
     expect(page.compiled_truth).not.toContain(TAKES_FENCE_BEGIN);
@@ -160,8 +155,7 @@ describe('per-token takes-holder allow-list — get_page body channel', () => {
 
   test('fuzzy resolution path also strips for remote token', async () => {
     const result = await dispatchToolCall(engine, 'get_page', { slug: 'people/bob-example', fuzzy: true }, {
-      remote: true,
-      takesHoldersAllowList: ['world', 'garry'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world', 'garry'],
     });
     const page = parseResult(result) as { compiled_truth: string };
     // Allow-list does not yet re-render filtered rows; whole fence is stripped.
@@ -184,8 +178,7 @@ describe('per-token takes-holder allow-list — get_versions body channel', () =
 
   test('remote token with allow-list strips fence from every snapshot', async () => {
     const result = await dispatchToolCall(engine, 'get_versions', { slug: SLUG }, {
-      remote: true,
-      takesHoldersAllowList: ['world'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world'],
     });
     const versions = parseResult(result) as Array<{ compiled_truth: string }>;
     expect(versions.length).toBeGreaterThan(0);
@@ -210,8 +203,7 @@ describe('think op — read-only on remote callers (Lane D landed)', () => {
     // configured machine fires a real LLM call and the warning flips to
     // LLM_OUTPUT_NOT_JSON. runThink then returns gather-only + NO_ANTHROPIC_API_KEY.
     const result = await withoutAnthropicKey(() => dispatchToolCall(engine, 'think', { question: 'q', save: true, take: true }, {
-      remote: true,
-      takesHoldersAllowList: ['world', 'garry', 'brain'],
+      remote: true, sourceId: 'default',      takesHoldersAllowList: ['world', 'garry', 'brain'],
     }));
     const env = parseResult(result) as {
       remote_persisted_blocked: boolean;