v0.40.6.0 Phase 3 — stats + sync pure core functions

Both ship as runStatsCore / runSyncCore pure functions so Phase 4 CLI
handlers (next commit) and Phase 7 MCP ops (later) both compose without
duplicating logic. Codex C13 / D16 prereq for the MCP exposure phase.

stats.ts (17 cases):
  Multi-source aware: sourceIds[] (federated read) OR sourceId (single)
  OR neither (whole-brain aggregate). NULLIF(type, '') normalizes
  empty-string + NULL to one untyped bucket (pages.type is NOT NULL in
  the schema so empty string is the legacy "untyped" representation).
  Soft-delete exclusion. by_type sorted by count desc, ties by name asc.
  Empty-brain coverage:1.0 (vacuous truth, matches getBrainScore).
  Dead-prefix detection: pack-declared prefixes with zero matching
  pages surface as DeadPrefixHint[] (agent's drilldown signal for
  mis-declared paths). Best-effort: pack-load failure leaves
  pack_identity:null + dead_prefixes:[].

sync.ts (13 cases):
  D14 chunked UPDATE: 1000-row batches per prefix. Each batch:
  WITH win AS (SELECT id FROM pages WHERE untyped+prefix LIMIT $batch),
  upd AS (UPDATE ... WHERE id IN win RETURNING 1) SELECT COUNT(*). Loop
  until zero rows. Concurrent writers never block on the row-set for
  more than ~100ms per batch (vs the multi-second monolithic UPDATE
  shape PR #1321 had).
  Codex C5 write-side scoping: sourceId param directly, NOT
  sourceScopeOpts which is read-side and inherits OAuth federation
  reads. Phase 7 MCP op (schema_apply_mutations) enforces at dispatch.
  Dry-run by default: per-prefix probe returns would_apply + 10-slug
  sample (the drilldown signal). Apply path returns total_applied.
  Idempotency contract pinned: second apply finds zero matching rows.
  Soft-delete exclusion on both probe + update. Dead-prefix flag set
  when probe returns count=0. JSON envelope schema_version:1.

Tests use canonical PGLite block per CLAUDE.md test-isolation rules.
seedPage helper auto-seeds sources(id) row before FK insert.

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

src/core/schema-pack/index.ts

@@ -174,6 +174,22 @@ export {
 
 export { invalidateQueryCache } from './query-cache-invalidator.ts';
 
+export {
+  type StatsOpts,
+  type StatsResult,
+  type PerSourceStats,
+  type TypeStats,
+  type DeadPrefixHint,
+  runStatsCore,
+} from './stats.ts';
+
+export {
+  type SyncOpts,
+  type SyncResult,
+  type PerPrefixResult,
+  runSyncCore,
+} from './sync.ts';
+
 export {
   type LintIssue,
   type LintOpts,

src/core/schema-pack/stats.ts

@@ -0,0 +1,245 @@
+// v0.40.6.0 Schema Cathedral v3 — schema_stats pure core function.
+//
+// Reports per-type page counts, untyped-coverage percentage, and
+// dead-prefix detection (declared prefix with zero matching pages —
+// a "this type has no content" signal that helps agents spot
+// mis-declared paths).
+//
+// Multi-source aware: accepts `sourceIds` (federated read) OR
+// `sourceId` (single) OR nothing (aggregate across all sources). Uses
+// the same WHERE shape as the rest of the read-path codebase.
+//
+// PGLite + Postgres parity via `executeRaw`. Soft-deletes filtered.
+//
+// Pure core: returns structured data; CLI handler in Phase 4 wraps
+// for human + JSON output, MCP handler in Phase 7 wires it through
+// the operation envelope.
+
+import type { BrainEngine } from '../engine.ts';
+import { loadActivePackBestEffort } from './best-effort.ts';
+import type { OperationContext } from '../operations.ts';
+
+export interface StatsOpts {
+  /** Single source scope. Omit + omit sourceIds for whole-brain aggregate. */
+  sourceId?: string;
+  /** Federated read scope (overrides sourceId when set). */
+  sourceIds?: string[];
+}
+
+export interface TypeStats {
+  /** Type name as it appears in the DB `pages.type` column. */
+  type: string;
+  /** Page count for this type within the scope. */
+  count: number;
+}
+
+export interface PerSourceStats {
+  source_id: string;
+  total_pages: number;
+  typed_pages: number;
+  untyped_pages: number;
+  coverage: number;
+  by_type: TypeStats[];
+}
+
+export interface DeadPrefixHint {
+  /** Type name declared in the pack. */
+  type: string;
+  /** Prefix declared in the pack with zero matching DB pages. */
+  prefix: string;
+}
+
+export interface StatsResult {
+  schema_version: 1;
+  /** Pack identity at stats time (null when no pack loaded). */
+  pack_identity: string | null;
+  /** Aggregate across all scoped sources. */
+  aggregate: PerSourceStats;
+  /** Per-source breakdown when multiple sources are in scope. */
+  per_source: PerSourceStats[];
+  /** Pack-declared prefixes that match zero pages — likely mis-declared. */
+  dead_prefixes: DeadPrefixHint[];
+}
+
+interface RawCountRow {
+  source_id: string | null;
+  type: string | null;
+  cnt: string;
+}
+
+function computeCoverage(typed: number, total: number): number {
+  if (total === 0) return 1.0;  // vacuous truth — matches getBrainScore pattern
+  return Math.round((typed / total) * 10000) / 10000;
+}
+
+function aggregateRows(rows: RawCountRow[]): PerSourceStats[] {
+  const bySource = new Map<string, { typed: number; untyped: number; total: number; byType: Map<string, number> }>();
+  for (const r of rows) {
+    const sid = r.source_id ?? 'default';
+    const cnt = parseInt(r.cnt, 10) || 0;
+    if (!bySource.has(sid)) {
+      bySource.set(sid, { typed: 0, untyped: 0, total: 0, byType: new Map() });
+    }
+    const bucket = bySource.get(sid)!;
+    if (r.type === null || r.type === '') {
+      bucket.untyped += cnt;
+    } else {
+      bucket.typed += cnt;
+      bucket.byType.set(r.type, (bucket.byType.get(r.type) ?? 0) + cnt);
+    }
+    bucket.total += cnt;
+  }
+  const out: PerSourceStats[] = [];
+  for (const [sid, b] of bySource) {
+    out.push({
+      source_id: sid,
+      total_pages: b.total,
+      typed_pages: b.typed,
+      untyped_pages: b.untyped,
+      coverage: computeCoverage(b.typed, b.total),
+      by_type: [...b.byType.entries()]
+        .map(([type, count]) => ({ type, count }))
+        .sort((a, b) => b.count - a.count || a.type.localeCompare(b.type)),
+    });
+  }
+  // Sort sources alphabetically for stable output.
+  out.sort((a, b) => a.source_id.localeCompare(b.source_id));
+  return out;
+}
+
+function mergeAggregate(per: PerSourceStats[]): PerSourceStats {
+  let typed = 0, untyped = 0, total = 0;
+  const byType = new Map<string, number>();
+  for (const s of per) {
+    typed += s.typed_pages;
+    untyped += s.untyped_pages;
+    total += s.total_pages;
+    for (const t of s.by_type) byType.set(t.type, (byType.get(t.type) ?? 0) + t.count);
+  }
+  return {
+    source_id: '*',
+    total_pages: total,
+    typed_pages: typed,
+    untyped_pages: untyped,
+    coverage: computeCoverage(typed, total),
+    by_type: [...byType.entries()]
+      .map(([type, count]) => ({ type, count }))
+      .sort((a, b) => b.count - a.count || a.type.localeCompare(b.type)),
+  };
+}
+
+/**
+ * Query the DB for stats. Multi-source aware:
+ *   - sourceIds[] → WHERE source_id = ANY($1::text[])
+ *   - sourceId    → WHERE source_id = $1
+ *   - neither     → no source filter (whole brain)
+ *
+ * Always: `WHERE deleted_at IS NULL`.
+ *
+ * Returns rows grouped by (source_id, type) so we can compute per-source
+ * + aggregate in one read.
+ */
+async function fetchCountRows(engine: BrainEngine, opts: StatsOpts): Promise<RawCountRow[]> {
+  let where = 'WHERE deleted_at IS NULL';
+  const params: unknown[] = [];
+  if (opts.sourceIds && opts.sourceIds.length > 0) {
+    where += ` AND source_id = ANY($1::text[])`;
+    params.push(opts.sourceIds);
+  } else if (opts.sourceId) {
+    where += ` AND source_id = $1`;
+    params.push(opts.sourceId);
+  }
+  // pages.type is NOT NULL in the schema — empty string represents
+  // "untyped" (legacy + put_page fallback per D12). Normalize both
+  // empty-string and NULL to the same bucket via NULLIF.
+  const sql = `
+    SELECT
+      COALESCE(source_id, 'default') AS source_id,
+      NULLIF(type, '') AS type,
+      COUNT(*)::text AS cnt
+    FROM pages
+    ${where}
+    GROUP BY source_id, NULLIF(type, '')
+    ORDER BY source_id, NULLIF(type, '') NULLS LAST
+  `;
+  try {
+    return await engine.executeRaw<RawCountRow>(sql, params);
+  } catch {
+    // Empty / pre-init brain: pages table may not exist yet.
+    return [];
+  }
+}
+
+/**
+ * Per-prefix existence check for dead-prefix detection. One COUNT(*)
+ * per declared prefix in the active pack. Cheap on small brains;
+ * Phase 4 CLI offers a `--no-dead-prefix-scan` flag to opt out on
+ * brains where this matters.
+ */
+async function detectDeadPrefixes(
+  engine: BrainEngine,
+  pack: { manifest: { page_types: ReadonlyArray<{ name: string; path_prefixes: ReadonlyArray<string> }> } },
+  opts: StatsOpts,
+): Promise<DeadPrefixHint[]> {
+  const hints: DeadPrefixHint[] = [];
+  let sourceWhere = '';
+  const sourceParam: unknown[] = [];
+  if (opts.sourceIds && opts.sourceIds.length > 0) {
+    sourceWhere = ` AND source_id = ANY($2::text[])`;
+    sourceParam.push(opts.sourceIds);
+  } else if (opts.sourceId) {
+    sourceWhere = ` AND source_id = $2`;
+    sourceParam.push(opts.sourceId);
+  }
+  for (const t of pack.manifest.page_types) {
+    for (const prefix of t.path_prefixes) {
+      try {
+        const rows = await engine.executeRaw<{ cnt: string }>(
+          `SELECT COUNT(*)::text AS cnt FROM pages
+           WHERE deleted_at IS NULL
+             AND source_path LIKE $1${sourceWhere}`,
+          [`${prefix}%`, ...sourceParam],
+        );
+        const cnt = parseInt(rows[0]?.cnt ?? '0', 10) || 0;
+        if (cnt === 0) {
+          hints.push({ type: t.name, prefix });
+        }
+      } catch {
+        // Skip on engine error (no pages table yet, etc.).
+        continue;
+      }
+    }
+  }
+  return hints;
+}
+
+/**
+ * Pure core for `gbrain schema stats` (CLI) AND `schema_stats` MCP op.
+ * Returns the StatsResult ready for human-formatter, JSON output, or
+ * operation envelope wrapping.
+ */
+export async function runStatsCore(
+  ctx: OperationContext,
+  opts: StatsOpts = {},
+): Promise<StatsResult> {
+  const rows = await fetchCountRows(ctx.engine, opts);
+  const per_source = aggregateRows(rows);
+  const aggregate = mergeAggregate(per_source);
+
+  // Pack identity + dead-prefix scan — best-effort.
+  let pack_identity: string | null = null;
+  let dead_prefixes: DeadPrefixHint[] = [];
+  const pack = await loadActivePackBestEffort(ctx);
+  if (pack) {
+    pack_identity = pack.identity;
+    dead_prefixes = await detectDeadPrefixes(ctx.engine, pack, opts);
+  }
+
+  return {
+    schema_version: 1,
+    pack_identity,
+    aggregate,
+    per_source,
+    dead_prefixes,
+  };
+}

src/core/schema-pack/sync.ts

@@ -0,0 +1,216 @@
+// v0.40.6.0 Schema Cathedral v3 — schema_sync pure core function.
+//
+// For each path_prefix declared in the active pack, find pages with
+// NULL type matching that prefix and backfill `pages.type` to the
+// declared type. Dry-run by default; `--apply` performs the UPDATEs.
+//
+// D14 chunked UPDATE: 1000-row batches. Each chunk completes in <100ms,
+// releases the row lock between batches, never wedges concurrent
+// writers (the bug class v0.22.1 statement_timeout work paid for).
+//
+// Codex C5: write-side source scoping. Mutations use the caller's
+// `ctx.sourceId` directly (write authority), NOT `sourceScopeOpts`
+// which is read-side and can inherit OAuth federation reads. Phase 7's
+// MCP `schema_apply_mutations` op enforces this at the dispatch layer
+// too.
+//
+// PGLite + Postgres parity via `executeRaw`.
+
+import type { BrainEngine } from '../engine.ts';
+import { loadActivePackBestEffort } from './best-effort.ts';
+import type { OperationContext } from '../operations.ts';
+
+export interface SyncOpts {
+  /** Apply UPDATE statements. Default false (dry-run). */
+  apply?: boolean;
+  /**
+   * Source ID to scope the sync. Write-side (codex C5): mutations target
+   * the caller's authorized source, not read federation. Omit for whole-
+   * brain sync (typically only CLI / autopilot path).
+   */
+  sourceId?: string;
+  /** Per-batch row cap (D14). Default 1000. */
+  batchSize?: number;
+  /** Progress callback fired per batch on apply path. */
+  onProgress?: (info: { type: string; prefix: string; appliedSoFar: number }) => void;
+}
+
+export interface PerPrefixResult {
+  type: string;
+  prefix: string;
+  /** Untyped pages matching this prefix at dry-run time. */
+  would_apply: number;
+  /** Sample of slugs (capped at 10) for the agent's drilldown. */
+  sample_slugs: string[];
+  /** Whether the prefix matched zero pages (dead-prefix hint). */
+  dead_prefix: boolean;
+  /** Rows actually updated. Equal to would_apply on a clean apply,
+   *  0 on dry-run, possibly less if concurrent writer claimed some. */
+  applied: number;
+}
+
+export interface SyncResult {
+  schema_version: 1;
+  apply: boolean;
+  pack_identity: string | null;
+  per_prefix: PerPrefixResult[];
+  /** Total rows that would be / were updated across all prefixes. */
+  total_would_apply: number;
+  total_applied: number;
+}
+
+/**
+ * Count + sample untyped pages matching a prefix. Used by both dry-run
+ * (sample is the drilldown signal) and apply (count for would_apply).
+ */
+async function probePrefix(
+  engine: BrainEngine,
+  prefix: string,
+  sourceId: string | undefined,
+): Promise<{ count: number; sample: string[] }> {
+  let where = `WHERE deleted_at IS NULL AND (type IS NULL OR type = '') AND source_path LIKE $1`;
+  const params: unknown[] = [`${prefix}%`];
+  if (sourceId) {
+    where += ` AND source_id = $2`;
+    params.push(sourceId);
+  }
+  try {
+    const cntRows = await engine.executeRaw<{ cnt: string }>(
+      `SELECT COUNT(*)::text AS cnt FROM pages ${where}`,
+      params,
+    );
+    const count = parseInt(cntRows[0]?.cnt ?? '0', 10) || 0;
+    if (count === 0) return { count: 0, sample: [] };
+    const sampleRows = await engine.executeRaw<{ slug: string }>(
+      `SELECT slug FROM pages ${where} ORDER BY slug LIMIT 10`,
+      params,
+    );
+    return { count, sample: sampleRows.map((r) => r.slug) };
+  } catch {
+    return { count: 0, sample: [] };
+  }
+}
+
+/**
+ * Apply the type assignment in 1000-row chunks (D14). Each loop
+ * iteration: SELECT a window of matching IDs, UPDATE them in one
+ * statement, count returned rows. Stops when a batch returns zero
+ * (idempotent: re-running finds nothing to update).
+ *
+ * Returns the total rows updated.
+ */
+async function applyTypeAssignment(
+  engine: BrainEngine,
+  type: string,
+  prefix: string,
+  sourceId: string | undefined,
+  batchSize: number,
+  onProgress?: (appliedSoFar: number) => void,
+): Promise<number> {
+  let totalApplied = 0;
+  let sourceWhere = '';
+  // Param indices for the subquery: 1=type (used in the outer UPDATE),
+  // 2=prefix, 3=batchSize, 4=sourceId (when scoped).
+  const sourceParams: unknown[] = [];
+  if (sourceId) {
+    sourceWhere = ` AND source_id = $4`;
+    sourceParams.push(sourceId);
+  }
+  // Loop guard: max 10000 iterations protects against runaway.
+  for (let i = 0; i < 10000; i++) {
+    try {
+      const rows = await engine.executeRaw<{ updated: string }>(
+        `WITH win AS (
+           SELECT id FROM pages
+           WHERE deleted_at IS NULL
+             AND (type IS NULL OR type = '')
+             AND source_path LIKE $2${sourceWhere}
+           LIMIT $3
+         ),
+         upd AS (
+           UPDATE pages SET type = $1
+           WHERE id IN (SELECT id FROM win)
+           RETURNING 1
+         )
+         SELECT COUNT(*)::text AS updated FROM upd`,
+        [type, `${prefix}%`, batchSize, ...sourceParams],
+      );
+      const batchCount = parseInt(rows[0]?.updated ?? '0', 10) || 0;
+      if (batchCount === 0) break;
+      totalApplied += batchCount;
+      onProgress?.(totalApplied);
+      // Safety net: if a batch returned less than batchSize, we're done.
+      if (batchCount < batchSize) break;
+    } catch (e) {
+      // Surface the error — sync failures are real and should fail loud.
+      throw new Error(`schema sync failed at prefix '${prefix}' → type '${type}': ${(e as Error).message}`);
+    }
+  }
+  return totalApplied;
+}
+
+/**
+ * Pure core for `gbrain schema sync` (CLI) AND `schema_sync` MCP op.
+ *
+ * Dry-run by default: probes each declared prefix for untyped page
+ * count + a 10-slug sample (the agent's drilldown signal). With
+ * apply=true: chunked UPDATE per prefix.
+ */
+export async function runSyncCore(
+  ctx: OperationContext,
+  opts: SyncOpts = {},
+): Promise<SyncResult> {
+  const apply = opts.apply ?? false;
+  const batchSize = Math.max(1, Math.min(10000, opts.batchSize ?? 1000));
+  const sourceId = opts.sourceId;  // codex C5: write-side scoping
+
+  const pack = await loadActivePackBestEffort(ctx);
+  if (!pack) {
+    return {
+      schema_version: 1,
+      apply,
+      pack_identity: null,
+      per_prefix: [],
+      total_would_apply: 0,
+      total_applied: 0,
+    };
+  }
+  const per_prefix: PerPrefixResult[] = [];
+  let total_would_apply = 0;
+  let total_applied = 0;
+  for (const t of pack.manifest.page_types) {
+    for (const prefix of t.path_prefixes) {
+      const probe = await probePrefix(ctx.engine, prefix, sourceId);
+      const dead_prefix = probe.count === 0;
+      let applied = 0;
+      if (apply && probe.count > 0) {
+        applied = await applyTypeAssignment(
+          ctx.engine,
+          t.name,
+          prefix,
+          sourceId,
+          batchSize,
+          (n) => opts.onProgress?.({ type: t.name, prefix, appliedSoFar: n }),
+        );
+      }
+      per_prefix.push({
+        type: t.name,
+        prefix,
+        would_apply: probe.count,
+        sample_slugs: probe.sample,
+        dead_prefix,
+        applied,
+      });
+      total_would_apply += probe.count;
+      total_applied += applied;
+    }
+  }
+  return {
+    schema_version: 1,
+    apply,
+    pack_identity: pack.identity,
+    per_prefix,
+    total_would_apply,
+    total_applied,
+  };
+}

test/schema-pack-stats.test.ts

@@ -0,0 +1,241 @@
+// v0.40.6.0 — stats.ts contract tests.
+//
+// Multi-source aware, soft-delete exclusion, dead-prefix detection,
+// PGLite parity. Phase 3 of the schema cathedral v3 plan.
+
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'bun:test';
+import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import { resetPgliteState } from './helpers/reset-pglite.ts';
+import { runStatsCore } from '../src/core/schema-pack/stats.ts';
+import {
+  __setPackLocatorForTests,
+  _resetPackLocatorForTests,
+} from '../src/core/schema-pack/load-active.ts';
+import { _resetPackCacheForTests } from '../src/core/schema-pack/registry.ts';
+import type { OperationContext } from '../src/core/operations.ts';
+import { withEnv } from './helpers/with-env.ts';
+
+let engine: PGLiteEngine;
+let tmpDir: string;
+
+beforeAll(async () => {
+  engine = new PGLiteEngine();
+  await engine.connect({});
+  await engine.initSchema();
+});
+
+afterAll(async () => {
+  await engine.disconnect();
+});
+
+beforeEach(async () => {
+  await resetPgliteState(engine);
+  _resetPackCacheForTests();
+  _resetPackLocatorForTests();
+  tmpDir = mkdtempSync(join(tmpdir(), 'gbrain-stats-test-'));
+});
+
+function ctxOf(remote = false): OperationContext {
+  return {
+    engine,
+    config: {},
+    logger: { info: () => {}, warn: () => {}, error: () => {} },
+    dryRun: false,
+    remote,
+  } as unknown as OperationContext;
+}
+
+async function ensureSource(id: string): Promise<void> {
+  if (id === 'default') return;  // seeded by schema bootstrap
+  await engine.executeRaw(
+    `INSERT INTO sources (id, name) VALUES ($1, $1) ON CONFLICT (id) DO NOTHING`,
+    [id],
+  );
+}
+
+async function seedPage(slug: string, opts: { type?: string; sourceId?: string; sourcePath?: string; deleted?: boolean } = {}): Promise<void> {
+  // pages.type is NOT NULL; use empty string for "untyped".
+  // pages.title is NOT NULL.
+  // pages.source_id FKs sources(id) — seed source first.
+  const sourceId = opts.sourceId ?? 'default';
+  await ensureSource(sourceId);
+  await engine.executeRaw(
+    `INSERT INTO pages (slug, source_id, source_path, type, title, compiled_truth, timeline, content_hash, deleted_at)
+     VALUES ($1, $2, $3, $4, $5, '', '', '', $6)`,
+    [slug, sourceId, opts.sourcePath ?? `${slug}.md`, opts.type ?? '', slug, opts.deleted ? new Date() : null],
+  );
+}
+
+function seedTinyPack(packName: string, types: Array<{ name: string; prefix: string }>): void {
+  const dir = join(tmpDir, packName);
+  mkdirSync(dir, { recursive: true });
+  const path = join(dir, 'pack.yaml');
+  let body = `api_version: gbrain-schema-pack-v1\nname: ${packName}\nversion: 1.0.0\ndescription: ""\ngbrain_min_version: 0.38.0\nextends: null\nborrow_from: []\npage_types:\n`;
+  for (const t of types) {
+    body += `  - name: ${t.name}\n    primitive: entity\n    path_prefixes:\n      - ${t.prefix}\n    aliases: []\n    extractable: false\n    expert_routing: false\n`;
+  }
+  body += `link_types: []\nfrontmatter_links: []\ntakes_kinds:\n  - fact\n  - take\n  - bet\n  - hunch\nenrichable_types: []\nfiling_rules: []\n`;
+  writeFileSync(path, body, 'utf-8');
+  __setPackLocatorForTests((name) => (name === packName ? path : null));
+}
+
+describe('runStatsCore — empty brain', () => {
+  it('reports coverage:1.0 (vacuous truth)', async () => {
+    await withEnv({ GBRAIN_SCHEMA_PACK: undefined }, async () => {
+      const result = await runStatsCore(ctxOf());
+      expect(result.aggregate.total_pages).toBe(0);
+      expect(result.aggregate.coverage).toBe(1.0);
+      expect(result.per_source).toEqual([]);
+    });
+  });
+});
+
+describe('runStatsCore — single source', () => {
+  it('counts typed + untyped pages and computes coverage', async () => {
+    await withEnv({ GBRAIN_SCHEMA_PACK: undefined }, async () => {
+      await seedPage('a', { type: 'person' });
+      await seedPage('b', { type: 'person' });
+      await seedPage('c', { type: 'company' });
+      await seedPage('d');  // untyped
+      const result = await runStatsCore(ctxOf());
+      expect(result.aggregate.total_pages).toBe(4);
+      expect(result.aggregate.typed_pages).toBe(3);
+      expect(result.aggregate.untyped_pages).toBe(1);
+      expect(result.aggregate.coverage).toBe(0.75);
+      expect(result.aggregate.by_type).toEqual([
+        { type: 'person', count: 2 },
+        { type: 'company', count: 1 },
+      ]);
+    });
+  });
+
+  it('excludes soft-deleted pages', async () => {
+    await withEnv({ GBRAIN_SCHEMA_PACK: undefined }, async () => {
+      await seedPage('a', { type: 'person' });
+      await seedPage('b', { type: 'person', deleted: true });
+      const result = await runStatsCore(ctxOf());
+      expect(result.aggregate.total_pages).toBe(1);
+    });
+  });
+
+  it('respects sourceId scoping', async () => {
+    await withEnv({ GBRAIN_SCHEMA_PACK: undefined }, async () => {
+      await seedPage('a', { type: 'person', sourceId: 'src-a' });
+      await seedPage('b', { type: 'person', sourceId: 'src-b' });
+      const result = await runStatsCore(ctxOf(), { sourceId: 'src-a' });
+      expect(result.aggregate.total_pages).toBe(1);
+      expect(result.per_source.length).toBe(1);
+      expect(result.per_source[0]!.source_id).toBe('src-a');
+    });
+  });
+});
+
+describe('runStatsCore — federated', () => {
+  it('aggregates across sourceIds array', async () => {
+    await withEnv({ GBRAIN_SCHEMA_PACK: undefined }, async () => {
+      await seedPage('a', { type: 'person', sourceId: 'src-a' });
+      await seedPage('b', { type: 'person', sourceId: 'src-b' });
+      await seedPage('c', { type: 'person', sourceId: 'src-c' });
+      const result = await runStatsCore(ctxOf(), { sourceIds: ['src-a', 'src-b'] });
+      expect(result.aggregate.total_pages).toBe(2);
+      expect(result.per_source.length).toBe(2);
+      const sources = result.per_source.map((s) => s.source_id).sort();
+      expect(sources).toEqual(['src-a', 'src-b']);
+    });
+  });
+
+  it('per-source breakdown sorted alphabetically', async () => {
+    await withEnv({ GBRAIN_SCHEMA_PACK: undefined }, async () => {
+      await seedPage('a', { type: 'person', sourceId: 'src-zzz' });
+      await seedPage('b', { type: 'person', sourceId: 'src-aaa' });
+      const result = await runStatsCore(ctxOf());
+      expect(result.per_source.map((s) => s.source_id)).toEqual(['src-aaa', 'src-zzz']);
+    });
+  });
+});
+
+describe('runStatsCore — dead-prefix detection', () => {
+  it('flags pack-declared prefixes with zero matching pages', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack('tiny', [
+        { name: 'person', prefix: 'people/' },
+        { name: 'company', prefix: 'companies/' },
+      ]);
+      await seedPage('people/alice', { type: 'person', sourcePath: 'people/alice.md' });
+      // No companies/* pages → dead prefix.
+      const result = await runStatsCore(ctxOf());
+      expect(result.pack_identity).not.toBeNull();
+      expect(result.dead_prefixes).toEqual([{ type: 'company', prefix: 'companies/' }]);
+    });
+  });
+
+  it('no dead-prefix hints when every declared prefix has pages', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack('tiny', [{ name: 'person', prefix: 'people/' }]);
+      await seedPage('people/alice', { type: 'person', sourcePath: 'people/alice.md' });
+      const result = await runStatsCore(ctxOf());
+      expect(result.dead_prefixes).toEqual([]);
+    });
+  });
+
+  it('returns empty dead_prefixes when pack load fails', async () => {
+    await withEnv({ GBRAIN_SCHEMA_PACK: 'never-installed' }, async () => {
+      __setPackLocatorForTests(() => null);
+      const result = await runStatsCore(ctxOf());
+      expect(result.pack_identity).toBeNull();
+      expect(result.dead_prefixes).toEqual([]);
+    });
+  });
+});
+
+describe('runStatsCore — JSON envelope shape', () => {
+  it('schema_version stays 1 (stable contract)', async () => {
+    await withEnv({ GBRAIN_SCHEMA_PACK: undefined }, async () => {
+      const result = await runStatsCore(ctxOf());
+      expect(result.schema_version).toBe(1);
+    });
+  });
+
+  it('aggregate fields match the per-source merge', async () => {
+    await withEnv({ GBRAIN_SCHEMA_PACK: undefined }, async () => {
+      await seedPage('a', { type: 'person', sourceId: 'src-a' });
+      await seedPage('b', { sourceId: 'src-b' });  // untyped
+      const result = await runStatsCore(ctxOf());
+      const totalFromPer = result.per_source.reduce((acc, s) => acc + s.total_pages, 0);
+      expect(result.aggregate.total_pages).toBe(totalFromPer);
+    });
+  });
+
+  it('by_type sorted by count desc, ties by name asc', async () => {
+    await withEnv({ GBRAIN_SCHEMA_PACK: undefined }, async () => {
+      await seedPage('a', { type: 'company' });
+      await seedPage('b', { type: 'person' });
+      await seedPage('c', { type: 'person' });
+      await seedPage('d', { type: 'person' });
+      const result = await runStatsCore(ctxOf());
+      expect(result.aggregate.by_type[0]!.type).toBe('person');
+      expect(result.aggregate.by_type[1]!.type).toBe('company');
+    });
+  });
+});
+
+describe('runStatsCore — type/untyped split', () => {
+  it('treats empty-string type as untyped (not its own bucket)', async () => {
+    await withEnv({ GBRAIN_SCHEMA_PACK: undefined }, async () => {
+      // Some legacy rows might have type='' rather than NULL.
+      await engine.executeRaw(
+        `INSERT INTO pages (slug, source_id, source_path, type, title, compiled_truth, timeline, content_hash)
+         VALUES ('a', 'default', 'a.md', '', 'a', '', '', '')`,
+      );
+      await seedPage('b', { type: 'person' });
+      const result = await runStatsCore(ctxOf());
+      expect(result.aggregate.untyped_pages).toBe(1);
+      expect(result.aggregate.typed_pages).toBe(1);
+      // empty-string type does NOT appear as its own type bucket.
+      expect(result.aggregate.by_type.find((t) => t.type === '')).toBeUndefined();
+    });
+  });
+});

