feat(engine): listAllSources + updateSourceConfig for per-source autopilot

Two lean engine-layer methods that the v0.38 per-source autopilot wave
consumes. Both have parity implementations on Postgres + PGLite.

listAllSources(opts?):
- Returns the bare SourceRow shape (id, name, local_path, last_sync_at,
  config) without sources-ops.listSources's per-source page_count
  enrichment (N+1 expensive; out of scope for hot-loop callers).
- `includeArchived` defaults false (matches sources-ops semantics).
- `localPathOnly` filters local_path IS NOT NULL so autopilot fan-out
  doesn't dispatch jobs for pure-DB sources whose handler would fall
  back to global sync.repo_path (codex r1 P1-4).
- Ordering: (id = 'default') DESC, id — same as sources-ops for
  operator-output stability.

updateSourceConfig(sourceId, patch):
- Atomic JSONB merge via Postgres `config || $patch::jsonb` operator.
  No read-modify-write race; same-key overwrites (no deep merge —
  flat patches only, matches the v0.38 use case of last_full_cycle_at).
- Returns true when a row was updated, false when sourceId doesn't
  exist (best-effort no-op; caller decides how to handle).
- Postgres: sql.json(patch) per the canonical pattern; PGLite:
  JSON.stringify + ::jsonb cast on positional param.

New SourceRow type exported from engine.ts. Imported by both engine
impls. 11 integration tests in test/list-all-sources.test.ts cover
defaults, filters, JSONB round-trip, archived flag, and merge semantics.

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

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<string, unknown>;
+}
+
 export interface TraverseGraphOpts {
   sourceId?: string;
   sourceIds?: string[];
@@ -632,6 +646,44 @@ export interface BrainEngine {
    */
   listAllPageRefs(): Promise<Array<{ slug: string; source_id: string }>>;
 
+  /**
+   * 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<string, unknown>` — 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<SourceRow[]>;
+
+  /**
+   * 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: '<ISO>' }
+   * 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<string, unknown>): Promise<boolean>;
+
   /**
    * 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

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<SourceRow[]> {
+    // 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<string, unknown>
+        : ((r.config as Record<string, unknown> | null) ?? {}),
+    }));
+  }
+
+  async updateSourceConfig(sourceId: string, patch: Record<string, unknown>): Promise<boolean> {
+    // 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.

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<SourceRow[]> {
+    // 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<string, unknown> | null) ?? {}),
+    }));
+  }
+
+  async updateSourceConfig(sourceId: string, patch: Record<string, unknown>): Promise<boolean> {
+    // 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<typeof sql.json>[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

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<string, unknown> } = {},
+): Promise<void> {
+  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');
+  });
+});