Merge origin/master; re-bump to v0.42.39.0 (queue collision: 0.42.38.0 claimed by #2015)

test/cycle-abort.test.ts

@@ -152,3 +152,50 @@ describe('autopilot-cycle handler contract (v0.20.5)', () => {
     expect(checkCalls).toBeGreaterThanOrEqual(6);
   });
 });
+
+describe('#1972 — complete cooperative-abort coverage', () => {
+  test('aborted signal makes runCycle report partial + reason "aborted" (terminal guard)', async () => {
+    // phases:[] means no between-phase checkAborted fires, so execution reaches
+    // the TERMINAL guard (Codex #9). With a pre-aborted signal it must NOT
+    // report success — status 'partial', reason 'aborted'.
+    const { runCycle } = await import('../src/core/cycle.ts');
+    const abort = new AbortController();
+    abort.abort(new Error('timeout'));
+    const report = await runCycle(null, {
+      brainDir: '/nonexistent-for-test',
+      phases: [],
+      signal: abort.signal,
+    });
+    expect(report.status).toBe('partial');
+    expect(report.reason).toBe('aborted');
+  });
+
+  test('cycle.ts threads opts.signal into every long phase + guards the success stamp', async () => {
+    const fs = await import('fs');
+    const src = fs.readFileSync(new URL('../src/core/cycle.ts', import.meta.url), 'utf8');
+    const body = src.slice(src.indexOf('export async function runCycle'));
+    // Each long phase receives the signal.
+    expect(body).toContain('runPhaseExtract(engine, brainDir, dryRun, syncPagesAffected, opts.signal)');
+    expect(body).toMatch(/runPhaseExtractFacts\([^)]*opts\.signal\)/);
+    expect(body).toContain('signal: opts.signal'); // consolidate opts
+    expect(body).toContain('runPhaseLint(brainDir, dryRun, engine, opts.signal)');
+    // Reaper runs at cycle start.
+    expect(body).toContain('reapDeadHolderLocks(engine)');
+    // Terminal guard: the success stamp is gated on !aborted, and the report
+    // carries reason 'aborted'.
+    expect(body).toContain('!aborted');
+    expect(body).toContain("reason: 'aborted'");
+    // Phase-duration force-evict attribution log (T11).
+    expect(body).toContain('FORCE_EVICT_DEADLINE_MS');
+  });
+
+  test('every long phase core checks isAborted in its batch loop', async () => {
+    const fs = await import('fs');
+    const read = (p: string) => fs.readFileSync(new URL(p, import.meta.url), 'utf8');
+    expect(read('../src/commands/extract.ts')).toContain('if (isAborted(signal)) return;');
+    expect(read('../src/core/cycle/extract-facts.ts')).toContain('if (isAborted(opts.signal)) break;');
+    expect(read('../src/core/cycle/phases/consolidate.ts')).toContain('if (isAborted(opts.signal)) break;');
+    expect(read('../src/core/cycle/phantom-redirect.ts')).toContain('if (isAborted(signal)) break;');
+    expect(read('../src/commands/lint.ts')).toContain('if (isAborted(opts.signal)) break;');
+  });
+});

test/cycle-consolidate.test.ts

@@ -10,6 +10,7 @@
 
 import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
 import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import { configureGateway } from '../src/core/ai/gateway.ts';
 import { runPhaseConsolidate } from '../src/core/cycle/phases/consolidate.ts';
 
 let engine: PGLiteEngine;
@@ -17,6 +18,20 @@ let engine: PGLiteEngine;
 beforeAll(async () => {
   engine = new PGLiteEngine();
   await engine.connect({});
+  // initSchema() bakes the facts.embedding dim from the gateway's configured
+  // embedding model; the default is now 1280-d (ZE). This file's fixtures are
+  // 1536-d, so pin the legacy 1536-d OpenAI config (matching
+  // test/helpers/legacy-embedding-preload.ts) right before initSchema. The
+  // global preload sets this, but a co-sharded test that calls resetGateway()
+  // in its teardown nulls it, leaving initSchema to fall back to the 1280-d
+  // default and build a halfvec(1280) column the 1536-d inserts can't fill.
+  // Re-pinning here makes the schema deterministic regardless of shard
+  // neighbors (surfaced when #1972's new test files reshuffled the shards).
+  configureGateway({
+    embedding_model: 'openai:text-embedding-3-large',
+    embedding_dimensions: 1536,
+    env: { ...process.env },
+  });
   await engine.initSchema();
 });
 