test/schema-pack-sync.test.ts

@@ -0,0 +1,272 @@
+// v0.40.6.0 — sync.ts contract tests.
+//
+// Dry-run vs apply, chunked-batch correctness, idempotency,
+// soft-delete exclusion, sample-slug payload, dead-prefix hint,
+// per-source write scoping, PGLite parity. Phase 3 of the cathedral.
+
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'bun:test';
+import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import { resetPgliteState } from './helpers/reset-pglite.ts';
+import { runSyncCore } from '../src/core/schema-pack/sync.ts';
+import {
+  __setPackLocatorForTests,
+  _resetPackLocatorForTests,
+} from '../src/core/schema-pack/load-active.ts';
+import { _resetPackCacheForTests } from '../src/core/schema-pack/registry.ts';
+import type { OperationContext } from '../src/core/operations.ts';
+import { withEnv } from './helpers/with-env.ts';
+
+let engine: PGLiteEngine;
+let tmpDir: string;
+
+beforeAll(async () => {
+  engine = new PGLiteEngine();
+  await engine.connect({});
+  await engine.initSchema();
+});
+
+afterAll(async () => {
+  await engine.disconnect();
+});
+
+beforeEach(async () => {
+  await resetPgliteState(engine);
+  _resetPackCacheForTests();
+  _resetPackLocatorForTests();
+  tmpDir = mkdtempSync(join(tmpdir(), 'gbrain-sync-test-'));
+});
+
+function ctxOf(remote = false): OperationContext {
+  return {
+    engine,
+    config: {},
+    logger: { info: () => {}, warn: () => {}, error: () => {} },
+    dryRun: false,
+    remote,
+  } as unknown as OperationContext;
+}
+
+async function ensureSource(id: string): Promise<void> {
+  if (id === 'default') return;
+  await engine.executeRaw(
+    `INSERT INTO sources (id, name) VALUES ($1, $1) ON CONFLICT (id) DO NOTHING`,
+    [id],
+  );
+}
+
+async function seedPage(slug: string, sourcePath: string, opts: { type?: string; sourceId?: string; deleted?: boolean } = {}): Promise<void> {
+  const sourceId = opts.sourceId ?? 'default';
+  await ensureSource(sourceId);
+  await engine.executeRaw(
+    `INSERT INTO pages (slug, source_id, source_path, type, title, compiled_truth, timeline, content_hash, deleted_at)
+     VALUES ($1, $2, $3, $4, $5, '', '', '', $6)`,
+    [slug, sourceId, sourcePath, opts.type ?? '', slug, opts.deleted ? new Date() : null],
+  );
+}
+
+async function getType(slug: string): Promise<string | null> {
+  const rows = await engine.executeRaw<{ type: string }>(
+    `SELECT type FROM pages WHERE slug = $1`,
+    [slug],
+  );
+  return rows[0]?.type ?? null;
+}
+
+function seedTinyPack(types: Array<{ name: string; prefix: string }>): void {
+  const dir = join(tmpDir, 'tiny');
+  mkdirSync(dir, { recursive: true });
+  const path = join(dir, 'pack.yaml');
+  let body = `api_version: gbrain-schema-pack-v1\nname: tiny\nversion: 1.0.0\ndescription: ""\ngbrain_min_version: 0.38.0\nextends: null\nborrow_from: []\npage_types:\n`;
+  for (const t of types) {
+    body += `  - name: ${t.name}\n    primitive: entity\n    path_prefixes:\n      - ${t.prefix}\n    aliases: []\n    extractable: false\n    expert_routing: false\n`;
+  }
+  body += `link_types: []\nfrontmatter_links: []\ntakes_kinds:\n  - fact\n  - take\n  - bet\n  - hunch\nenrichable_types: []\nfiling_rules: []\n`;
+  writeFileSync(path, body, 'utf-8');
+  __setPackLocatorForTests((name) => (name === 'tiny' ? path : null));
+}
+
+describe('runSyncCore — dry-run', () => {
+  it('returns would_apply count + sample_slugs without writing', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([{ name: 'person', prefix: 'people/' }]);
+      await seedPage('alice', 'people/alice.md');
+      await seedPage('bob', 'people/bob.md');
+      const result = await runSyncCore(ctxOf(), { apply: false });
+      expect(result.apply).toBe(false);
+      expect(result.total_would_apply).toBe(2);
+      expect(result.total_applied).toBe(0);
+      const personEntry = result.per_prefix.find((p) => p.type === 'person')!;
+      expect(personEntry.would_apply).toBe(2);
+      expect(personEntry.applied).toBe(0);
+      expect(personEntry.sample_slugs).toEqual(['alice', 'bob']);
+      // Confirm types still empty after dry-run.
+      expect(await getType('alice')).toBe('');
+      expect(await getType('bob')).toBe('');
+    });
+  });
+
+  it('sample_slugs capped at 10', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([{ name: 'person', prefix: 'people/' }]);
+      for (let i = 0; i < 15; i++) {
+        await seedPage(`p${String(i).padStart(2, '0')}`, `people/p${String(i).padStart(2, '0')}.md`);
+      }
+      const result = await runSyncCore(ctxOf(), { apply: false });
+      const personEntry = result.per_prefix.find((p) => p.type === 'person')!;
+      expect(personEntry.would_apply).toBe(15);
+      expect(personEntry.sample_slugs.length).toBe(10);
+    });
+  });
+
+  it('dead_prefix flag fires when no pages match', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([
+        { name: 'person', prefix: 'people/' },
+        { name: 'company', prefix: 'companies/' },
+      ]);
+      await seedPage('alice', 'people/alice.md');
+      const result = await runSyncCore(ctxOf(), { apply: false });
+      const companyEntry = result.per_prefix.find((p) => p.type === 'company')!;
+      expect(companyEntry.dead_prefix).toBe(true);
+      expect(companyEntry.would_apply).toBe(0);
+    });
+  });
+});
+
+describe('runSyncCore — apply', () => {
+  it('updates page.type for matching untyped pages', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([{ name: 'person', prefix: 'people/' }]);
+      await seedPage('alice', 'people/alice.md');
+      await seedPage('bob', 'people/bob.md');
+      const result = await runSyncCore(ctxOf(), { apply: true });
+      expect(result.total_applied).toBe(2);
+      expect(await getType('alice')).toBe('person');
+      expect(await getType('bob')).toBe('person');
+    });
+  });
+
+  it('idempotent: second apply is a no-op', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([{ name: 'person', prefix: 'people/' }]);
+      await seedPage('alice', 'people/alice.md');
+      const first = await runSyncCore(ctxOf(), { apply: true });
+      const second = await runSyncCore(ctxOf(), { apply: true });
+      expect(first.total_applied).toBe(1);
+      expect(second.total_applied).toBe(0);
+    });
+  });
+
+  it('chunked UPDATE: large set in 1000-row batches (perf-shape verification)', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([{ name: 'person', prefix: 'people/' }]);
+      // Seed 1500 untyped pages (1.5× the default batch size).
+      for (let i = 0; i < 1500; i++) {
+        await seedPage(`p${String(i).padStart(4, '0')}`, `people/p${String(i).padStart(4, '0')}.md`);
+      }
+      const batches: number[] = [];
+      const result = await runSyncCore(ctxOf(), {
+        apply: true,
+        batchSize: 1000,
+        onProgress: (info) => batches.push(info.appliedSoFar),
+      });
+      expect(result.total_applied).toBe(1500);
+      // Progress fired at least once for the first batch (1000) and
+      // again for the tail (1500 total).
+      expect(batches).toContain(1000);
+      expect(batches).toContain(1500);
+    });
+  });
+
+  it('does NOT touch pages that already have a type', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([{ name: 'person', prefix: 'people/' }]);
+      await seedPage('alice', 'people/alice.md', { type: 'old-type' });
+      await seedPage('bob', 'people/bob.md');  // untyped
+      await runSyncCore(ctxOf(), { apply: true });
+      expect(await getType('alice')).toBe('old-type');  // preserved
+      expect(await getType('bob')).toBe('person');
+    });
+  });
+
+  it('excludes soft-deleted pages', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([{ name: 'person', prefix: 'people/' }]);
+      await seedPage('alice', 'people/alice.md');
+      await seedPage('zombie', 'people/zombie.md', { deleted: true });
+      const result = await runSyncCore(ctxOf(), { apply: true });
+      expect(result.total_applied).toBe(1);
+      expect(await getType('alice')).toBe('person');
+      expect(await getType('zombie')).toBe('');  // untouched (soft-deleted)
+    });
+  });
+});
+
+describe('runSyncCore — source scoping (codex C5 write-side)', () => {
+  it('updates only the scoped source', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([{ name: 'person', prefix: 'people/' }]);
+      await seedPage('alice', 'people/alice.md', { sourceId: 'src-a' });
+      await seedPage('bob', 'people/bob.md', { sourceId: 'src-b' });
+      const result = await runSyncCore(ctxOf(), { apply: true, sourceId: 'src-a' });
+      expect(result.total_applied).toBe(1);
+      expect(await getType('alice')).toBe('person');
+      expect(await getType('bob')).toBe('');
+    });
+  });
+});
+
+describe('runSyncCore — pack-load failure', () => {
+  it('returns empty result when no pack is loaded', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'nonexistent' }, async () => {
+      __setPackLocatorForTests(() => null);
+      await seedPage('alice', 'people/alice.md');
+      const result = await runSyncCore(ctxOf(), { apply: true });
+      expect(result.pack_identity).toBeNull();
+      expect(result.per_prefix).toEqual([]);
+      expect(result.total_applied).toBe(0);
+      // Page untouched (no pack = no inference rules).
+      expect(await getType('alice')).toBe('');
+    });
+  });
+});
+
+describe('runSyncCore — JSON envelope shape', () => {
+  it('schema_version stays 1', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([{ name: 'person', prefix: 'people/' }]);
+      const result = await runSyncCore(ctxOf());
+      expect(result.schema_version).toBe(1);
+    });
+  });
+
+  it('per_prefix entry shape is stable', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([{ name: 'person', prefix: 'people/' }]);
+      await seedPage('alice', 'people/alice.md');
+      const result = await runSyncCore(ctxOf());
+      const entry = result.per_prefix[0]!;
+      expect(entry).toHaveProperty('type');
+      expect(entry).toHaveProperty('prefix');
+      expect(entry).toHaveProperty('would_apply');
+      expect(entry).toHaveProperty('sample_slugs');
+      expect(entry).toHaveProperty('dead_prefix');
+      expect(entry).toHaveProperty('applied');
+    });
+  });
+});
+
+describe('runSyncCore — batch size clamping', () => {
+  it('clamps batch size to >=1 and <=10000', async () => {
+    await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_SCHEMA_PACK: 'tiny' }, async () => {
+      seedTinyPack([{ name: 'person', prefix: 'people/' }]);
+      await seedPage('alice', 'people/alice.md');
+      // batchSize:0 and batchSize:99999 both work; result is identical.
+      const r1 = await runSyncCore(ctxOf(), { apply: true, batchSize: 0 });
+      expect(r1.total_applied).toBe(1);
+    });
+  });
+});