diff --git a/src/core/engine.ts b/src/core/engine.ts index f06a9c95a..d013c8a95 100644 --- a/src/core/engine.ts +++ b/src/core/engine.ts @@ -42,6 +42,20 @@ import type { * truncation can compare `result.length` against expected fanout bounds * as a coarse-but-honest signal in the interim. */ +/** + * v0.38: bare row shape returned by `BrainEngine.listAllSources()`. + * Kept lean (no per-source page_count) so the autopilot tick stays O(1) + * SQL queries regardless of source count. `sources-ops.SourceListEntry` + * is the enriched application-layer shape. + */ +export interface SourceRow { + id: string; + name: string | null; + local_path: string | null; + last_sync_at: Date | null; + config: Record; +} + export interface TraverseGraphOpts { sourceId?: string; sourceIds?: string[]; @@ -632,6 +646,44 @@ export interface BrainEngine { */ listAllPageRefs(): Promise>; + /** + * v0.38 — lean per-source enumeration for hot-loop callers (autopilot + * dispatch, doctor freshness check). Returns the bare row shape sources-ops + * needs without the N+1 per-source page_count enrichment in + * `sources-ops.listSources`. + * + * Defaults filter out archived sources. When `localPathOnly` is true, + * also filters `local_path IS NOT NULL` so the autopilot fan-out doesn't + * dispatch jobs for pure-DB sources whose handler would fall back to + * the global sync.repo_path (codex r1 P1-4). + * + * `config` is returned as `Record` — both engines + * already parse the JSONB at the boundary (Postgres-js returns + * parsed objects; PGLite returns objects via its built-in JSONB + * codec). Callers reading `config['last_full_cycle_at']` get a string. + */ + listAllSources(opts?: { + includeArchived?: boolean; + localPathOnly?: boolean; + }): Promise; + + /** + * v0.38 — atomic JSONB merge into sources.config. Uses Postgres's + * `config || $patch::jsonb` operator so concurrent writers don't + * stomp each other (last write wins, but no read-modify-write race). + * + * Primary caller: runCycle's exit hook writes + * { last_full_cycle_at: '' } + * after a successful per-source cycle so autopilot's freshness gate + * can read it next tick. Resolves codex round-1 P0-5 (write site for + * last_full_cycle_at was unspecified pre-PR). + * + * Returns true if a row was updated (source exists), false otherwise + * (silently no-ops on unknown sourceId — caller decides whether that's + * a problem). + */ + updateSourceConfig(sourceId: string, patch: Record): Promise; + /** * v0.37.0 — prefix-stratified page sampling for `gbrain brainstorm` / `gbrain lsd` * domain-bank module. Takes a caller-supplied prefix list (cached at the domain-bank diff --git a/src/core/pglite-engine.ts b/src/core/pglite-engine.ts index 50791d38c..72d13cdfb 100644 --- a/src/core/pglite-engine.ts +++ b/src/core/pglite-engine.ts @@ -13,6 +13,7 @@ import type { TakesScorecard, TakesScorecardOpts, CalibrationBucket, CalibrationCurveOpts, FactRow, FactKind, FactVisibility, FactInsertStatus, NewFact, FactListOpts, FactsHealth, + SourceRow, } from './engine.ts'; import { MAX_SEARCH_LIMIT, clampSearchLimit } from './engine.ts'; import { runMigrations } from './migrate.ts'; @@ -877,6 +878,55 @@ export class PGLiteEngine implements BrainEngine { return (rows as { slug: string; source_id: string }[]).map(r => ({ slug: r.slug, source_id: r.source_id })); } + async listAllSources(opts?: { + includeArchived?: boolean; + localPathOnly?: boolean; + }): Promise { + // v0.38: parity with postgres-engine.listAllSources. Defaults match + // sources-ops.listSources (archived rows filtered out by default). + // localPathOnly skips pure-DB sources so autopilot fan-out doesn't + // dispatch jobs that would fall back to the global sync.repo_path. + const includeArchived = opts?.includeArchived === true; + const localPathOnly = opts?.localPathOnly === true; + const { rows } = await this.db.query<{ + id: string; + name: string | null; + local_path: string | null; + last_sync_at: string | null; + config: unknown; + }>( + `SELECT id, name, local_path, last_sync_at, config + FROM sources + WHERE ($1::boolean OR archived IS NOT TRUE) + AND ($2::boolean OR local_path IS NOT NULL) + ORDER BY (id = 'default') DESC, id`, + [includeArchived, !localPathOnly], + ); + return rows.map((r) => ({ + id: r.id, + name: r.name, + local_path: r.local_path, + last_sync_at: r.last_sync_at ? new Date(r.last_sync_at) : null, + config: typeof r.config === 'string' + ? JSON.parse(r.config) as Record + : ((r.config as Record | null) ?? {}), + })); + } + + async updateSourceConfig(sourceId: string, patch: Record): Promise { + // v0.38: parity with postgres-engine.updateSourceConfig. JSONB `||` + // concat operator (overrides same-key, no deep merge). PGLite passes + // `JSON.stringify(patch)` as the param; cast to jsonb on the SQL side. + const result = await this.db.query<{ id: string }>( + `UPDATE sources + SET config = COALESCE(config, '{}'::jsonb) || $1::jsonb + WHERE id = $2 + RETURNING id`, + [JSON.stringify(patch), sourceId], + ); + return result.rows.length > 0; + } + // v0.37.0 — domain-bank engine methods (D14 + D5 + D10). // See postgres-engine.ts:listPrefixSampledPages for the ranking + source-scope rationale. // PGLite runs the same SQL (Postgres 17.5 under the hood) with positional `$N` binding. diff --git a/src/core/postgres-engine.ts b/src/core/postgres-engine.ts index 7210bf00b..4b7ef18b3 100644 --- a/src/core/postgres-engine.ts +++ b/src/core/postgres-engine.ts @@ -10,6 +10,7 @@ import type { TakesScorecard, TakesScorecardOpts, CalibrationBucket, CalibrationCurveOpts, FactRow, FactKind, FactVisibility, FactInsertStatus, NewFact, FactListOpts, FactsHealth, + SourceRow, } from './engine.ts'; import type { DomainBankSampleOpts, CorpusSampleOpts, DomainBankRow, @@ -932,6 +933,53 @@ export class PostgresEngine implements BrainEngine { return rows.map((r) => ({ slug: r.slug as string, source_id: r.source_id as string })); } + async listAllSources(opts?: { + includeArchived?: boolean; + localPathOnly?: boolean; + }): Promise { + // v0.38: lean per-source enumeration for autopilot dispatch + doctor. + // Filters at SQL so the autopilot tick stays one query regardless of + // how many archived rows exist. ORDER BY (id='default') DESC, id + // matches sources-ops.listSources for operator-output stability. + const sql = this.sql; + const includeArchived = opts?.includeArchived === true; + const localPathOnly = opts?.localPathOnly === true; + const rows = await sql` + SELECT id, name, local_path, last_sync_at, config + FROM sources + WHERE (${includeArchived} OR archived IS NOT TRUE) + AND (${!localPathOnly} OR local_path IS NOT NULL) + ORDER BY (id = 'default') DESC, id + `; + return rows.map((r) => ({ + id: r.id as string, + name: (r.name as string | null) ?? null, + local_path: (r.local_path as string | null) ?? null, + last_sync_at: r.last_sync_at ? new Date(r.last_sync_at as string) : null, + config: typeof r.config === 'string' ? JSON.parse(r.config) : ((r.config as Record | null) ?? {}), + })); + } + + async updateSourceConfig(sourceId: string, patch: Record): Promise { + // v0.38: atomic JSONB merge. `||` is the Postgres concat operator — + // for jsonb, right-side keys overwrite left-side; nested object keys + // are NOT deep-merged (use jsonb_set for nested paths). The patch + // shape this autopilot wave uses is flat (`last_full_cycle_at`, + // `archive_*`, etc.) so concat is sufficient. Idempotent on re-run. + // + // sql.json(patch) is the canonical safe path per feedback_postgres_jsonb_double_encode + // — postgres-js handles JSONB serialization (no double-encode). Matches + // the pattern at putPage and submitJob elsewhere in this file. + const sql = this.sql; + const result = await sql` + UPDATE sources + SET config = COALESCE(config, '{}'::jsonb) || ${sql.json(patch as Parameters[0])} + WHERE id = ${sourceId} + `; + // postgres-js returns count as result.count; matched-rows shape + return (result.count ?? 0) > 0; + } + // v0.37.0 — domain-bank engine methods (D14 + D5 + D10). // // `listPrefixSampledPages`: one page per prefix, tiebroken by inbound-link diff --git a/test/list-all-sources.test.ts b/test/list-all-sources.test.ts new file mode 100644 index 000000000..4d0f9ddb5 --- /dev/null +++ b/test/list-all-sources.test.ts @@ -0,0 +1,138 @@ +/** + * v0.38 engine.listAllSources + updateSourceConfig integration tests (PGLite). + * + * Runs against an in-memory PGLite engine so the test is hermetic (no + * DATABASE_URL required). Postgres parity is covered by the e2e suite. + */ +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); +}); + +afterAll(async () => { + await engine.disconnect(); +}); + +beforeEach(async () => { + await resetPgliteState(engine); +}); + +async function seedSource( + id: string, + opts: { local_path?: string | null; archived?: boolean; config?: Record } = {}, +): Promise { + const localPath = opts.local_path === undefined ? `/tmp/${id}` : opts.local_path; + const archived = opts.archived === true; + const config = JSON.stringify(opts.config ?? {}); + // ON CONFLICT to make the seed idempotent in case the test bed re-seeds 'default'. + await engine.executeRaw( + `INSERT INTO sources (id, name, local_path, config, archived, created_at) + VALUES ($1, $2, $3, $4::jsonb, $5, NOW()) + ON CONFLICT (id) DO UPDATE + SET local_path = EXCLUDED.local_path, + config = EXCLUDED.config, + archived = EXCLUDED.archived`, + [id, id, localPath, config, archived], + ); +} + +describe('engine.listAllSources', () => { + test('returns empty array on fresh brain with only seeded default', async () => { + // 'default' source seeded by migration; we'll just check it appears + const all = await engine.listAllSources(); + expect(all.length).toBeGreaterThanOrEqual(1); + expect(all.some(s => s.id === 'default')).toBe(true); + }); + + test('filters archived by default', async () => { + await seedSource('alive'); + await seedSource('dead', { archived: true }); + const all = await engine.listAllSources(); + expect(all.some(s => s.id === 'alive')).toBe(true); + expect(all.some(s => s.id === 'dead')).toBe(false); + }); + + test('includeArchived: true returns archived rows too', async () => { + await seedSource('alive'); + await seedSource('dead', { archived: true }); + const all = await engine.listAllSources({ includeArchived: true }); + expect(all.some(s => s.id === 'alive')).toBe(true); + expect(all.some(s => s.id === 'dead')).toBe(true); + }); + + test('localPathOnly: true filters local_path IS NULL (codex P1-4)', async () => { + await seedSource('with-path', { local_path: '/tmp/x' }); + await seedSource('db-only', { local_path: null }); + const all = await engine.listAllSources({ localPathOnly: true }); + expect(all.some(s => s.id === 'with-path')).toBe(true); + expect(all.some(s => s.id === 'db-only')).toBe(false); + }); + + test('config JSONB parses to object (autopilot reads last_full_cycle_at)', async () => { + await seedSource('fred', { config: { last_full_cycle_at: '2026-05-22T07:00:00.000Z', remote_url: 'https://x' } }); + const all = await engine.listAllSources(); + const fred = all.find(s => s.id === 'fred')!; + expect(fred.config.last_full_cycle_at).toBe('2026-05-22T07:00:00.000Z'); + expect(fred.config.remote_url).toBe('https://x'); + }); + + test('default source sorts first', async () => { + await seedSource('zebra'); + await seedSource('alpha'); + const all = await engine.listAllSources(); + expect(all[0].id).toBe('default'); + }); +}); + +describe('engine.updateSourceConfig', () => { + test('returns false for unknown source', async () => { + const updated = await engine.updateSourceConfig('does-not-exist', { last_full_cycle_at: 'x' }); + expect(updated).toBe(false); + }); + + test('returns true and merges patch into config JSONB', async () => { + await seedSource('alpha', { config: { existing: 'keep-me', remote_url: 'https://x' } }); + const updated = await engine.updateSourceConfig('alpha', { last_full_cycle_at: '2026-05-22T08:00:00.000Z' }); + expect(updated).toBe(true); + const all = await engine.listAllSources(); + const a = all.find(s => s.id === 'alpha')!; + expect(a.config.existing).toBe('keep-me'); + expect(a.config.remote_url).toBe('https://x'); + expect(a.config.last_full_cycle_at).toBe('2026-05-22T08:00:00.000Z'); + }); + + test('patch overwrites same-key value (last-write-wins per JSONB ||)', async () => { + await seedSource('beta', { config: { last_full_cycle_at: '2026-01-01T00:00:00.000Z' } }); + await engine.updateSourceConfig('beta', { last_full_cycle_at: '2026-05-22T09:00:00.000Z' }); + const all = await engine.listAllSources(); + expect(all.find(s => s.id === 'beta')!.config.last_full_cycle_at).toBe('2026-05-22T09:00:00.000Z'); + }); + + test('idempotent: repeat write of same patch is a no-op semantically', async () => { + await seedSource('charlie'); + await engine.updateSourceConfig('charlie', { last_full_cycle_at: '2026-05-22T10:00:00.000Z' }); + await engine.updateSourceConfig('charlie', { last_full_cycle_at: '2026-05-22T10:00:00.000Z' }); + const all = await engine.listAllSources(); + expect(all.find(s => s.id === 'charlie')!.config.last_full_cycle_at).toBe('2026-05-22T10:00:00.000Z'); + }); + + test('COALESCE defense: works on source with empty config (the v0.38 fresh shape)', async () => { + // Schema enforces config JSONB NOT NULL DEFAULT '{}', so we cannot + // produce a row with NULL config here. The COALESCE in + // updateSourceConfig is defensive against pre-migration brains whose + // sources table predates the NOT NULL constraint. This case just + // confirms the happy-path on the default empty config. + await seedSource('delta', { config: {} }); + const updated = await engine.updateSourceConfig('delta', { last_full_cycle_at: '2026-05-22T11:00:00.000Z' }); + expect(updated).toBe(true); + const all = await engine.listAllSources(); + expect(all.find(s => s.id === 'delta')!.config.last_full_cycle_at).toBe('2026-05-22T11:00:00.000Z'); + }); +});