feat(ops): volunteer_context op — CLI (stdin) + MCP, drained event sink (#2095)

New read-scope op on the contract surface (CLI `gbrain volunteer-context`
with stdin → window, MCP tool for free): takes a rolling conversation
window, returns confidence-gated page pointers with rationales + synopses.
`window` is optional-unless-stats (validated in the handler, codex D9);
`stats: true` returns the volunteered-vs-used precision summary, labeled
APPROXIMATE (the 5-min last-retrieved throttle and unrelated reads both
bias the join). Source scope threads through sourceScopeOpts — federated
grants narrow the volunteer to the granted sources.

Event logging is fire-and-forget through a new `volunteer-events`
background-work sink (volunteer-events.ts, mirrors last-retrieved: tracked
dangling promise set + bounded drain + snapshot-drop on timeout so a
long-lived process never accumulates ghosts). ONE batched INSERT per call,
drained on every exit path by the commit-1 drain hoist; failure never
fails the op (pinned by an injected failing-engine test).

cli formatResult renders both shapes (pointer lines with confidence/arm/
rationale; the stats summary with per-arm precision).

Tests: op contract surface, window-required validation, sink round-trip
with session_id/turn attribution, failing-engine fail-open, federated
grant scoping, stats mode (26 green on PGLite) + a real-Postgres e2e
proving the op + sink + stats join AND that context_volunteer_events has
RLS enabled (keeps the auto-RLS event-trigger mechanism honest for v116).

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

src/cli.ts

@@ -788,6 +788,27 @@ async function makeContext(engine: BrainEngine, params: Record<string, unknown>)
 
 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') {

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<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,
+ * 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<typeof setTimeout> | 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

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)

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

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