test/db-lock-heartbeat-takeover.test.ts

@@ -16,6 +16,7 @@
 import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
 import { hostname } from 'os';
 import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import { withEnv } from './helpers/with-env.ts';
 import {
   tryAcquireDbLock,
   resolveStealGraceSeconds,
@@ -81,22 +82,16 @@ describe('resolveStealGraceSeconds', () => {
     expect(resolveStealGraceSeconds(1)).toBe(60);
   });
 
-  test('env override wins', () => {
-    process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS = '123';
-    try {
+  test('env override wins', async () => {
+    await withEnv({ GBRAIN_LOCK_STEAL_GRACE_SECONDS: '123' }, async () => {
       expect(resolveStealGraceSeconds(30)).toBe(123);
-    } finally {
-      delete process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS;
-    }
+    });
   });
 
-  test('bad env override falls back to derived', () => {
-    process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS = 'nope';
-    try {
+  test('bad env override falls back to derived', async () => {
+    await withEnv({ GBRAIN_LOCK_STEAL_GRACE_SECONDS: 'nope' }, async () => {
       expect(resolveStealGraceSeconds(30)).toBe(600);
-    } finally {
-      delete process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS;
-    }
+    });
   });
 });
 

test/db-lock-reap.test.ts

@@ -0,0 +1,162 @@
+/**
+ * #1972 — host-scoped reaper for dead-holder sync/cycle locks.
+ *
+ * Covers:
+ *   - reapDeadHolderLocks: namespace scope (sync/cycle only, NOT election/etc),
+ *     same-host dead-PID reaped regardless of TTL, live/cross-host/within-grace
+ *     kept. Uses the injectable process.kill seam so it's deterministic.
+ *   - deleteLockRowExact: snapshot-matched delete (the TOCTOU defense) — a
+ *     non-matching acquired_at is a no-op, the matching one deletes.
+ */
+
+import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
+import { hostname } from 'os';
+import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import {
+  reapDeadHolderLocks,
+  deleteLockRowExact,
+  inspectLock,
+  HOLDER_TAKEOVER_GRACE_MS,
+  type HolderLivenessOpts,
+} from '../src/core/db-lock.ts';
+
+let engine: PGLiteEngine;
+
+beforeAll(async () => {
+  engine = new PGLiteEngine();
+  await engine.connect({ database_url: '' });
+  await engine.initSchema();
+});
+
+afterAll(async () => {
+  await engine.disconnect();
+});
+
+beforeEach(async () => {
+  await engine.executeRaw(`DELETE FROM gbrain_cycle_locks`);
+});
+
+const LOCAL = hostname();
+const OLD_S = Math.ceil(HOLDER_TAKEOVER_GRACE_MS / 1000) + 60; // comfortably past grace
+const YOUNG_S = 5;
+
+/** Insert a lock row with a given age + namespace. ttlFuture controls whether
+ *  the row is TTL-expired (to prove the reaper reaps dead holders regardless). */
+async function seedLock(
+  id: string,
+  holderPid: number,
+  holderHost: string,
+  ageSeconds: number,
+  ttlFuture = true,
+) {
+  const ttl = ttlFuture ? `NOW() + INTERVAL '10 minutes'` : `NOW() - INTERVAL '1 minute'`;
+  await engine.executeRaw(
+    `INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at)
+     VALUES ($1, $2, $3, NOW() - ($4 || ' seconds')::interval, ${ttl}, NOW() - ($4 || ' seconds')::interval)`,
+    [id, holderPid, holderHost, String(ageSeconds)],
+  );
+}
+
+/** process.kill seam: every pid is dead (ESRCH) except those in `live`. */
+function killSeam(live: Set<number>): HolderLivenessOpts {
+  return {
+    localHost: LOCAL,
+    processKill: (pid: number) => {
+      if (live.has(pid)) return; // alive
+      const e = new Error('no such process') as NodeJS.ErrnoException;
+      e.code = 'ESRCH';
+      throw e;
+    },
+  };
+}
+
+async function lockIds(): Promise<string[]> {
+  const rows = await engine.executeRaw<{ id: string }>(`SELECT id FROM gbrain_cycle_locks ORDER BY id`);
+  return rows.map(r => r.id);
+}
+
+describe('reapDeadHolderLocks', () => {
+  test('reaps same-host dead-PID sync + cycle locks (regardless of TTL)', async () => {
+    await seedLock('gbrain-sync:src-a', 900001, LOCAL, OLD_S, /*ttlFuture*/ true);  // dead, TTL NOT expired
+    await seedLock('gbrain-cycle', 900002, LOCAL, OLD_S, false);                     // dead, TTL expired
+    await seedLock('gbrain-cycle:src-b', 900003, LOCAL, OLD_S, true);                // dead, TTL NOT expired
+
+    const { reaped, reapedIds } = await reapDeadHolderLocks(engine, killSeam(new Set()));
+    expect(reaped).toBe(3);
+    expect(reapedIds.sort()).toEqual(['gbrain-cycle', 'gbrain-cycle:src-b', 'gbrain-sync:src-a']);
+    expect(await lockIds()).toEqual([]);
+  });
+
+  test('keeps a live same-host holder', async () => {
+    await seedLock('gbrain-sync:src-live', process.pid, LOCAL, OLD_S);
+    const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set([process.pid])));
+    expect(reaped).toBe(0);
+    expect(await lockIds()).toEqual(['gbrain-sync:src-live']);
+  });
+
+  test('keeps a dead holder still within the PID-reuse grace window', async () => {
+    await seedLock('gbrain-sync:src-young', 900010, LOCAL, YOUNG_S);
+    const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set()));
+    expect(reaped).toBe(0);
+    expect(await lockIds()).toEqual(['gbrain-sync:src-young']);
+  });
+
+  test('keeps a cross-host holder (cannot probe a remote PID)', async () => {
+    await seedLock('gbrain-sync:src-xhost', 900011, 'a-different-host', OLD_S);
+    const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set()));
+    expect(reaped).toBe(0);
+    expect(await lockIds()).toEqual(['gbrain-sync:src-xhost']);
+  });
+
+  test('NEVER reaps a non-sync/cycle namespace, even with a dead PID (blast radius)', async () => {
+    await seedLock('gbrain-election:leader', 900020, LOCAL, OLD_S);
+    await seedLock('some-other-lock', 900021, LOCAL, OLD_S);
+    const { reaped } = await reapDeadHolderLocks(engine, killSeam(new Set()));
+    expect(reaped).toBe(0);
+    expect(await lockIds()).toEqual(['gbrain-election:leader', 'some-other-lock']);
+  });
+
+  test('empty table → {reaped:0}', async () => {
+    const r = await reapDeadHolderLocks(engine, killSeam(new Set()));
+    expect(r).toEqual({ reaped: 0, reapedIds: [] });
+  });
+
+  test('mixed set: reaps only the eligible rows', async () => {
+    await seedLock('gbrain-sync:dead', 900030, LOCAL, OLD_S);          // reap
+    await seedLock('gbrain-sync:live', process.pid, LOCAL, OLD_S);     // keep (live)
+    await seedLock('gbrain-cycle:xhost', 900031, 'other', OLD_S);      // keep (cross-host)
+    await seedLock('gbrain-election:x', 900032, LOCAL, OLD_S);         // keep (namespace)
+    const { reaped, reapedIds } = await reapDeadHolderLocks(engine, killSeam(new Set([process.pid])));
+    expect(reaped).toBe(1);
+    expect(reapedIds).toEqual(['gbrain-sync:dead']);
+    expect(await lockIds()).toEqual(['gbrain-cycle:xhost', 'gbrain-election:x', 'gbrain-sync:live']);
+  });
+});
+
+describe('deleteLockRowExact (snapshot-matched, TOCTOU defense)', () => {
+  test('no-op when acquired_at does not match; deletes when it does', async () => {
+    await seedLock('gbrain-sync:exact', 900040, LOCAL, OLD_S);
+    const snap = await inspectLock(engine, 'gbrain-sync:exact');
+    expect(snap).not.toBeNull();
+
+    // Simulate a reused-PID takeover: same id + pid, but a different (newer)
+    // acquired_at than the one the reaper snapshotted → must NOT delete.
+    const wrong = new Date(snap!.acquired_at.getTime() + 3_600_000);
+    const miss = await deleteLockRowExact(engine, 'gbrain-sync:exact', 900040, wrong);
+    expect(miss.deleted).toBe(false);
+    expect(await lockIds()).toEqual(['gbrain-sync:exact']);
+
+    // The matching snapshot deletes.
+    const hit = await deleteLockRowExact(engine, 'gbrain-sync:exact', 900040, snap!.acquired_at);
+    expect(hit.deleted).toBe(true);
+    expect(await lockIds()).toEqual([]);
+  });
+
+  test('no-op when holder_pid does not match', async () => {
+    await seedLock('gbrain-sync:pidguard', 900050, LOCAL, OLD_S);
+    const snap = await inspectLock(engine, 'gbrain-sync:pidguard');
+    const res = await deleteLockRowExact(engine, 'gbrain-sync:pidguard', 111111, snap!.acquired_at);
+    expect(res.deleted).toBe(false);
+    expect(await lockIds()).toEqual(['gbrain-sync:pidguard']);
+  });
+});

