fix: pre-landing review hardening — federated alias parallelism, trust-boundary clamps, shared protocol helpers (#2095)

Five specialist reviewers (testing/maintainability/security/performance/
data-migration) on the reconciled diff; every finding applied:

Performance: the alias arm now resolves all granted sources CONCURRENTLY
(a federated caller paid M sequential RTTs per turn — ~355ms at 5 sources
cross-region, inside the reflex's 1.5s budget); watch's session dedupe is
O(1) Set membership instead of a monotonically growing priorContext string
(O(T²) over a long-lived session); getWindowTurns iterates from the tail
(per-turn cost no longer grows with session length); the resolver's
provenance maps fold into the existing candidate pass.

Security: volunteer_context clamps caller-supplied attribution at the trust
boundary — session_id capped at 256 chars (a read-scoped token could bank
~1MiB TEXT per request, retained 90 days), turn logged only when a safe
integer (a non-integer threw inside the batched INSERT and silently dropped
the whole batch). The privacy comments now state precisely what rationale
may contain (the matched entity's surface form — which by construction
resolved to an existing alias/title/slug — never free conversation text).

Maintainability: TURN_PREFIX_RE + formatVolunteeredPage exported from
volunteer.ts and shared by watch/cli (the two surfaces can no longer
drift); volunteerEventRowsFrom is the single VolunteerEventRow assembly
site for all three channels; watch's window default now honors the same
retrieval_reflex_window_turns config knob the reflex reads; the stale
pre-v116 comments swept to pre-v117.

Testing: the two flake-class CRITICALs fixed (pipe test asserts the
backstop banner instead of a cold-CI-hostile 9s wall bound; the SIGINT test
waits on watch's new machine-readable ready line instead of a fixed 15s
sleep — 2.5s and deterministic now); new coverage for the sink's timeout
branch + ghost-reference drop, watch per-turn fail-open, untrusted knob
clamps (min_confidence/max_pages/days), window-cap ordering (newest user
mention survives), serve-IPC suppression passthrough + channel=reflex
logging, windowTurnCount edge semantics, and structural pins for the sink
registration + cycle purge wiring. The exit-verdict pin's grep is now
operator/whitespace-tolerant.

Deferred with TODOs: resolver index shapes for the per-turn query;
batched first-prune after a long dream-cycle gap.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

TODOS.md

@@ -23,6 +23,24 @@ are the bar). Plan + GSTACK REVIEW REPORT at
   string window (`user:`/`assistant:` prefixes) to avoid a dual-shape contract.
   If MCP callers accumulate parsing bugs, add a structured array param beside it.
   **Where:** `src/core/operations.ts:volunteer_context` + `src/core/context/volunteer.ts:parseWindow`.
+- [ ] **P3 — index shapes for the per-turn resolver query.** The arm-2 resolver
+  (`retrieval-reflex.ts`: `lower(title) = ANY() OR slug = ANY() OR slug LIKE
+  ANY('%/...')`) predates #2095 but now runs per turn on three channels
+  (reflex window, volunteer_context, watch) federated across sources. Neither
+  the leading-wildcard suffix arm nor `lower(title)` is index-served. If
+  per-turn latency telemetry on large brains comes back hot: add
+  `(source_id, lower(title))` btree + a reverse(slug) text_pattern_ops (or
+  gin_trgm) index, or split the OR into three index-friendly queries.
+  **Where:** `src/core/context/retrieval-reflex.ts`, migration.
+- [ ] **P3 — batch the volunteer-events pruner's first run after a long gap.**
+  `purgeStaleVolunteerEvents` is one unbatched DELETE with a bare
+  `volunteered_at` predicate (full scan; fine for a TTL-bounded table). Edge:
+  a brain whose dream cycle was off for months could hit the pooler's ~2min
+  statement_timeout on the first prune, get swallowed by the catch, and never
+  make progress. If observed: id-batched chunks (`DELETE ... WHERE id IN
+  (SELECT ... LIMIT 10000)` looped). **Where:**
+  `src/core/context/volunteer-events.ts:purgeStaleVolunteerEvents`.
+
 ## gbrain triage wave follow-ups (filed v0.42.41.0)
 
 Deferred from the v0.42.41.0 fix wave (eng-reviewed as separate scope, not hotfixes).

src/cli.ts

@@ -24,6 +24,7 @@ import type { GBrainConfig } from './core/config.ts';
 import type { AIGatewayConfig } from './core/ai/types.ts';
 import type { BrainEngine } from './core/engine.ts';
 import { operations, OperationError } from './core/operations.ts';
+import { formatVolunteeredPage } from './core/context/volunteer.ts';
 import type { Operation, OperationContext } from './core/operations.ts';
 import { shouldForceExitAfterMain, finishCliTeardown, flushThenExit, currentExitCode, setCliExitVerdict } from './core/cli-force-exit.ts';
 import { serializeMarkdown } from './core/markdown.ts';
@@ -823,10 +824,7 @@ export function formatResult(opName: string, result: unknown): string {
       }
       const pages = (r?.pages ?? []) as any[];
       if (!pages.length) return 'Nothing volunteered (no entity cleared the confidence gate).\n';
-      return pages.map((p) =>
-        `${p.display} → ${p.slug} (${p.confidence.toFixed(2)}, ${p.arm}) — ${p.rationale}` +
-        (p.synopsis ? `\n    ${p.synopsis}` : ''),
-      ).join('\n') + '\n';
+      return pages.map((p) => formatVolunteeredPage(p)).join('\n') + '\n';
     }
     case 'get_page': {
       const r = result as any;

src/commands/watch.ts

@@ -25,12 +25,15 @@ import { createInterface } from 'node:readline';
 import type { BrainEngine } from '../core/engine.ts';
 import {
   volunteerContext,
+  formatVolunteeredPage,
+  TURN_PREFIX_RE,
   VOLUNTEER_DEFAULT_MAX_PAGES,
   VOLUNTEER_DEFAULT_MIN_CONFIDENCE,
 } from '../core/context/volunteer.ts';
 import type { WindowTurn } from '../core/context/entity-salience.ts';
-import { DEFAULT_WINDOW_TURNS } from '../core/context/reflex.ts';
-import { logVolunteerEventsFireAndForget } from '../core/context/volunteer-events.ts';
+import { DEFAULT_WINDOW_TURNS, windowTurnCount } from '../core/context/reflex.ts';
+import { loadConfig } from '../core/config.ts';
+import { logVolunteerEventsFireAndForget, volunteerEventRowsFrom } from '../core/context/volunteer-events.ts';
 
 export const WATCH_HELP = `gbrain watch — push-based context: volunteer brain pages per conversation turn (#2095)
 
@@ -52,8 +55,6 @@ Flags:
   --help                 this text
 `;
 
-const TURN_PREFIX_RE = /^(user|assistant)\s*:\s?(.*)$/i;
-
 function numFlag(args: string[], flag: string): number | undefined {
   const i = args.indexOf(flag);
   if (i < 0 || i + 1 >= args.length) return undefined;
@@ -80,7 +81,12 @@ export async function runWatch(engine: BrainEngine, args: string[], deps: WatchI
   }
 
   const json = args.includes('--json');
-  const windowTurns = Math.max(1, Math.floor(numFlag(args, '--window-turns') ?? DEFAULT_WINDOW_TURNS));
+  // --window-turns wins; otherwise the same config knob the ambient reflex
+  // honors (retrieval_reflex_window_turns, default 4) applies here too.
+  const windowTurns = Math.max(
+    1,
+    Math.floor(numFlag(args, '--window-turns') ?? windowTurnCount(loadConfig())),
+  );
   const maxPages = numFlag(args, '--max-pages');
   const minConfidence = numFlag(args, '--min-confidence');
   const write = deps.write ?? ((s: string) => process.stdout.write(s));
@@ -91,9 +97,14 @@ export async function runWatch(engine: BrainEngine, args: string[], deps: WatchI
   const sourceIds = [sourceId];
   const sessionId = `watch-${process.pid}-${Date.now().toString(36)}`;
 
-  if (isTTY && !deps.lines) {
+  if (!deps.lines) {
+    // The ready line doubles as a machine-readable readiness signal for
+    // scripted consumers (and the SIGINT lifecycle test): engine + source
+    // resolution are done, the stdin loop starts next.
     process.stderr.write(
-      `[watch] interactive session ${sessionId} — type turns ('assistant: ...' to set role), Ctrl-C to end\n`,
+      isTTY
+        ? `[watch] interactive session ${sessionId} ready — type turns ('assistant: ...' to set role), Ctrl-C to end\n`
+        : `[watch] session ${sessionId} ready\n`,
     );
   }
 
@@ -132,28 +143,21 @@ export async function runWatch(engine: BrainEngine, args: string[], deps: WatchI
           sourceIds,
           maxPages,
           minConfidence,
-          // Already-pushed slugs ride priorContext → the core's slug-only
-          // suppression dedupes for the whole session.
-          priorContext: pushedSlugs.size ? Array.from(pushedSlugs).join('\n') : undefined,
         });
       } catch {
         continue; // fail-open per turn: a transient DB error never kills the stream
       }
+      // Session dedupe via O(1) Set membership — a slug is volunteered at most
+      // once per session. (Feeding pushed slugs back as priorContext would
+      // rebuild + rescan a monotonically growing string every turn: O(T²)
+      // over a long-lived session.)
+      pages = pages.filter((p) => !pushedSlugs.has(p.slug));
       if (!pages.length) continue;
 
       for (const p of pages) pushedSlugs.add(p.slug);
       logVolunteerEventsFireAndForget(
         engine,
-        pages.map((p) => ({
-          source_id: p.source_id,
-          slug: p.slug,
-          confidence: p.confidence,
-          match_arm: p.arm,
-          rationale: p.rationale,
-          channel: 'watch' as const,
-          session_id: sessionId,
-          turn: turnNo,
-        })),
+        volunteerEventRowsFrom(pages, { channel: 'watch', session_id: sessionId, turn: turnNo }),
       );
 
       if (json) {
@@ -162,10 +166,7 @@ export async function runWatch(engine: BrainEngine, args: string[], deps: WatchI
         }
       } else {
         for (const p of pages) {
-          write(
-            `${p.display} → ${p.slug} (${p.confidence.toFixed(2)}, ${p.arm}) — ${p.rationale}` +
-              (p.synopsis ? `\n    ${p.synopsis}` : '') + '\n',
-          );
+          write(formatVolunteeredPage(p) + '\n');
         }
       }
     }

src/core/context-engine.ts

@@ -187,14 +187,19 @@ function getLastUserText(messages: AgentMessage[]): string {
  */
 const WINDOW_TURNS_HARD_CAP = 12;
 function getWindowTurns(messages: AgentMessage[]): Array<{ role: 'user' | 'assistant'; text: string }> {
+  // Iterate from the END: this runs on the per-turn hot path (1.5s reflex
+  // budget) and only the last 12 turns matter — flattening every content
+  // block of a multi-hundred-turn session just to slice the tail would make
+  // the cost grow with session length.
   const out: Array<{ role: 'user' | 'assistant'; text: string }> = [];
-  for (const m of messages) {
+  for (let i = messages.length - 1; i >= 0 && out.length < WINDOW_TURNS_HARD_CAP; i--) {
+    const m = messages[i];
     if (m?.role !== 'user' && m?.role !== 'assistant') continue;
     const text = messageText(m.content);
     if (!text) continue;
     out.push({ role: m.role, text });
   }
-  return out.slice(-WINDOW_TURNS_HARD_CAP);
+  return out.reverse();
 }
 
 /**

src/core/context/retrieval-reflex.ts

@@ -148,16 +148,23 @@ export async function resolveEntitiesToPointers(
   const titlesLc: string[] = [];
   const exactSlugs: string[] = [];
   const slugSuffixes: string[] = [];
+  // Reverse maps for arm-2 provenance (which candidate produced a row) —
+  // populated in this same pass so the derivations happen exactly once.
+  const titleToNorm = new Map<string, string>();
+  const slugToNorm = new Map<string, string>();
   for (const c of candidates) {
     const norm = normalizeAlias(c.query);
     if (!norm) continue;
     if (!displayByNorm.has(norm)) displayByNorm.set(norm, c.display);
     aliasNorms.push(norm);
-    titlesLc.push(c.query.toLowerCase());
+    const tl = c.query.toLowerCase();
+    titlesLc.push(tl);
+    if (!titleToNorm.has(tl)) titleToNorm.set(tl, norm);
     const s = slugify(c.query);
     if (s) {
       exactSlugs.push(s);
       slugSuffixes.push(`%/${s}`);
+      if (!slugToNorm.has(s)) slugToNorm.set(s, norm);
     }
   }
   if (!aliasNorms.length) return null;
@@ -179,31 +186,24 @@ export async function resolveEntitiesToPointers(
       resolved.push({ slug, source_id: src, arm, matchedNorm });
     }
   };
-  // Reverse maps for arm-2 provenance: which CANDIDATE produced this row
-  // (titles/slugs were built per candidate above, in the same pass).
-  const titleToNorm = new Map<string, string>();
-  const slugToNorm = new Map<string, string>();
-  for (const c of candidates) {
-    const norm = normalizeAlias(c.query);
-    if (!norm) continue;
-    const tl = c.query.toLowerCase();
-    if (!titleToNorm.has(tl)) titleToNorm.set(tl, norm);
-    const s = slugify(c.query);
-    if (s && !slugToNorm.has(s)) slugToNorm.set(s, norm);
-  }
-
   // Arm 1 — alias-first. Unambiguous single-slug hits only, per source (no
   // engine-interface change for federation). Guarded: pre-v110 brains throw
   // "relation page_aliases does not exist" — swallow and continue.
-  for (const src of sourceIds) {
-    try {
-      const aliasMap = await engine.resolveAliases(aliasNorms, { sourceId: src });
-      for (const norm of aliasNorms) {
-        const hits = aliasMap.get(norm);
-        if (hits && hits.length === 1) push(hits[0].slug, src, 'alias', norm);
-      }
-    } catch {
-      /* no page_aliases table (pre-v110) — degrade to the title/slug arm */
+  // Per-source lookups are independent — run them concurrently so a
+  // federated caller (M granted sources) pays one RTT, not M sequential
+  // ones (~71ms each cross-region; the reflex runs under a 1.5s budget).
+  // Results are folded back in sourceIds order so pointer ordering stays
+  // deterministic. Per-source failures degrade independently (pre-v110
+  // brains have no page_aliases table).
+  const aliasResults = await Promise.allSettled(
+    sourceIds.map((src) => engine.resolveAliases(aliasNorms, { sourceId: src })),
+  );
+  for (let i = 0; i < sourceIds.length; i++) {
+    const r = aliasResults[i];
+    if (r.status !== 'fulfilled') continue;
+    for (const norm of aliasNorms) {
+      const hits = r.value.get(norm);
+      if (hits && hits.length === 1) push(hits[0].slug, sourceIds[i], 'alias', norm);
     }
   }
 
@@ -288,17 +288,13 @@ export async function resolveEntitiesToPointers(
   // keeps the reflex hot path dependency-free when logging is off.
   if (opts.logChannel) {
     try {
-      const { logVolunteerEventsFireAndForget } = await import('./volunteer-events.ts');
+      const { logVolunteerEventsFireAndForget, volunteerEventRowsFrom } = await import('./volunteer-events.ts');
       logVolunteerEventsFireAndForget(
         engine,
-        pointers.map((p) => ({
-          source_id: p.source_id,
-          slug: p.slug,
-          confidence: p.confidence,
-          match_arm: p.arm,
-          rationale: `${p.arm} match "${p.display}"`,
-          channel: opts.logChannel as 'reflex',
-        })),
+        volunteerEventRowsFrom(
+          pointers.map((p) => ({ ...p, rationale: `${p.arm} match "${p.display}"` })),
+          { channel: opts.logChannel as 'reflex' },
+        ),
       );
     } catch {
       /* telemetry only — never blocks the pointer block */

src/core/context/volunteer-events.ts

@@ -13,8 +13,9 @@
  *
  * Retention: rows older than VOLUNTEER_EVENTS_TTL_DAYS are pruned by the
  * dream cycle's purge phase so conversation-adjacent telemetry never grows
- * unbounded. rationale is a deterministic template string, never raw
- * conversation text.
+ * unbounded. rationale is a deterministic template that may embed the matched
+ * entity's surface form (which by construction resolved to an existing
+ * alias/title/slug) — never free conversation text.
  */
 
 import type { BrainEngine } from './../engine.ts';
@@ -24,6 +25,27 @@ export const VOLUNTEER_EVENTS_TTL_DAYS = 90;
 
 export type VolunteerChannel = 'op' | 'reflex' | 'watch';
 
+/**
+ * Map volunteered pages to event rows for one channel — the ONE place the
+ * VolunteerEventRow shape is assembled (op / reflex / watch all call this,
+ * so adding a column is a one-site change).
+ */
+export function volunteerEventRowsFrom(
+  pages: Array<{ source_id: string; slug: string; confidence: number; arm: string; rationale: string }>,
+  opts: { channel: VolunteerChannel; session_id?: string | null; turn?: number | null },
+): VolunteerEventRow[] {
+  return pages.map((p) => ({
+    source_id: p.source_id,
+    slug: p.slug,
+    confidence: p.confidence,
+    match_arm: p.arm,
+    rationale: p.rationale,
+    channel: opts.channel,
+    session_id: opts.session_id ?? null,
+    turn: opts.turn ?? null,
+  }));
+}
+
 export interface VolunteerEventRow {
   source_id: string;
   slug: string;
@@ -81,7 +103,7 @@ const pendingVolunteerEventWrites = new Set<Promise<unknown>>();
 
 /**
  * Log volunteered pages without blocking the hot path. The batched INSERT
- * runs as a tracked dangling promise; errors are swallowed (pre-v116 brains,
+ * runs as a tracked dangling promise; errors are swallowed (pre-v117 brains,
  * transient DB failures — the volunteer result is unaffected).
  */
 export function logVolunteerEventsFireAndForget(
@@ -140,7 +162,7 @@ export function _peekPendingVolunteerEventWritesForTests(): number {
 
 /**
  * 90-day GC, called from the dream cycle's purge phase (mirrors
- * purgeStaleCheckpoints). Best-effort: returns 0 on any failure (pre-v116
+ * purgeStaleCheckpoints). Best-effort: returns 0 on any failure (pre-v117
  * brains have no table yet).
  */
 export async function purgeStaleVolunteerEvents(

src/core/context/volunteer.ts

@@ -63,7 +63,9 @@ export interface VolunteerOpts {
   minConfidence?: number;
 }
 
-const TURN_PREFIX_RE = /^(user|assistant)\s*:\s?(.*)$/i;
+/** Shared wire protocol for window turns — watch.ts imports this so the two
+ * channels can never desynchronize on the prefix grammar. */
+export const TURN_PREFIX_RE = /^(user|assistant)\s*:\s?(.*)$/i;
 
 /**
  * Lenient window parser: `user:` / `assistant:` line prefixes start a new
@@ -170,6 +172,18 @@ export async function volunteerContext(
   return out;
 }
 
+/**
+ * Canonical human rendering of one volunteered page — shared by
+ * `gbrain volunteer-context` (cli.ts formatResult) and `gbrain watch` so the
+ * two surfaces can't drift.
+ */
+export function formatVolunteeredPage(p: VolunteeredPage): string {
+  return (
+    `${p.display} → ${p.slug} (${p.confidence.toFixed(2)}, ${p.arm}) — ${p.rationale}` +
+    (p.synopsis ? `\n    ${p.synopsis}` : '')
+  );
+}
+
 // ── Usage stats (the feedback loop) ──────────────────────────────────────
 
 export interface VolunteerArmStats {
@@ -198,7 +212,7 @@ export const VOLUNTEER_STATS_NOTE =
 
 /**
  * Per-arm/channel precision over the last N days, source-scoped. Read-only;
- * returns zeroed stats on pre-v116 brains (no table).
+ * returns zeroed stats on pre-v117 brains (no table).
  */
 export async function volunteerUsageStats(
   engine: BrainEngine,
@@ -222,7 +236,7 @@ export async function volunteerUsageStats(
       [sourceIds, String(safeDays)],
     );
   } catch {
-    rows = []; // pre-v116 brain — table doesn't exist yet
+    rows = []; // pre-v117 brain — table doesn't exist yet
   }
   const by_arm: VolunteerArmStats[] = rows.map((r) => {
     const volunteered = Number(r.volunteered);

src/core/cycle.ts

@@ -1287,7 +1287,7 @@ async function runPhasePurge(engine: BrainEngine, dryRun: boolean): Promise<Phas
     }
     // v0.43 (#2095) — 90-day GC of the volunteered-context feedback log.
     // Conversation-adjacent telemetry must never grow unbounded. Best-effort:
-    // purgeStaleVolunteerEvents returns 0 on pre-v116 brains (no table).
+    // purgeStaleVolunteerEvents returns 0 on pre-v117 brains (no table).
     let purgedVolunteerEvents = 0;
     try {
       const { purgeStaleVolunteerEvents } = await import('./context/volunteer-events.ts');

src/core/operations.ts

@@ -3215,19 +3215,19 @@ const volunteer_context: Operation = {
     // volunteer-events sink (drained at exit). Never fails the op.
     if (pages.length) {
       try {
-        const { logVolunteerEventsFireAndForget } = await import('./context/volunteer-events.ts');
+        const { logVolunteerEventsFireAndForget, volunteerEventRowsFrom } = await import('./context/volunteer-events.ts');
+        // Trust-boundary clamps (remote MCP callers): cap session_id length so
+        // a read-scoped token can't bank unbounded TEXT per request, and only
+        // log integer turns — a non-integer would throw inside the single
+        // multi-row INSERT and silently drop the whole batch.
+        const sessionId = typeof p.session_id === 'string' ? p.session_id.slice(0, 256) : null;
+        const turn =
+          typeof p.turn === 'number' && Number.isInteger(p.turn) && Math.abs(p.turn) <= 2_147_483_647
+            ? p.turn
+            : null;
         logVolunteerEventsFireAndForget(
           ctx.engine,
-          pages.map((pg) => ({
-            source_id: pg.source_id,
-            slug: pg.slug,
-            confidence: pg.confidence,
-            match_arm: pg.arm,
-            rationale: pg.rationale,
-            channel: 'op' as const,
-            session_id: typeof p.session_id === 'string' ? p.session_id : null,
-            turn: typeof p.turn === 'number' ? p.turn : null,
-          })),
+          volunteerEventRowsFrom(pages, { channel: 'op', session_id: sessionId, turn }),
         );
       } catch {
         /* telemetry only */

test/cli-exit-verdict-pin.test.ts

@@ -22,8 +22,11 @@ describe('exit-verdict ownership — no raw process.exitCode assignments', () =>
     //    RESTORE around PGlite.create() — it keeps the GLOBAL tidy for
     //    external readers and is explicitly not a verdict write (the owned
     //    channel never reads process.exitCode).
+    // Whitespace/operator-tolerant: catches `process.exitCode=1`,
+    // `process.exitCode ??= 1`, `process.exitCode ||= 1`, and the bracket
+    // form — every shape is equally zeroed by the owned-verdict read.
     const hits = execSync(
-      `grep -rn "process.exitCode = " src --include='*.ts' | grep -v "core/cli-force-exit.ts" | grep -v "core/pglite-engine.ts" || true`,
+      String.raw`grep -rnE "process(\.|\[')exitCode('\])?[[:space:]]*([?|&]{2})?=[^=]" src --include='*.ts' | grep -v "core/cli-force-exit.ts" | grep -v "core/pglite-engine.ts" || true`,
       { encoding: 'utf-8', cwd: new URL('..', import.meta.url).pathname },
     ).trim();
     expect(hits).toBe('');

test/cli-pipe-truncation.test.ts

@@ -26,7 +26,7 @@ describe('cli pipe completeness — deliberate exit never truncates piped stdout
         env: { ...process.env, GBRAIN_SKIP_STARTUP_HOOKS: '1' },
         maxBuffer: 64 * 1024 * 1024,
       });
-      return { stdout: res.stdout ?? '', status: res.status, ms: Date.now() - t0 };
+      return { stdout: res.stdout ?? '', stderr: res.stderr ?? '', status: res.status, ms: Date.now() - t0 };
     };
     const first = run();
     expect(first.status).toBe(0);
@@ -34,8 +34,11 @@ describe('cli pipe completeness — deliberate exit never truncates piped stdout
     // Truncated JSON does not parse — the strongest single-run completeness check.
     const parsed = JSON.parse(first.stdout);
     expect(Array.isArray(parsed)).toBe(true);
-    // Deliberate exit, not the teardown backstop.
-    expect(first.ms).toBeLessThan(9_000);
+    // Deliberate exit, not the teardown backstop. A wall-clock bound is flaky
+    // on cold CI (bun parse alone runs 10-20s there) — the backstop's banner
+    // is the truthful signal, same assertion the pgbouncer e2e uses.
+    expect(first.stderr).not.toContain('force-exiting');
+    expect(first.stderr).not.toContain('did not return within');
 
     const second = run();
     expect(second.status).toBe(0);

test/fix-wave-structural.test.ts

@@ -250,3 +250,20 @@ describe('v0.41.8.0 #1340 — PGLite WASM init classifier', () => {
     expect(src).toMatch(/buildPgliteInitErrorMessage\(verdict, original\)/);
   });
 });
+
+describe('v0.42.43.0 #2095 — volunteer-events sink + cycle purge wiring (structural pins)', () => {
+  test('volunteer-events registers a background-work drainer (order 4)', () => {
+    // Deleting this registration would silently drop volunteer events on
+    // every CLI exit with no behavioral test failing — same pin class as the
+    // other four sinks above.
+    const src = readFileSync('src/core/context/volunteer-events.ts', 'utf8');
+    expect(src).toMatch(/registerBackgroundWorkDrainer\(\{[\s\S]*?name:\s*'volunteer-events'/);
+    expect(src).toMatch(/order:\s*4/);
+  });
+
+  test("the dream cycle's purge phase invokes purgeStaleVolunteerEvents and reports the count", () => {
+    const src = readFileSync('src/core/cycle.ts', 'utf8');
+    expect(src).toMatch(/purgeStaleVolunteerEvents\(engine\)/);
+    expect(src).toMatch(/purged_volunteer_events_count/);
+  });
+});

test/retrieval-reflex.test.ts

@@ -333,3 +333,70 @@ describe('ambient-channel event logging (codex D11 — logChannel: reflex)', ()
     expect(rows.length).toBe(0);
   });
 });
+
+describe('serve IPC wiring — suppression passthrough + reflex-channel logging (review hardening)', () => {
+  test('the IPC round-trip honors slug-only suppression and logs channel=reflex', async () => {
+    const { startResolveIpcServer, resolveViaIpc, resolveSocketPath, IPC_UNAVAILABLE } =
+      await import('../src/core/context/resolve-ipc.ts');
+    const { awaitPendingVolunteerEventWrites, _resetPendingVolunteerEventWritesForTests } =
+      await import('../src/core/context/volunteer-events.ts');
+    const { mkdtempSync, rmSync } = await import('fs');
+    const { join } = await import('path');
+    const { tmpdir } = await import('os');
+
+    _resetPendingVolunteerEventWritesForTests();
+    await engine.executeRaw('DELETE FROM context_volunteer_events').catch(() => {});
+    await seed('people/alice-example', 'Alice Example', 'A founder.');
+
+    const dir = mkdtempSync(join(tmpdir(), 'rr-ipc-'));
+    const sock = resolveSocketPath(dir);
+    // The SAME handler shape src/mcp/server.ts wires for serve: forwards
+    // suppression from the request and logs on the ambient reflex channel.
+    const server = await startResolveIpcServer(sock, (req) =>
+      resolveEntitiesToPointers(engine, req.sourceId || 'default', req.candidates ?? [], {
+        priorContextText: req.priorContextText,
+        maxPointers: req.maxPointers,
+        suppression: req.suppression,
+        logChannel: 'reflex',
+      }),
+    );
+    expect(server).not.toBeNull();
+    try {
+      // slug-only suppression: a TITLE mention in prior context must NOT
+      // suppress (the windowing contract), and the resolve must log.
+      const block = await resolveViaIpc(sock, {
+        candidates: extractCandidates('tell me about Alice Example'),
+        priorContextText: 'earlier turn merely mentioned Alice Example',
+        suppression: 'slug-only',
+      });
+      expect(block).not.toBe(IPC_UNAVAILABLE);
+      expect(block).not.toBeNull();
+      expect((block as { pointers: Array<{ slug: string }> }).pointers[0].slug).toBe('people/alice-example');
+
+      const { unfinished } = await awaitPendingVolunteerEventWrites(5_000);
+      expect(unfinished).toBe(0);
+      const rows = await engine.executeRaw<{ channel: string }>(
+        'SELECT channel FROM context_volunteer_events', [],
+      );
+      expect(rows.length).toBe(1);
+      expect(rows[0].channel).toBe('reflex');
+    } finally {
+      server!.close();
+      rmSync(dir, { recursive: true, force: true });
+    }
+  });
+});
+
+describe('windowTurnCount — knob edge semantics', () => {
+  test('0, negative, NaN, and absent all fall back to the default of 4 (1 = legacy off)', async () => {
+    const { windowTurnCount, DEFAULT_WINDOW_TURNS } = await import('../src/core/context/reflex.ts');
+    expect(DEFAULT_WINDOW_TURNS).toBe(4);
+    expect(windowTurnCount(null)).toBe(4);
+    expect(windowTurnCount({ retrieval_reflex_window_turns: 0 } as never)).toBe(4);
+    expect(windowTurnCount({ retrieval_reflex_window_turns: -3 } as never)).toBe(4);
+    expect(windowTurnCount({ retrieval_reflex_window_turns: Number.NaN } as never)).toBe(4);
+    // The documented "off" switch is 1 (legacy single-turn), not 0.
+    expect(windowTurnCount({ retrieval_reflex_window_turns: 1 } as never)).toBe(1);
+    expect(windowTurnCount({ retrieval_reflex_window_turns: 6.9 } as never)).toBe(6);
+  });
+});

test/volunteer-context.test.ts

@@ -393,3 +393,76 @@ describe('volunteer_context op (contract surface)', () => {
     expect(stats.note).toContain('approximate');
   });
 });
+
+describe('knob clamps — untrusted MCP caller inputs (review hardening)', () => {
+  test('minConfidence outside [0,1] (or NaN) falls back to the 0.7 default gate', async () => {
+    await seed('projects/widget-co', 'The Widget Company Project', 'A project.');
+    const turns = parseWindow('user: updates on Widget-Co?');
+    for (const bad of [5, -1, Number.NaN]) {
+      const pages = await volunteerContext(engine, turns, { sourceIds: ['default'], minConfidence: bad });
+      expect(pages).toEqual([]); // slug-suffix (0.6+boost) stays gated at the default 0.7
+    }
+  });
+
+  test('maxPages 0 / negative / NaN fall back to the 3-page default', async () => {
+    for (let i = 0; i < 5; i++) await seed(`people/person-${i}`, `Person Alpha${i}`, 'A person.');
+    const text = Array.from({ length: 5 }, (_, i) => `Person Alpha${i}`).join(' and ');
+    for (const bad of [0, -3, Number.NaN]) {
+      const pages = await volunteerContext(engine, parseWindow(`user: intro ${text}`), {
+        sourceIds: ['default'],
+        maxPages: bad,
+      });
+      expect(pages.length).toBeLessThanOrEqual(3);
+      expect(pages.length).toBeGreaterThan(0);
+    }
+  });
+
+  test('stats days <= 0 / NaN falls back to 30', async () => {
+    const stats = await volunteerUsageStats(engine, ['default'], -5);
+    expect(stats.days).toBe(30);
+    const stats2 = await volunteerUsageStats(engine, ['default'], Number.NaN);
+    expect(stats2.days).toBe(30);
+  });
+});
+
+describe('window-cap ordering — the newest user mention survives the cap', () => {
+  test('stale assistant-only chatter is dropped before a newest-turn user entity', async () => {
+    const { extractCandidatesFromWindow: extract, MAX_CANDIDATES: CAP } = await import('../src/core/context/entity-salience.ts');
+    // 14 stale assistant-introduced entities in turn 1, then the user names
+    // ONE entity in the newest turn. The cap (12) must keep the user's.
+    const stale = Array.from({ length: 14 }, (_, i) => `Stale Chatter${i}`).join(', ');
+    const cands = extract([
+      { role: 'assistant', text: `consider ${stale}.` },
+      { role: 'user', text: 'actually ask Alice Example first' },
+    ]);
+    expect(cands.length).toBeLessThanOrEqual(CAP);
+    const alice = cands.find((c) => normalizeAlias(c.query) === normalizeAlias('Alice Example'));
+    expect(alice).toBeDefined();
+    // Recency + user-role weighting puts the newest user mention FIRST.
+    expect(normalizeAlias(cands[0].query)).toBe(normalizeAlias('Alice Example'));
+  });
+});
+
+describe('volunteer-events sink — timeout branch (long-lived process safety)', () => {
+  test('a hung write reports unfinished and drops the snapshot (no ghost references)', async () => {
+    const {
+      logVolunteerEventsFireAndForget,
+      awaitPendingVolunteerEventWrites,
+      _resetPendingVolunteerEventWritesForTests,
+      _peekPendingVolunteerEventWritesForTests,
+    } = await import('../src/core/context/volunteer-events.ts');
+    _resetPendingVolunteerEventWritesForTests();
+    const hangingEngine = {
+      executeRaw: () => new Promise(() => { /* never settles */ }),
+    } as never;
+    logVolunteerEventsFireAndForget(hangingEngine, [
+      { source_id: 'default', slug: 'people/x', confidence: 0.9, match_arm: 'alias', rationale: 'r', channel: 'watch' },
+    ]);
+    const { unfinished } = await awaitPendingVolunteerEventWrites(20);
+    expect(unfinished).toBe(1);
+    // Snapshot dropped so a long-lived `gbrain watch` never accumulates
+    // references to forever-pending work (the last-retrieved C1 class).
+    expect(_peekPendingVolunteerEventWritesForTests()).toBe(0);
+    _resetPendingVolunteerEventWritesForTests();
+  });
+});

test/watch-command.test.ts

@@ -153,3 +153,33 @@ describe('gbrain watch — window + cap flags (ship coverage G4)', () => {
     expect(out.length).toBe(1);
   });
 });
+
+describe('gbrain watch — per-turn fail-open (review hardening)', () => {
+  test('a transient DB error on one turn never kills the stream', async () => {
+    await seed('people/alice-example', 'Alice Example', 'Alice is a founder.');
+    // Fail the FIRST resolver query against pages (turn 1's resolution);
+    // everything else — incl. resolveSourceId's pre-loop check — passes.
+    let pagesQueries = 0;
+    const flaky = new Proxy(engine, {
+      get(target, prop, receiver) {
+        if (prop === 'executeRaw') {
+          return (sql: string, params: unknown[]) => {
+            if (/FROM pages/.test(sql) && ++pagesQueries === 1) {
+              return Promise.reject(new Error('transient db hiccup'));
+            }
+            return target.executeRaw(sql, params);
+          };
+        }
+        return Reflect.get(target, prop, receiver);
+      },
+    });
+    const out: string[] = [];
+    await runWatch(flaky as never, ['--source', 'default'], {
+      lines: feed(['user: ping Alice Example', 'user: ping Alice Example please']),
+      write: (s) => out.push(s),
+      isTTY: false,
+    });
+    // Turn 1 failed open; turn 2 volunteered. The stream survived.
+    expect(out.join('')).toContain('people/alice-example');
+  });
+});

test/watch-sigint.serial.test.ts

@@ -33,15 +33,38 @@ describe('gbrain watch — SIGINT lifecycle (real subprocess)', () => {
       });
       proc.stdin.write('user: nothing relevant here\n');
       await proc.stdin.flush();
-      // Give the brain time to init + process the turn, then interrupt.
-      await new Promise((r) => setTimeout(r, 15_000));
+      // Readiness probe: watch prints "[watch] session <id> ready" on stderr
+      // once engine + source resolution are done and the stdin loop is live.
+      // A fixed sleep raced cold PGLite init (other tests budget 60s for it)
+      // — SIGINT before the handler registers means default-disposition kill.
+      const stderrChunks: string[] = [];
+      const reader = (proc.stderr as ReadableStream<Uint8Array>).getReader();
+      const decoder = new TextDecoder();
+      const deadline = Date.now() + 90_000;
+      let ready = false;
+      while (Date.now() < deadline) {
+        const { value, done } = await reader.read();
+        if (done) break;
+        stderrChunks.push(decoder.decode(value, { stream: true }));
+        if (stderrChunks.join('').includes('ready')) { ready = true; break; }
+      }
+      expect(ready).toBe(true);
+      // Brief settle so the first turn's volunteerContext round-trip is in
+      // flight or done, then interrupt mid-stream.
+      await new Promise((r) => setTimeout(r, 500));
       proc.kill('SIGINT');
       const killer = setTimeout(() => {
         try { proc.kill('SIGKILL'); } catch { /* already dead */ }
       }, 30_000);
       const exitCode = await proc.exited;
       clearTimeout(killer);
-      const stderr = await new Response(proc.stderr).text();
+      // Drain the rest of stderr for the banner assertion.
+      while (true) {
+        const { value, done } = await reader.read();
+        if (done) break;
+        stderrChunks.push(decoder.decode(value, { stream: true }));
+      }
+      const stderr = stderrChunks.join('');
       // Clean drain-then-exit: no force-exit banner, no SIGKILL (137), exit 0.
       expect(stderr).not.toContain('force-exiting');
       expect(exitCode).toBe(0);