From d8a4d8a54a407d3b05c5e90b05ed7a5cf18950b5 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Fri, 12 Jun 2026 10:53:27 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20pre-landing=20review=20hardening=20?= =?UTF-8?q?=E2=80=94=20federated=20alias=20parallelism,=20trust-boundary?= =?UTF-8?q?=20clamps,=20shared=20protocol=20helpers=20(#2095)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- TODOS.md | 18 +++++++ src/cli.ts | 6 +-- src/commands/watch.ts | 49 ++++++++++--------- src/core/context-engine.ts | 9 +++- src/core/context/retrieval-reflex.ts | 60 +++++++++++------------ src/core/context/volunteer-events.ts | 30 ++++++++++-- src/core/context/volunteer.ts | 20 ++++++-- src/core/cycle.ts | 2 +- src/core/operations.ts | 22 ++++----- test/cli-exit-verdict-pin.test.ts | 5 +- test/cli-pipe-truncation.test.ts | 9 ++-- test/fix-wave-structural.test.ts | 17 +++++++ test/retrieval-reflex.test.ts | 67 +++++++++++++++++++++++++ test/volunteer-context.test.ts | 73 ++++++++++++++++++++++++++++ test/watch-command.test.ts | 30 ++++++++++++ test/watch-sigint.serial.test.ts | 29 +++++++++-- 16 files changed, 358 insertions(+), 88 deletions(-) diff --git a/TODOS.md b/TODOS.md index 75a4098a6..0fd9bca7c 100644 --- a/TODOS.md +++ b/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). diff --git a/src/cli.ts b/src/cli.ts index ac174a849..4cab59fbd 100755 --- a/src/cli.ts +++ b/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; diff --git a/src/commands/watch.ts b/src/commands/watch.ts index 0d018892d..b051bedce 100644 --- a/src/commands/watch.ts +++ b/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'); } } } diff --git a/src/core/context-engine.ts b/src/core/context-engine.ts index 67e9c24ac..e38114b98 100644 --- a/src/core/context-engine.ts +++ b/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(); } /** diff --git a/src/core/context/retrieval-reflex.ts b/src/core/context/retrieval-reflex.ts index 084e85183..c721d9a75 100644 --- a/src/core/context/retrieval-reflex.ts +++ b/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(); + const slugToNorm = new Map(); 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(); - const slugToNorm = new Map(); - 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 */ diff --git a/src/core/context/volunteer-events.ts b/src/core/context/volunteer-events.ts index 081c0e0f7..2ec14a7b6 100644 --- a/src/core/context/volunteer-events.ts +++ b/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>(); /** * 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( diff --git a/src/core/context/volunteer.ts b/src/core/context/volunteer.ts index 59fa98b0f..e29f1eddd 100644 --- a/src/core/context/volunteer.ts +++ b/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); diff --git a/src/core/cycle.ts b/src/core/cycle.ts index 353faef18..17a31bccb 100644 --- a/src/core/cycle.ts +++ b/src/core/cycle.ts @@ -1287,7 +1287,7 @@ async function runPhasePurge(engine: BrainEngine, dryRun: boolean): Promise ({ - 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 */ diff --git a/test/cli-exit-verdict-pin.test.ts b/test/cli-exit-verdict-pin.test.ts index f94b04923..75322ca47 100644 --- a/test/cli-exit-verdict-pin.test.ts +++ b/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(''); diff --git a/test/cli-pipe-truncation.test.ts b/test/cli-pipe-truncation.test.ts index 05e85e97f..bea683ef7 100644 --- a/test/cli-pipe-truncation.test.ts +++ b/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); diff --git a/test/fix-wave-structural.test.ts b/test/fix-wave-structural.test.ts index e2b668e70..1b734617c 100644 --- a/test/fix-wave-structural.test.ts +++ b/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/); + }); +}); diff --git a/test/retrieval-reflex.test.ts b/test/retrieval-reflex.test.ts index 11657b7a6..78ebfc6a4 100644 --- a/test/retrieval-reflex.test.ts +++ b/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); + }); +}); diff --git a/test/volunteer-context.test.ts b/test/volunteer-context.test.ts index 7e7fb9848..06c88b7ab 100644 --- a/test/volunteer-context.test.ts +++ b/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(); + }); +}); diff --git a/test/watch-command.test.ts b/test/watch-command.test.ts index cc33db249..4a643a1fc 100644 --- a/test/watch-command.test.ts +++ b/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'); + }); +}); diff --git a/test/watch-sigint.serial.test.ts b/test/watch-sigint.serial.test.ts index 1197bae0d..62edfe713 100644 --- a/test/watch-sigint.serial.test.ts +++ b/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 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).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);