test/extract-abort.test.ts

@@ -0,0 +1,67 @@
+/**
+ * #1972 (Codex #7) — cooperative abort on the cycle-reachable extract paths.
+ *
+ * The cycle calls runExtractCore with `slugs` defined (incremental →
+ * extractForSlugs) or undefined (full walk → extractLinksFromDir /
+ * extractTimelineFromDir). All three must bail promptly on a pre-aborted
+ * signal. Behavioral, dryRun-only (no DB writes, so no FK setup needed).
+ */
+
+import { describe, test, expect, beforeAll, afterAll } from 'bun:test';
+import { mkdtempSync, writeFileSync, rmSync } from 'fs';
+import { join } from 'path';
+import { tmpdir } from 'os';
+import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import { runExtractCore } from '../src/commands/extract.ts';
+
+let engine: PGLiteEngine;
+let dir: string;
+
+beforeAll(async () => {
+  engine = new PGLiteEngine();
+  await engine.connect({ database_url: '' });
+  await engine.initSchema();
+  dir = mkdtempSync(join(tmpdir(), 'extract-abort-'));
+  // a.md links to b; both exist so the wikilink resolves.
+  writeFileSync(join(dir, 'a.md'), '# A\n\nsee [[b]] for more.\n');
+  writeFileSync(join(dir, 'b.md'), '# B\n\nplain.\n');
+});
+
+afterAll(async () => {
+  await engine.disconnect();
+  rmSync(dir, { recursive: true, force: true });
+});
+
+function aborted(): AbortSignal {
+  const a = new AbortController();
+  a.abort(new Error('timeout'));
+  return a.signal;
+}
+
+describe('extract cooperative abort', () => {
+  test('incremental path (extractForSlugs) processes 0 pages when pre-aborted', async () => {
+    const r = await runExtractCore(engine, {
+      mode: 'links', dir, dryRun: true, slugs: ['a', 'b'], signal: aborted(),
+    });
+    expect(r.pages_processed).toBe(0);
+  });
+
+  test('incremental path processes all pages without a signal', async () => {
+    const r = await runExtractCore(engine, {
+      mode: 'links', dir, dryRun: true, slugs: ['a', 'b'],
+    });
+    expect(r.pages_processed).toBe(2);
+  });
+
+  test('full-walk path (extractLinksFromDir) creates 0 links when pre-aborted', async () => {
+    const r = await runExtractCore(engine, {
+      mode: 'links', dir, dryRun: true, signal: aborted(),
+    });
+    expect(r.links_created).toBe(0);
+  });
+
+  test('full-walk path resolves the wikilink without a signal', async () => {
+    const r = await runExtractCore(engine, { mode: 'links', dir, dryRun: true });
+    expect(r.links_created).toBeGreaterThanOrEqual(1);
+  });
+});

