diff --git a/src/cli.ts b/src/cli.ts index bf547d46c..b67082108 100755 --- a/src/cli.ts +++ b/src/cli.ts @@ -788,6 +788,27 @@ async function makeContext(engine: BrainEngine, params: Record) function formatResult(opName: string, result: unknown): string { switch (opName) { + case 'volunteer_context': { + const r = result as any; + // Stats mode (the feedback loop). + if (r && r.approximate === true && Array.isArray(r.by_arm)) { + const lines = [ + `volunteered-context precision — last ${r.days} day(s) (${r.note})`, + `total: ${r.total_volunteered} volunteered, ${r.total_used} used`, + ]; + for (const a of r.by_arm) { + lines.push(` ${a.match_arm}/${a.channel}: ${a.used}/${a.volunteered} used (precision ${a.precision})`); + } + if (!r.by_arm.length) lines.push(' (no volunteer events in the window)'); + return lines.join('\n') + '\n'; + } + 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'; + } case 'get_page': { const r = result as any; if (r.error === 'ambiguous_slug') { diff --git a/src/core/context/volunteer-events.ts b/src/core/context/volunteer-events.ts index 09f1f4e8e..5783a8961 100644 --- a/src/core/context/volunteer-events.ts +++ b/src/core/context/volunteer-events.ts @@ -18,6 +18,7 @@ */ import type { BrainEngine } from './../engine.ts'; +import { registerBackgroundWorkDrainer } from '../background-work.ts'; export const VOLUNTEER_EVENTS_TTL_DAYS = 90; @@ -70,6 +71,73 @@ export async function insertVolunteerEvents( ); } +// ── Fire-and-forget sink (eng-review D4) ───────────────────────────────── +// Mirrors src/core/last-retrieved.ts: track every dangling INSERT promise in +// a module Set, register a drainer so drainThenDisconnect settles them +// against a live engine before teardown on EVERY CLI exit path (the commit-1 +// drain hoist). Logging failure never fails the caller. + +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, + * transient DB failures — the volunteer result is unaffected). + */ +export function logVolunteerEventsFireAndForget( + engine: BrainEngine, + rows: VolunteerEventRow[], +): void { + if (!rows.length) return; + const p = insertVolunteerEvents(engine, rows).catch(() => { + /* best-effort telemetry — never surfaces */ + }); + pendingVolunteerEventWrites.add(p); + void p.finally(() => pendingVolunteerEventWrites.delete(p)); +} + +/** Drain pending event writes (bounded). Same snapshot semantics as last-retrieved. */ +export async function awaitPendingVolunteerEventWrites( + timeoutMs = 5_000, +): Promise<{ unfinished: number }> { + if (pendingVolunteerEventWrites.size === 0) return { unfinished: 0 }; + const snapshot = Array.from(pendingVolunteerEventWrites); + let timer: ReturnType | undefined; + const timeout = new Promise<'timeout'>((resolve) => { + timer = setTimeout(() => resolve('timeout'), timeoutMs); + }); + const drain = Promise.allSettled(snapshot).then(() => 'drained' as const); + const outcome = await Promise.race([drain, timeout]); + if (timer) clearTimeout(timer); + if (outcome === 'timeout') { + const unfinished = pendingVolunteerEventWrites.size; + // Drop the snapshot so a long-lived process (`gbrain watch`) doesn't + // accumulate references to forever-pending work (last-retrieved C1). + for (const p of snapshot) pendingVolunteerEventWrites.delete(p); + return { unfinished }; + } + return { unfinished: 0 }; +} + +// Registered in the enqueue-owning module (background-work contract): module +// not imported ⇒ nothing enqueued ⇒ nothing to drain. Order 4 — after facts / +// last-retrieved / search-cache / eval-capture; bare INSERTs, no abort. +registerBackgroundWorkDrainer({ + name: 'volunteer-events', + order: 4, + drain: (ms) => awaitPendingVolunteerEventWrites(ms), +}); + +/** Test seam — clears the pending set so each test starts clean. */ +export function _resetPendingVolunteerEventWritesForTests(): void { + pendingVolunteerEventWrites.clear(); +} + +/** Test seam — peek the current pending count. */ +export function _peekPendingVolunteerEventWritesForTests(): number { + return pendingVolunteerEventWrites.size; +} + /** * 90-day GC, called from the dream cycle's purge phase (mirrors * purgeStaleCheckpoints). Best-effort: returns 0 on any failure (pre-v116 diff --git a/src/core/operations.ts b/src/core/operations.ts index aa6f41e41..2a0c68dc0 100644 --- a/src/core/operations.ts +++ b/src/core/operations.ts @@ -3145,6 +3145,99 @@ const get_recent_salience: Operation = { cliHints: { name: 'salience' }, }; +// --- v0.43 (#2095): push-based context — the brain volunteers pages --- + +const volunteer_context: Operation = { + name: 'volunteer_context', + description: + 'Push-based context: volunteer brain pages relevant to a rolling conversation window ' + + 'WITHOUT being asked. Zero-LLM, confidence-gated (alias 0.9 / exact-title 0.8 / ' + + 'slug-suffix 0.6, +0.05 for multi-turn or newest-turn mentions; default gate 0.7), ' + + 'capped at 3 pages (max 5). Returns pointers with one-line rationales + synopses — ' + + 'open the page (get_page) before relying on details. Pass stats: true for the ' + + 'approximate volunteered-vs-used precision summary (the feedback loop).', + scope: 'read', + params: { + window: { + type: 'string', + description: + "Recent conversation turns, oldest → newest, as 'user:' / 'assistant:' prefixed " + + 'lines (unprefixed text = one user turn). Required unless stats: true. ' + + 'CLI: piped stdin fills this.', + }, + prior_context: { + type: 'string', + description: + 'Already-surfaced context (pointer blocks / opened page bodies). Pages whose slug ' + + 'appears here are not re-volunteered.', + }, + max_pages: { type: 'number', description: 'Max pages to volunteer (default 3, hard cap 5).' }, + min_confidence: { + type: 'number', + description: + 'Confidence gate 0..1 (default 0.7 — slug-suffix matches need an explicit lower gate).', + }, + session_id: { type: 'string', description: 'Optional caller session id, logged for attribution.' }, + turn: { type: 'number', description: 'Optional caller turn number, logged for attribution.' }, + stats: { + type: 'boolean', + description: + 'Return the volunteered-vs-used precision summary instead of volunteering. ' + + 'APPROXIMATE: "used" = pages.last_retrieved_at > volunteered_at.', + }, + days: { type: 'number', description: 'Stats window in days (default 30; stats mode only).' }, + }, + handler: async (ctx, p) => { + const { parseWindow, volunteerContext, volunteerUsageStats } = await import('./context/volunteer.ts'); + const scope = sourceScopeOpts(ctx); + const sourceIds = scope.sourceIds ?? (scope.sourceId ? [scope.sourceId] : ['default']); + + if (p.stats === true) { + return volunteerUsageStats(ctx.engine, sourceIds, typeof p.days === 'number' ? p.days : undefined); + } + + if (typeof p.window !== 'string' || !p.window.trim()) { + throw new OperationError( + 'invalid_params', + 'window is required unless stats: true', + 'Pass the recent turns as a string (CLI: pipe them on stdin), or use --stats.', + ); + } + const turns = parseWindow(p.window); + const pages = await volunteerContext(ctx.engine, turns, { + sourceIds, + priorContext: typeof p.prior_context === 'string' ? p.prior_context : undefined, + maxPages: typeof p.max_pages === 'number' ? p.max_pages : undefined, + minConfidence: typeof p.min_confidence === 'number' ? p.min_confidence : undefined, + }); + + // Feedback-loop logging: fire-and-forget batched INSERT through the + // volunteer-events sink (drained at exit). Never fails the op. + if (pages.length) { + try { + const { logVolunteerEventsFireAndForget } = await import('./context/volunteer-events.ts'); + 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, + })), + ); + } catch { + /* telemetry only */ + } + } + return { pages, count: pages.length, window_turns: turns.length }; + }, + cliHints: { name: 'volunteer-context', stdin: 'window' }, +}; + const find_anomalies: Operation = { name: 'find_anomalies', description: FIND_ANOMALIES_DESCRIPTION, @@ -4856,6 +4949,8 @@ export const operations: Operation[] = [ whoami, sources_add, sources_list, sources_remove, sources_status, // v0.29: Salience + anomalies + recent transcripts get_recent_salience, find_anomalies, get_recent_transcripts, + // v0.43 (#2095): push-based context + volunteer_context, // v0.31: hot memory (facts table) extract_facts, recall, forget_fact, // v0.32.6: contradiction probe MCP surface (M3) diff --git a/test/e2e/volunteer-context-postgres.test.ts b/test/e2e/volunteer-context-postgres.test.ts new file mode 100644 index 000000000..9f8505f2f --- /dev/null +++ b/test/e2e/volunteer-context-postgres.test.ts @@ -0,0 +1,97 @@ +/** + * v0.43 (#2095) — volunteer_context on REAL Postgres (engine parity beyond + * PGLite) + the RLS pin for the v116 table. + * + * The unit suite covers the volunteer core hermetically on PGLite; this file + * proves the same op handler against pgvector Postgres: resolution arms, + * the fire-and-forget event sink landing rows, the stats join, and that + * context_volunteer_events has ROW LEVEL SECURITY enabled (the v35 + * auto_rls_on_create_table event trigger covers migration-created tables — + * this is the assertion that keeps that mechanism honest for new tables). + * + * Gated by DATABASE_URL — skips gracefully without real Postgres. + */ + +import { describe, test, expect, beforeAll, afterAll } from 'bun:test'; +import { hasDatabase, setupDB, teardownDB, getEngine } from './helpers.ts'; +import { operationsByName } from '../../src/core/operations.ts'; +import { + awaitPendingVolunteerEventWrites, + _resetPendingVolunteerEventWritesForTests, +} from '../../src/core/context/volunteer-events.ts'; + +const SKIP = !hasDatabase(); +const describePg = SKIP ? describe.skip : describe; + +function mkCtx(engine: unknown, overrides: Record = {}) { + return { + engine, + config: {} as never, + logger: { info: () => {}, warn: () => {}, error: () => {} } as never, + dryRun: false, + remote: false, + sourceId: 'default', + ...overrides, + } as never; +} + +describePg('volunteer_context on real Postgres (#2095)', () => { + beforeAll(async () => { + await setupDB(); + const engine = getEngine(); + await engine.putPage('people/alice-example', { + type: 'person', + title: 'Alice Example', + compiled_truth: 'Alice is an early founder.', + timeline: '', + }); + }, 240_000); + + afterAll(async () => { + await teardownDB(); + }); + + test('op volunteers, logs through the sink, and stats join works', async () => { + _resetPendingVolunteerEventWritesForTests(); + const engine = getEngine(); + const op = operationsByName.volunteer_context; + + const result = (await op.handler(mkCtx(engine), { + window: 'user: who knows the seed market?\nassistant: Alice Example does.\nuser: ask her', + session_id: 'e2e-pg', + })) as any; + expect(result.count).toBe(1); + expect(result.pages[0].slug).toBe('people/alice-example'); + expect(result.pages[0].arm).toBe('title'); + + const { unfinished } = await awaitPendingVolunteerEventWrites(10_000); + expect(unfinished).toBe(0); + + const rows = await engine.executeRaw<{ slug: string; session_id: string }>( + `SELECT slug, session_id FROM context_volunteer_events WHERE session_id = 'e2e-pg'`, + [], + ); + expect(rows.length).toBe(1); + + // Mark the page as opened after volunteering → stats counts it used. + await engine.executeRaw( + `UPDATE pages SET last_retrieved_at = now() + interval '1 minute' WHERE slug = 'people/alice-example'`, + [], + ); + const stats = (await op.handler(mkCtx(engine), { stats: true })) as any; + expect(stats.approximate).toBe(true); + expect(stats.total_volunteered).toBeGreaterThanOrEqual(1); + expect(stats.total_used).toBeGreaterThanOrEqual(1); + }, 120_000); + + test('RLS is enabled on context_volunteer_events (auto-RLS covers v116)', async () => { + const engine = getEngine(); + const rows = await engine.executeRaw<{ relrowsecurity: boolean }>( + `SELECT relrowsecurity FROM pg_class + WHERE oid = 'public.context_volunteer_events'::regclass`, + [], + ); + expect(rows.length).toBe(1); + expect(rows[0].relrowsecurity).toBe(true); + }, 60_000); +}); diff --git a/test/volunteer-context.test.ts b/test/volunteer-context.test.ts index 28905f3d0..7e7fb9848 100644 --- a/test/volunteer-context.test.ts +++ b/test/volunteer-context.test.ts @@ -281,3 +281,115 @@ describe('resolveEntitiesToPointers — new provenance surface (back-compat)', ( expect(block).toBeNull(); }); }); + +describe('volunteer_context op (contract surface)', () => { + const { operationsByName } = require('../src/core/operations.ts') as typeof import('../src/core/operations.ts'); + const { + awaitPendingVolunteerEventWrites, + _resetPendingVolunteerEventWritesForTests, + } = require('../src/core/context/volunteer-events.ts') as typeof import('../src/core/context/volunteer-events.ts'); + + function mkCtx(overrides: Record = {}) { + return { + engine, + config: {} as never, + logger: { info: () => {}, warn: () => {}, error: () => {} } as never, + dryRun: false, + remote: false, + sourceId: 'default', + ...overrides, + } as never; + } + + test('registered, read-scope, not localOnly, stdin cliHint on window', () => { + const op = operationsByName.volunteer_context; + expect(op).toBeDefined(); + expect(op.scope).toBe('read'); + expect(op.localOnly).toBeFalsy(); + expect(op.cliHints?.name).toBe('volunteer-context'); + expect(op.cliHints?.stdin).toBe('window'); + }); + + test('window required unless stats: true', async () => { + const op = operationsByName.volunteer_context; + await expect(op.handler(mkCtx(), {})).rejects.toThrow(/window is required/); + const stats = (await op.handler(mkCtx(), { stats: true })) as any; + expect(stats.approximate).toBe(true); + }); + + test('volunteers pages and logs events through the drained sink', async () => { + _resetPendingVolunteerEventWritesForTests(); + await seed('people/alice-example', 'Alice Example', 'Founder.'); + const op = operationsByName.volunteer_context; + const result = (await op.handler(mkCtx(), { + window: 'user: ping Alice Example about the deal', + session_id: 's-42', + turn: 7, + })) as any; + expect(result.count).toBe(1); + expect(result.pages[0].slug).toBe('people/alice-example'); + expect(result.window_turns).toBe(1); + // Event row lands via the fire-and-forget sink once drained. + const { unfinished } = await awaitPendingVolunteerEventWrites(5_000); + expect(unfinished).toBe(0); + const rows = await engine.executeRaw<{ slug: string; channel: string; session_id: string; turn: number }>( + `SELECT slug, channel, session_id, turn FROM context_volunteer_events`, [], + ); + expect(rows.length).toBe(1); + expect(rows[0].slug).toBe('people/alice-example'); + expect(rows[0].channel).toBe('op'); + expect(rows[0].session_id).toBe('s-42'); + expect(Number(rows[0].turn)).toBe(7); + }); + + test('event-log failure never fails the op (failing engine injected for the INSERT)', async () => { + _resetPendingVolunteerEventWritesForTests(); + await seed('people/alice-example', 'Alice Example', 'Founder.'); + const op = operationsByName.volunteer_context; + // Wrap the engine: reads pass through, INSERTs into the events table throw. + const failingEngine = new Proxy(engine, { + get(target, prop, receiver) { + if (prop === 'executeRaw') { + return (sql: string, params: unknown[]) => { + if (/INSERT INTO context_volunteer_events/.test(sql)) { + return Promise.reject(new Error('telemetry db down')); + } + return target.executeRaw(sql, params); + }; + } + return Reflect.get(target, prop, receiver); + }, + }); + const result = (await op.handler(mkCtx({ engine: failingEngine }), { + window: 'user: ping Alice Example', + })) as any; + expect(result.count).toBe(1); // the volunteer result is unaffected + const { unfinished } = await awaitPendingVolunteerEventWrites(5_000); + expect(unfinished).toBe(0); // failed write settled (swallowed), not stuck + }); + + test('federated grant scopes the volunteer (allowedSources)', async () => { + await seed('people/alice-example', 'Alice Example', 'Founder.', 'default'); + await seed('people/bob-sample', 'Bob Sample', 'Engineer.', 'grant-brain'); + await seed('people/eve-other', 'Eve Other', 'Out of grant.', 'secret-brain'); + const op = operationsByName.volunteer_context; + const result = (await op.handler( + mkCtx({ remote: true, auth: { allowedSources: ['grant-brain'] } }), + { window: 'user: intro Alice Example to Bob Sample and Eve Other' }, + )) as any; + expect(result.pages.map((p: any) => p.slug)).toEqual(['people/bob-sample']); + }); + + test('stats mode is source-scoped and returns the approximate note', async () => { + await seed('people/alice-example', 'Alice Example', 'Founder.'); + await insertVolunteerEvents(engine, [ + { source_id: 'default', slug: 'people/alice-example', confidence: 0.9, match_arm: 'alias', rationale: 'r', channel: 'watch' }, + ]); + const op = operationsByName.volunteer_context; + const stats = (await op.handler(mkCtx(), { stats: true, days: 7 })) as any; + expect(stats.days).toBe(7); + expect(stats.total_volunteered).toBe(1); + expect(stats.by_arm[0].channel).toBe('watch'); + expect(stats.note).toContain('approximate'); + }); +});