diff --git a/src/core/schema-pack/index.ts b/src/core/schema-pack/index.ts index bdc3d1085..e4fd7b462 100644 --- a/src/core/schema-pack/index.ts +++ b/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, diff --git a/src/core/schema-pack/stats.ts b/src/core/schema-pack/stats.ts new file mode 100644 index 000000000..6fe42634a --- /dev/null +++ b/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 }>(); + 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(); + 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 { + 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(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 }> } }, + opts: StatsOpts, +): Promise { + 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 { + 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, + }; +} diff --git a/src/core/schema-pack/sync.ts b/src/core/schema-pack/sync.ts new file mode 100644 index 000000000..068d8dad5 --- /dev/null +++ b/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 { + 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 { + 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, + }; +} diff --git a/test/schema-pack-stats.test.ts b/test/schema-pack-stats.test.ts new file mode 100644 index 000000000..67e18e10a --- /dev/null +++ b/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 { + 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 { + // 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(); + }); + }); +}); diff --git a/test/schema-pack-sync.test.ts b/test/schema-pack-sync.test.ts new file mode 100644 index 000000000..8f1fe62b9 --- /dev/null +++ b/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 { + 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 { + 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 { + 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); + }); + }); +});