test/phantom-redirect-per-source-lock.test.ts

@@ -31,7 +31,11 @@ describe('phantom-redirect lock contract', () => {
   test('IRON-RULE: acquireLockWithRetry call passes syncLockId(sourceId)', () => {
     // Banned: the bare `SYNC_LOCK_ID` constant slipped back in.
     // Required: the per-source helper threaded with the active sourceId.
-    expect(SRC).toMatch(/acquireLockWithRetry\s*\(\s*engine\s*,\s*syncLockId\s*\(\s*sourceId\s*\)\s*\)/);
+    // #1972: the call now threads an optional abort signal as a third arg
+    // (acquireLockWithRetry(engine, syncLockId(sourceId), signal)). The IRON-RULE
+    // is unchanged: it must pass the per-source syncLockId(sourceId), never bare
+    // SYNC_LOCK_ID. Allow the optional trailing signal arg.
+    expect(SRC).toMatch(/acquireLockWithRetry\s*\(\s*engine\s*,\s*syncLockId\s*\(\s*sourceId\s*\)\s*(?:,\s*signal\s*)?\)/);
     expect(SRC).not.toMatch(/acquireLockWithRetry\s*\(\s*engine\s*,\s*SYNC_LOCK_ID\s*\)/);
   });
 

test/postgres-disconnect-bounded.test.ts

@@ -0,0 +1,40 @@
+/**
+ * #1972 — gbrain-owned hard bound on pool teardown.
+ *
+ * The bug: `pool.end()` against PgBouncer transaction-mode never drains, so
+ * disconnect blocked until the CLI's 10s force-exit fired and truncated stdout.
+ * postgres.js's own `{ timeout }` is internal (a stub ignores it; it's not a
+ * guarantee we own), so `endPoolBounded` wraps every end in a Promise.race we
+ * control. These tests assert the bound is real (resolves even when `.end()`
+ * never settles) and that we still pass `{ timeout }` so a healthy drain is fast.
+ */
+
+import { describe, test, expect } from 'bun:test';
+import { endPoolBounded, POOL_END_TIMEOUT_SECONDS } from '../src/core/db.ts';
+
+describe('endPoolBounded', () => {
+  test('resolves fast when .end() settles quickly, forwarding { timeout }', async () => {
+    let calledWith: unknown;
+    const pool = { end: async (opts?: { timeout?: number }) => { calledWith = opts; } };
+    const t0 = Date.now();
+    await endPoolBounded(pool);
+    expect(Date.now() - t0).toBeLessThan(500);
+    expect(calledWith).toEqual({ timeout: POOL_END_TIMEOUT_SECONDS });
+  });
+
+  test('resolves within the gbrain bound even when .end() NEVER settles', async () => {
+    // This is the PgBouncer hang: .end() returns a promise that never resolves.
+    // The bare `await pool.end()` would hang until the CLI's 10s force-exit.
+    const pool = { end: () => new Promise<void>(() => { /* never resolves */ }) };
+    const t0 = Date.now();
+    await endPoolBounded(pool);
+    const elapsed = Date.now() - t0;
+    expect(elapsed).toBeGreaterThanOrEqual(POOL_END_TIMEOUT_SECONDS * 1000);
+    expect(elapsed).toBeLessThan(5000); // well under the CLI's 10s force-exit deadline
+  });
+
+  test('never throws when .end() rejects (teardown must not propagate)', async () => {
+    const pool = { end: async () => { throw new Error('pool boom'); } };
+    await expect(endPoolBounded(pool)).resolves.toBeUndefined();
+  });
+});

test/postgres-engine-singleton-ownership.test.ts

@@ -81,7 +81,10 @@ describe('postgres-engine / module-singleton ownership (#1471)', () => {
     const disconnect = stripComments(extractFn(DB_SRC, 'disconnect'));
     const snapshotIdx = disconnect.search(/const\s+s\s*=\s*sql/);
     const nullIdx = disconnect.search(/\bsql\s*=\s*null/);
-    const endIdx = disconnect.search(/\bawait\s+s\.end\s*\(/);
+    // #1972: the pool end is now wrapped in the gbrain-owned hard bound
+    // `endPoolBounded(s)` instead of a bare `s.end()`. The ordering contract is
+    // unchanged: snapshot + null the singleton BEFORE awaiting the end.
+    const endIdx = disconnect.search(/\bawait\s+(?:s\.end\s*\(|endPoolBounded\s*\(\s*s\b)/);
     expect(snapshotIdx).toBeGreaterThanOrEqual(0);
     expect(nullIdx).toBeGreaterThanOrEqual(0);
     expect(endIdx).toBeGreaterThanOrEqual(0);