From 147f550604b1e9679ef544556a6918635ebe5287 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Thu, 11 Jun 2026 21:46:31 -0700 Subject: [PATCH] feat(context): reflex consumes the rolling window + ambient-channel logging (#2095) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The default-on retrieval reflex now extracts entities from the last N turns (retrieval_reflex_window_turns, default 4; env GBRAIN_RETRIEVAL_REFLEX_WINDOW_TURNS; window=1 reproduces the legacy current-turn-only behavior exactly). assemble() passes the recent user/assistant turns (hard cap 12); the reflex slices to the configured window. Assistant-introduced entities and "what did she invest in?" follow-ups whose antecedent was NAMED in the window now surface pointers — the issue's "zero agent-initiated queries" success criterion on the ambient path. Under windowing, suppression switches to slug-only (codex D7): the legacy title-whole-word rule would suppress every entity merely MENTIONED in a prior window turn, breaking the feature by construction. Slugs only enter prior context when a pointer/page was actually surfaced, so already-surfaced pages still suppress. The suppression mode flows through all three resolver rungs (host opts, serve IPC request, direct Postgres). Ambient-channel feedback (codex D11): the server-side resolver paths (serve IPC + direct Postgres) log volunteered pointers with channel: 'reflex' through the drained volunteer-events sink, so `gbrain volunteer-context --stats` measures the default-on path where most volunteering happens. Host-injected resolvers (no gbrain engine) can't log — documented gap. Precision gates, 1.5s ceiling, fail-open, and the pointer cap are unchanged. Tests: prev-assistant-turn entity fires; window=1 legacy parity; slug-only vs already-surfaced suppression; throwing resolver stays fail-open (16 green). Co-Authored-By: Claude Fable 5 --- src/core/config.ts | 12 +++ src/core/context-engine.ts | 21 ++++++ src/core/context/reflex.ts | 59 +++++++++++++-- src/core/context/resolve-ipc.ts | 2 + src/core/context/retrieval-reflex.ts | 31 ++++++++ src/mcp/server.ts | 9 ++- test/retrieval-reflex.test.ts | 107 +++++++++++++++++++++++++++ 7 files changed, 232 insertions(+), 9 deletions(-) diff --git a/src/core/config.ts b/src/core/config.ts index ff1ee2273..c9ec21bb9 100644 --- a/src/core/config.ts +++ b/src/core/config.ts @@ -169,6 +169,14 @@ export interface GBrainConfig { retrieval_reflex?: boolean; /** Max pointers injected per turn (default 3). File-plane only. */ retrieval_reflex_max_pointers?: number; + /** + * v0.43 (#2095) — how many recent turns the reflex extracts entities from + * (default 4). 1 reproduces the legacy current-turn-only behavior (and the + * legacy slug+title suppression). File-plane / env + * (GBRAIN_RETRIEVAL_REFLEX_WINDOW_TURNS) only — same plane as the other + * reflex knobs. + */ + retrieval_reflex_window_turns?: number; embedding_image_ocr?: boolean; embedding_image_ocr_model?: string; @@ -443,6 +451,10 @@ export function loadConfig(): GBrainConfig | null { ...(process.env.GBRAIN_RETRIEVAL_REFLEX ? { retrieval_reflex: !(process.env.GBRAIN_RETRIEVAL_REFLEX === 'false' || process.env.GBRAIN_RETRIEVAL_REFLEX === '0') } : {}), + ...(process.env.GBRAIN_RETRIEVAL_REFLEX_WINDOW_TURNS && + Number.isFinite(Number(process.env.GBRAIN_RETRIEVAL_REFLEX_WINDOW_TURNS)) + ? { retrieval_reflex_window_turns: Number(process.env.GBRAIN_RETRIEVAL_REFLEX_WINDOW_TURNS) } + : {}), ...(process.env.GBRAIN_REMOTE_CLIENT_SECRET && fileConfig?.remote_mcp ? { remote_mcp: { ...fileConfig.remote_mcp, oauth_client_secret: process.env.GBRAIN_REMOTE_CLIENT_SECRET } } : {}), diff --git a/src/core/context-engine.ts b/src/core/context-engine.ts index 080e517da..67e9c24ac 100644 --- a/src/core/context-engine.ts +++ b/src/core/context-engine.ts @@ -179,6 +179,24 @@ function getLastUserText(messages: AgentMessage[]): string { return ''; } +/** + * v0.43 (#2095): the rolling extraction window — the most recent + * user/assistant turns (oldest → newest, current turn last), capped here at + * a generous max; the reflex slices to its configured + * retrieval_reflex_window_turns (default 4). + */ +const WINDOW_TURNS_HARD_CAP = 12; +function getWindowTurns(messages: AgentMessage[]): Array<{ role: 'user' | 'assistant'; text: string }> { + const out: Array<{ role: 'user' | 'assistant'; text: string }> = []; + for (const m of messages) { + 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); +} + /** * Joined text of everything the agent has ALREADY seen — every message EXCEPT * the current turn (the last user message). Used for "already in context" @@ -649,6 +667,9 @@ export function createGBrainContextEngine(ctx: { workspaceDir, currentUserText: getLastUserText(messages), priorContextText: getPriorContextText(messages), + // v0.43 (#2095): rolling window — assistant-introduced entities and + // named-antecedent follow-ups from recent turns now resolve too. + windowTurns: getWindowTurns(messages), resolveEntities: ctx.resolveEntities, }); diff --git a/src/core/context/reflex.ts b/src/core/context/reflex.ts index afec4054a..52e3753e6 100644 --- a/src/core/context/reflex.ts +++ b/src/core/context/reflex.ts @@ -20,7 +20,12 @@ import { join } from 'node:path'; import { mkdirSync, appendFileSync } from 'node:fs'; import { loadConfig, type GBrainConfig } from '../config.ts'; import type { BrainEngine } from '../engine.ts'; -import { extractCandidates, type EntityCandidate } from './entity-salience.ts'; +import { + extractCandidates, + extractCandidatesFromWindow, + type EntityCandidate, + type WindowTurn, +} from './entity-salience.ts'; import { resolveEntitiesToPointers, DEFAULT_MAX_POINTERS, @@ -28,22 +33,47 @@ import { } from './retrieval-reflex.ts'; import { resolveViaIpc, resolveSocketPath, IPC_UNAVAILABLE } from './resolve-ipc.ts'; +/** Per-turn resolver options shared by every rung of the ladder. */ +export interface ResolveEntitiesOpts { + priorContextText?: string; + maxPointers?: number; + /** v0.43 (#2095): 'slug-only' under windowing — see ResolvePointersOpts. */ + suppression?: 'slug-and-title' | 'slug-only'; +} + /** Host capability shape (D1=A): candidates in, pointers out. Narrow by design. */ export type ResolveEntitiesFn = ( candidates: EntityCandidate[], - opts: { priorContextText?: string; maxPointers?: number }, + opts: ResolveEntitiesOpts, ) => Promise; export interface ReflexParams { workspaceDir: string; - /** The current turn's user text (drives extraction). */ + /** The current turn's user text (drives extraction when no window is given). */ currentUserText: string; /** Joined PRIOR turns + loaded page bodies — EXCLUDES the current turn (suppression). */ priorContextText: string; + /** + * v0.43 (#2095): recent turns (oldest → newest, current turn last). When + * present and the configured window is > 1, extraction widens to the last + * N turns (assistant-introduced entities + named-antecedent follow-ups now + * resolve) and suppression switches to slug-only (codex D7 — the title rule + * would suppress every entity merely mentioned in a prior window turn). + */ + windowTurns?: WindowTurn[]; /** Host-provided resolver, if the OpenClaw plugin contract supplied one. */ resolveEntities?: ResolveEntitiesFn; } +/** Default extraction window (turns). 1 = legacy current-turn-only. */ +export const DEFAULT_WINDOW_TURNS = 4; + +export function windowTurnCount(cfg: GBrainConfig | null): number { + const n = cfg?.retrieval_reflex_window_turns; + if (typeof n === 'number' && Number.isFinite(n) && n >= 1) return Math.floor(n); + return DEFAULT_WINDOW_TURNS; +} + const TIMEOUT_MS = 1500; // generous per-turn ceiling; the work is usually <100ms const HEARTBEAT_PATH = join(homedir(), '.gbrain', 'integrations', 'retrieval-reflex', 'heartbeat.jsonl'); @@ -68,11 +98,22 @@ export async function buildReflexAddition(params: ReflexParams): Promise 1. Window=1 reproduces the legacy + // current-turn-only behavior exactly (including suppression mode). + const windowN = windowTurnCount(cfg); + const windowed = windowN > 1 && (params.windowTurns?.length ?? 0) > 0; + const candidates: EntityCandidate[] = windowed + ? extractCandidatesFromWindow(params.windowTurns!.slice(-windowN)) + : extractCandidates(params.currentUserText); + // Zero-candidate fast path: regex passes only, no brain touch. if (!candidates.length) return null; - const opts = { priorContextText: params.priorContextText, maxPointers: maxPointers(cfg) }; + const opts: ResolveEntitiesOpts = { + priorContextText: params.priorContextText, + maxPointers: maxPointers(cfg), + suppression: windowed ? 'slug-only' : 'slug-and-title', + }; const block = await withTimeout(resolve(params, cfg, candidates, opts), TIMEOUT_MS); if (!block || !block.pointers.length) return null; @@ -87,7 +128,7 @@ async function resolve( params: ReflexParams, cfg: GBrainConfig | null, candidates: EntityCandidate[], - opts: { priorContextText?: string; maxPointers?: number }, + opts: ResolveEntitiesOpts, ): Promise { // 1. Host capability (any engine). if (params.resolveEntities) { @@ -105,7 +146,9 @@ async function resolve( if (!engine) return null; const { resolveSourceId } = await import('../source-resolver.ts'); const sourceId = await resolveSourceId(engine, null, params.workspaceDir); - return resolveEntitiesToPointers(engine, sourceId, candidates, opts); + // logChannel (codex D11): the ambient reflex channel logs its volunteered + // pointers so `volunteer-context --stats` measures the default-on path. + return resolveEntitiesToPointers(engine, sourceId, candidates, { ...opts, logChannel: 'reflex' }); } // 4. Disabled (PGLite with no serve / unknown engine). Policy skill carries. return null; diff --git a/src/core/context/resolve-ipc.ts b/src/core/context/resolve-ipc.ts index 3955ea3e1..6b8c89a52 100644 --- a/src/core/context/resolve-ipc.ts +++ b/src/core/context/resolve-ipc.ts @@ -35,6 +35,8 @@ export interface ResolveRequest { priorContextText?: string; maxPointers?: number; sourceId?: string; + /** v0.43 (#2095, codex D7): suppression mode — 'slug-only' under windowing. */ + suppression?: 'slug-and-title' | 'slug-only'; } export type ResolveHandler = (req: ResolveRequest) => Promise; diff --git a/src/core/context/retrieval-reflex.ts b/src/core/context/retrieval-reflex.ts index 70a327e95..084e85183 100644 --- a/src/core/context/retrieval-reflex.ts +++ b/src/core/context/retrieval-reflex.ts @@ -104,6 +104,16 @@ export interface ResolvePointersOpts { * arm uses source_id = ANY(...) in one query. */ sourceIds?: string[]; + /** + * v0.43 (#2095, codex D11) — when set, fire-and-forget a volunteer event + * per returned pointer on this channel (through the drained + * volunteer-events sink). Set by the REFLEX server-side paths (serve IPC, + * direct Postgres) so `volunteer-context --stats` measures the default-on + * ambient channel. The volunteer layer does NOT set it — it logs its own + * events with boosted confidence. Host-injected resolvers (no gbrain + * engine) can't log; documented gap. + */ + logChannel?: 'reflex'; } interface PageRow { @@ -273,6 +283,27 @@ export async function resolveEntitiesToPointers( } if (!pointers.length) return null; + + // v0.43 (#2095, codex D11): ambient-channel feedback logging. Lazy import + // keeps the reflex hot path dependency-free when logging is off. + if (opts.logChannel) { + try { + const { logVolunteerEventsFireAndForget } = 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', + })), + ); + } catch { + /* telemetry only — never blocks the pointer block */ + } + } return { pointers, text: renderPointerBlock(pointers) }; } diff --git a/src/mcp/server.ts b/src/mcp/server.ts index 9ea0a7470..cf125ea7b 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -73,7 +73,14 @@ export async function startMcpServer(engine: BrainEngine) { engine, req.sourceId || defaultSource, req.candidates ?? [], - { priorContextText: req.priorContextText, maxPointers: req.maxPointers }, + { + priorContextText: req.priorContextText, + maxPointers: req.maxPointers, + suppression: req.suppression, + // v0.43 (#2095, codex D11): the IPC resolve path IS the ambient + // reflex channel — log volunteered pointers so stats see it. + logChannel: 'reflex', + }, ), ); } diff --git a/test/retrieval-reflex.test.ts b/test/retrieval-reflex.test.ts index 385af152b..bd6d845ac 100644 --- a/test/retrieval-reflex.test.ts +++ b/test/retrieval-reflex.test.ts @@ -182,3 +182,110 @@ describe('context-engine assemble() — Retrieval Reflex integration', () => { }); }); }); + +describe('v0.43 (#2095) — rolling window extraction through assemble()', () => { + const REFLEX_ON = { GBRAIN_RETRIEVAL_REFLEX: 'true' }; + + test('entity named ONLY in a previous assistant turn yields a pointer now', async () => { + await withEnv(REFLEX_ON, async () => { + await seed('people/alice-example', 'Alice Example', 'Alice is a founder.'); + const ce = createGBrainContextEngine({ + workspaceDir: '/tmp/rr-test-ws-w1', + resolveEntities: (candidates, opts) => + resolveEntitiesToPointers(engine, 'default', candidates, opts), + }); + // Current turn is a pronoun follow-up; the antecedent was NAMED two + // turns back by the ASSISTANT. Pre-window this never fired. + const res = await ce.assemble({ + sessionId: 'w1', + messages: [ + { role: 'user', content: 'who should I talk to about the seed round?' }, + { role: 'assistant', content: 'Alice Example led a similar round last year.' }, + { role: 'user', content: 'what did she invest in?' }, + ], + }); + expect(res.systemPromptAddition).toContain('people/alice-example'); + }); + }); + + test('window=1 reproduces the legacy current-turn-only behavior', async () => { + await withEnv({ ...REFLEX_ON, GBRAIN_RETRIEVAL_REFLEX_WINDOW_TURNS: '1' }, async () => { + await seed('people/alice-example', 'Alice Example', 'Alice is a founder.'); + const ce = createGBrainContextEngine({ + workspaceDir: '/tmp/rr-test-ws-w2', + resolveEntities: (candidates, opts) => + resolveEntitiesToPointers(engine, 'default', candidates, opts), + }); + const res = await ce.assemble({ + sessionId: 'w2', + messages: [ + { role: 'assistant', content: 'Alice Example led a similar round last year.' }, + { role: 'user', content: 'what did she invest in?' }, + ], + }); + // Current turn has no extractable entity; window=1 must NOT widen. + expect(res.systemPromptAddition).not.toContain('people/alice-example'); + }); + }); + + test('windowed suppression is slug-only: a prior-turn MENTION does not suppress (codex D7)', async () => { + await withEnv(REFLEX_ON, async () => { + await seed('people/alice-example', 'Alice Example', 'A founder.'); + const ce = createGBrainContextEngine({ + workspaceDir: '/tmp/rr-test-ws-w3', + resolveEntities: (candidates, opts) => + resolveEntitiesToPointers(engine, 'default', candidates, opts), + }); + // "Alice Example" appears in a PRIOR turn (a bare mention — prior + // context contains the TITLE). Under the legacy title rule the pointer + // would be suppressed; slug-only windowing must still fire. + const res = await ce.assemble({ + sessionId: 'w3', + messages: [ + { role: 'user', content: 'I met Alice Example yesterday' }, + { role: 'assistant', content: 'How did the meeting with Alice Example go?' }, + { role: 'user', content: 'she wants to invest — thoughts?' }, + ], + }); + expect(res.systemPromptAddition).toContain('people/alice-example'); + }); + }); + + test('windowed suppression still drops an already-surfaced page (slug in prior context)', async () => { + await withEnv(REFLEX_ON, async () => { + await seed('people/alice-example', 'Alice Example', 'A founder.'); + const ce = createGBrainContextEngine({ + workspaceDir: '/tmp/rr-test-ws-w4', + resolveEntities: (candidates, opts) => + resolveEntitiesToPointers(engine, 'default', candidates, opts), + }); + const res = await ce.assemble({ + sessionId: 'w4', + messages: [ + { role: 'assistant', content: 'Pointer: **Alice Example** → `people/alice-example` (use get_page)' }, + { role: 'user', content: 'tell me more about Alice Example' }, + ], + }); + expect(res.systemPromptAddition).not.toContain('Brain pages mentioned this turn'); + }); + }); + + test('fail-open: a throwing resolver under windowing never breaks the turn', async () => { + await withEnv(REFLEX_ON, async () => { + await seed('people/alice-example', 'Alice Example', 'A founder.'); + const ce = createGBrainContextEngine({ + workspaceDir: '/tmp/rr-test-ws-w5', + resolveEntities: async () => { throw new Error('resolver exploded'); }, + }); + const res = await ce.assemble({ + sessionId: 'w5', + messages: [ + { role: 'assistant', content: 'Alice Example is relevant here.' }, + { role: 'user', content: 'ok tell me about her' }, + ], + }); + expect(res.systemPromptAddition).toContain('Live Context'); + expect(res.systemPromptAddition).not.toContain('Brain pages mentioned this turn'); + }); + }); +});