From 8f7e4f5437f1c2d9f80cb71f330542f7a5c9807e Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Sat, 4 Jul 2026 15:14:10 -0600 Subject: [PATCH 1/4] feat(memory): optional randomized retrieval holdout (epsilon-dropout) for treatment-effect logging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Opt-in, default-off hook between adapter.search() and renderMemoryContext(): with per-session probability epsilon, suppress one watchlist item from every retrieval of that session (sticky, no backfill) and emit one counterfactual-slot event per call — including no-drop calls — with the full eligibility set and exact pick/drop propensities. No I/O in the library; persistence is the consumer's job via onEvent. Unconfigured behavior is byte-identical (tested). Design, estimator, and sample-size analysis: research repo, projects/probabilistic-agent-optimization/notes/2026-07-03-DRAFT-o3-holdout-design.md (O3 / EXP-007). --- src/memory/adapter.ts | 17 ++- src/memory/holdout.ts | 237 +++++++++++++++++++++++++++++++ src/memory/index.ts | 1 + src/memory/types.ts | 6 + tests/memory-holdout.test.ts | 266 +++++++++++++++++++++++++++++++++++ 5 files changed, 524 insertions(+), 3 deletions(-) create mode 100644 src/memory/holdout.ts create mode 100644 tests/memory-holdout.test.ts diff --git a/src/memory/adapter.ts b/src/memory/adapter.ts index 1a9017a..f29e462 100644 --- a/src/memory/adapter.ts +++ b/src/memory/adapter.ts @@ -1,3 +1,4 @@ +import { applySessionStickyRetrievalHoldout } from './holdout' import { memoryHitToSourceRecord } from './source-record' import type { AgentMemoryAdapter, @@ -12,11 +13,21 @@ export async function defaultGetMemoryContext( options: AgentMemorySearchOptions = {}, ): Promise { const hits = await adapter.search(query, options) + // The holdout applies between search (eligibility set) and render (delivered set), and only + // when explicitly configured: the unconfigured path must stay byte-identical, including + // hit-array identity, so default behavior cannot drift. + const delivered = options.holdout + ? applySessionStickyRetrievalHoldout(hits, options.holdout, { + query, + ...(options.scope !== undefined ? { scope: options.scope } : {}), + ...(options.scope?.tags?.taskId !== undefined ? { taskId: options.scope.tags.taskId } : {}), + }).delivered + : hits return { query, - hits, - sourceRecords: hits.map((hit) => memoryHitToSourceRecord(hit, { scope: options.scope })), - text: renderMemoryContext(hits), + hits: delivered, + sourceRecords: delivered.map((hit) => memoryHitToSourceRecord(hit, { scope: options.scope })), + text: renderMemoryContext(delivered), } } diff --git a/src/memory/holdout.ts b/src/memory/holdout.ts new file mode 100644 index 0000000..c84a0c8 --- /dev/null +++ b/src/memory/holdout.ts @@ -0,0 +1,237 @@ +import { randomUUID } from 'node:crypto' +import { sha256 } from '../ids' +import type { AgentMemoryHit, AgentMemoryScope } from './types' + +// Randomized retrieval holdout (epsilon-dropout) for per-item treatment-effect logging. +// Default-off: nothing in this module runs unless a consumer passes a RetrievalHoldoutConfig. +// The library never does I/O here; persistence is the consumer's job via onEvent. +// Design + estimator + sample-size analysis: research repo, +// projects/probabilistic-agent-optimization/notes/2026-07-03-DRAFT-o3-holdout-design.md (O3 / EXP-007). + +export interface RetrievalHoldoutConfig { + /** Per-session probability that one eligible watchlist item is suppressed. 0 logs the full schema without ever dropping. */ + epsilon: number + /** Item ids eligible for suppression. Empty or absent means no item can ever be dropped. */ + watchlist?: string[] + /** Ties every event to the exact epsilon/watchlist in force, for audit and replay. */ + configVersion?: string + /** Copied onto every event so multi-adapter logs stay attributable. */ + adapterId?: string + /** Corpus/store version stamp; an edited item under the same id is a different treatment. */ + corpusVersion?: string + /** + * Uniform-[0,1) generator keyed by a string. Defaults to a sha256-derived deterministic + * generator so every assignment is replayable from the logged keys alone (design rule D5). + */ + rng?: (key: string) => number + /** Receives one event per retrieval call, INCLUDING no-drop calls: control-arm membership is half the data. */ + onEvent: (event: RetrievalHoldoutEvent) => void +} + +export interface RetrievalHoldoutEligibleItem { + id: string + /** 1-based position in the post-filter hit list. */ + rank: number + score?: number + kind: string + /** sha256(hit.text) prefix; effects are estimated per (id, contentHash) pair. */ + contentHash: string +} + +export interface RetrievalHoldoutEvent { + v: 1 + eventId: string + ts: string + adapterId?: string + sessionId?: string + taskId?: string + /** 1-based call counter within the session; 0 when the call is outside session randomization. */ + callIndex: number + /** sha256(sessionId) prefix — the seed key for the replayable assignment draws. */ + rngKey?: string + queryHash?: string + scope?: AgentMemoryScope + config: { epsilon: number; watchlist: string[]; configVersion?: string } + /** False when no sessionId is available, so the fraction-under-experiment denominator stays honest. */ + holdoutEligible: boolean + /** The full post-filter eligibility set E, logged on every call (control arm + interference probes). */ + eligible: RetrievalHoldoutEligibleItem[] + /** Ids in watchlist ∩ E, in eligibility order. */ + watchlistEligible: string[] + sessionHoldout: boolean + /** The session's sticky drop target once drawn; distinguishes "target absent from E" from "not yet drawn". */ + sessionTargetId: string | null + /** The item suppressed in THIS call, or null. */ + droppedId: string | null + /** 1/|watchlist ∩ E| recorded at draw time; the exact inverse-propensity weight input. */ + pickPropensity: number | null + /** epsilon * pickPropensity, recorded at draw time so analysis never re-derives assignment probabilities. */ + dropPropensity: number | null + deliveredIds: string[] + corpusVersion?: string +} + +export interface RetrievalHoldoutSessionState { + sessionId: string + /** Calls observed so far in this session. */ + callCount: number + sessionHoldout: boolean + /** Sticky drop target; drawn once at the first call whose eligibility set intersects the watchlist. */ + targetId: string | null + pickPropensity: number | null +} + +export interface RetrievalHoldoutCallContext { + sessionId?: string + taskId?: string + /** Raw query; only its sha256 prefix is logged. */ + query?: string + scope?: AgentMemoryScope + /** State returned by the previous call of this session; threading it is what makes suppression sticky. */ + session?: RetrievalHoldoutSessionState +} + +export interface RetrievalHoldoutResult { + delivered: AgentMemoryHit[] + event: RetrievalHoldoutEvent + session?: RetrievalHoldoutSessionState +} + +/** Deterministic uniform [0,1): 13 hex chars = 52 bits, exactly representable in a double. */ +export function deterministicRng(key: string): number { + return Number.parseInt(sha256(key).slice(0, 13), 16) / 16 ** 13 +} + +/** + * Pure per-call holdout: takes post-filter hits, returns delivered hits plus the log event + * and the next session state. Suppression only removes items (no backfill), and it happens + * before rendering so a drop session's context is byte-identical to a natural smaller retrieval. + */ +export function applyRetrievalHoldout( + hits: AgentMemoryHit[], + config: RetrievalHoldoutConfig, + ctx: RetrievalHoldoutCallContext = {}, +): RetrievalHoldoutResult { + const rng = config.rng ?? deterministicRng + const watchlist = config.watchlist ?? [] + const sessionId = ctx.sessionId ?? ctx.scope?.sessionId + const holdoutEligible = typeof sessionId === 'string' && sessionId.length > 0 + + let session: RetrievalHoldoutSessionState | undefined + if (holdoutEligible) { + const prior = ctx.session?.sessionId === sessionId ? ctx.session : undefined + session = prior ?? { + sessionId, + callCount: 0, + // The epsilon coin depends only on sessionId, so assignment is independent of task + // features by construction and exactly replayable (design rules D1 + D5). + sessionHoldout: rng(`${sessionId}#holdout`) < config.epsilon, + targetId: null, + pickPropensity: null, + } + session = { ...session, callCount: session.callCount + 1 } + } + + const watchlistSet = new Set(watchlist) + const watchlistEligible = hits.filter((hit) => watchlistSet.has(hit.id)).map((hit) => hit.id) + + if (session?.sessionHoldout && session.targetId === null && watchlistEligible.length > 0) { + // Draw once, uniformly over watchlist ∩ E at the first intersecting call, then stay sticky. + // Candidates are sorted by id so the draw does not depend on retrieval ordering. + const candidates = [...watchlistEligible].sort() + const index = Math.min( + Math.floor(rng(`${session.sessionId}#pick`) * candidates.length), + candidates.length - 1, + ) + const picked = candidates[index] + if (picked !== undefined) { + session = { ...session, targetId: picked, pickPropensity: 1 / candidates.length } + } + } + + const targetId = session?.sessionHoldout ? (session.targetId ?? null) : null + const droppedId = targetId !== null && hits.some((hit) => hit.id === targetId) ? targetId : null + const delivered = droppedId === null ? hits : hits.filter((hit) => hit.id !== droppedId) + + const pickPropensity = session?.sessionHoldout ? (session.pickPropensity ?? null) : null + const event: RetrievalHoldoutEvent = { + v: 1, + eventId: randomUUID(), + ts: new Date().toISOString(), + ...(config.adapterId !== undefined ? { adapterId: config.adapterId } : {}), + ...(sessionId !== undefined ? { sessionId } : {}), + ...(ctx.taskId !== undefined ? { taskId: ctx.taskId } : {}), + callIndex: session?.callCount ?? 0, + ...(sessionId !== undefined ? { rngKey: sha256(sessionId).slice(0, 16) } : {}), + ...(ctx.query !== undefined ? { queryHash: sha256(ctx.query).slice(0, 16) } : {}), + ...(ctx.scope !== undefined ? { scope: ctx.scope } : {}), + config: { + epsilon: config.epsilon, + watchlist: [...watchlist], + ...(config.configVersion !== undefined ? { configVersion: config.configVersion } : {}), + }, + holdoutEligible, + eligible: hits.map((hit, index) => ({ + id: hit.id, + rank: index + 1, + ...(typeof (hit.normalizedScore ?? hit.score) === 'number' + ? { score: hit.normalizedScore ?? hit.score } + : {}), + kind: hit.kind, + contentHash: sha256(hit.text).slice(0, 16), + })), + watchlistEligible, + sessionHoldout: session?.sessionHoldout ?? false, + sessionTargetId: targetId, + droppedId, + pickPropensity, + dropPropensity: pickPropensity !== null ? config.epsilon * pickPropensity : null, + deliveredIds: delivered.map((hit) => hit.id), + ...(config.corpusVersion !== undefined ? { corpusVersion: config.corpusVersion } : {}), + } + config.onEvent(event) + + return droppedId === null + ? { delivered: hits, event, ...(session !== undefined ? { session } : {}) } + : { delivered, event, ...(session !== undefined ? { session } : {}) } +} + +// Session states are tiny (5 scalar fields); the cap only guards unbounded long-lived processes. +// Evicting a live session degrades detectably, not silently: a re-drawn target that differs +// shows up in the log as a mixed-exposure session, which analysis excludes and counts. +const MAX_TRACKED_SESSIONS = 10_000 +const sessionRegistry = new WeakMap< + RetrievalHoldoutConfig, + Map +>() + +/** + * Convenience wrapper that threads session state internally, keyed by config object identity. + * Reuse ONE config object across all calls of a session (normally one per harness) or + * stickiness falls back to the deterministic per-session draws alone. + */ +export function applySessionStickyRetrievalHoldout( + hits: AgentMemoryHit[], + config: RetrievalHoldoutConfig, + ctx: Omit = {}, +): RetrievalHoldoutResult { + const sessionId = ctx.sessionId ?? ctx.scope?.sessionId + let sessions = sessionRegistry.get(config) + if (!sessions) { + sessions = new Map() + sessionRegistry.set(config, sessions) + } + const prior = sessionId !== undefined ? sessions.get(sessionId) : undefined + const result = applyRetrievalHoldout(hits, config, { + ...ctx, + ...(prior ? { session: prior } : {}), + }) + if (sessionId !== undefined && result.session) { + if (!sessions.has(sessionId) && sessions.size >= MAX_TRACKED_SESSIONS) { + const oldest = sessions.keys().next().value + if (oldest !== undefined) sessions.delete(oldest) + } + sessions.set(sessionId, result.session) + } + return result +} diff --git a/src/memory/index.ts b/src/memory/index.ts index 8d7569e..da6fffb 100644 --- a/src/memory/index.ts +++ b/src/memory/index.ts @@ -1,4 +1,5 @@ export * from './adapter' +export * from './holdout' export * from './neo4j' export * from './schemas' export * from './source-record' diff --git a/src/memory/types.ts b/src/memory/types.ts index 138dda1..a538d1f 100644 --- a/src/memory/types.ts +++ b/src/memory/types.ts @@ -1,4 +1,5 @@ import type { SourceRecord } from '../types' +import type { RetrievalHoldoutConfig } from './holdout' export type AgentMemoryKind = | 'message' @@ -45,6 +46,11 @@ export interface AgentMemorySearchOptions { minScore?: number kinds?: AgentMemoryKind[] metadata?: Record + /** + * Opt-in randomized retrieval holdout (epsilon-dropout) for treatment-effect logging. + * Absent by default; when absent, retrieval behavior is unchanged. See ./holdout. + */ + holdout?: RetrievalHoldoutConfig } export interface AgentMemoryWriteInput { diff --git a/tests/memory-holdout.test.ts b/tests/memory-holdout.test.ts new file mode 100644 index 0000000..a33f302 --- /dev/null +++ b/tests/memory-holdout.test.ts @@ -0,0 +1,266 @@ +import { describe, expect, it } from 'vitest' +import { sha256 } from '../src/ids' +import type { + AgentMemoryHit, + RetrievalHoldoutConfig, + RetrievalHoldoutEvent, +} from '../src/memory/index' +import { + applyRetrievalHoldout, + applySessionStickyRetrievalHoldout, + defaultGetMemoryContext, + deterministicRng, + renderMemoryContext, +} from '../src/memory/index' + +function hit(id: string, text: string, score: number): AgentMemoryHit { + return { id, uri: `memory://test/${id}`, kind: 'fact', text, normalizedScore: score } +} + +function collectingConfig( + overrides: Partial & Pick, +): { config: RetrievalHoldoutConfig; events: RetrievalHoldoutEvent[] } { + const events: RetrievalHoldoutEvent[] = [] + return { config: { onEvent: (event) => events.push(event), ...overrides }, events } +} + +describe('retrieval holdout: default-off is a byte-identical no-op', () => { + it('leaves defaultGetMemoryContext untouched when no holdout is configured', async () => { + const hits = [hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8)] + const context = await defaultGetMemoryContext({ search: async () => hits }, 'q', { + scope: { sessionId: 's-1' }, + }) + // Same array instance: the unconfigured code path never copies or filters. + expect(context.hits).toBe(hits) + // Frozen from the pre-holdout renderer output; a byte-level drift here fails the test. + expect(context.text).toBe('[1] fact:m1 score=0.900\nalpha\n\n[2] fact:m7 score=0.800\nbeta') + expect(context.sourceRecords).toHaveLength(2) + }) +}) + +describe('retrieval holdout: enabled behavior and event schema', () => { + // Forces holdout (epsilon 1) and picks index floor(0.6 * |candidates|) deterministically. + const forcedRng = (key: string) => (key.endsWith('#pick') ? 0.6 : 0) + + it('drops exactly one watchlist item and emits the registered event schema', () => { + const { config, events } = collectingConfig({ + epsilon: 1, + watchlist: ['m1', 'm7'], + configVersion: '2026-07-04a', + adapterId: 'test-adapter', + corpusVersion: 'kb-test-1', + rng: forcedRng, + }) + const hits = [hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8), hit('m3', 'gamma', 0.7)] + const { delivered, event } = applyRetrievalHoldout(hits, config, { + sessionId: 's-1', + taskId: 'task-17', + query: 'the query', + scope: { sessionId: 's-1', namespace: 'ns' }, + }) + + // sorted(watchlist ∩ E) = ['m1', 'm7']; floor(0.6 * 2) = 1 → 'm7'. + expect(delivered.map((h) => h.id)).toEqual(['m1', 'm3']) + expect(events).toHaveLength(1) + expect(Object.keys(event).sort()).toEqual( + [ + 'adapterId', + 'callIndex', + 'config', + 'corpusVersion', + 'deliveredIds', + 'droppedId', + 'dropPropensity', + 'eligible', + 'eventId', + 'holdoutEligible', + 'pickPropensity', + 'queryHash', + 'rngKey', + 'scope', + 'sessionHoldout', + 'sessionId', + 'sessionTargetId', + 'taskId', + 'ts', + 'v', + 'watchlistEligible', + ].sort(), + ) + expect(event.v).toBe(1) + expect(event.sessionId).toBe('s-1') + expect(event.taskId).toBe('task-17') + expect(event.callIndex).toBe(1) + expect(event.holdoutEligible).toBe(true) + expect(event.sessionHoldout).toBe(true) + expect(event.sessionTargetId).toBe('m7') + expect(event.droppedId).toBe('m7') + // Draw rule: pick uniform over |watchlist ∩ E| = 2, drop = epsilon * pick. + expect(event.pickPropensity).toBe(0.5) + expect(event.dropPropensity).toBe(0.5) + expect(event.deliveredIds).toEqual(['m1', 'm3']) + expect(event.watchlistEligible).toEqual(['m1', 'm7']) + expect(event.eligible).toEqual([ + { id: 'm1', rank: 1, score: 0.9, kind: 'fact', contentHash: sha256('alpha').slice(0, 16) }, + { id: 'm7', rank: 2, score: 0.8, kind: 'fact', contentHash: sha256('beta').slice(0, 16) }, + { id: 'm3', rank: 3, score: 0.7, kind: 'fact', contentHash: sha256('gamma').slice(0, 16) }, + ]) + expect(event.rngKey).toBe(sha256('s-1').slice(0, 16)) + expect(event.queryHash).toBe(sha256('the query').slice(0, 16)) + expect(event.config).toEqual({ + epsilon: 1, + watchlist: ['m1', 'm7'], + configVersion: '2026-07-04a', + }) + expect(event.adapterId).toBe('test-adapter') + expect(event.corpusVersion).toBe('kb-test-1') + // The manipulation is invisible: the rendered context renumbers with no gap (design rule D6). + expect(renderMemoryContext(delivered)).toBe( + '[1] fact:m1 score=0.900\nalpha\n\n[2] fact:m3 score=0.700\ngamma', + ) + }) + + it('suppresses the same target on every call of the session (sticky), via the adapter path', async () => { + const { config, events } = collectingConfig({ + epsilon: 1, + watchlist: ['m1', 'm7'], + rng: forcedRng, + }) + const options = { scope: { sessionId: 's-2', tags: { taskId: 'task-9' } }, holdout: config } + const call1 = await defaultGetMemoryContext( + { search: async () => [hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8)] }, + 'q1', + options, + ) + const call2 = await defaultGetMemoryContext( + { search: async () => [hit('m7', 'beta', 0.8), hit('m3', 'gamma', 0.7)] }, + 'q2', + options, + ) + // Third call's eligibility set no longer contains the target: nothing is dropped, + // but the sticky target is still reported for mixed-exposure audits. + const call3 = await defaultGetMemoryContext( + { search: async () => [hit('m3', 'gamma', 0.7)] }, + 'q3', + options, + ) + + expect(call1.hits.map((h) => h.id)).toEqual(['m1']) + expect(call2.hits.map((h) => h.id)).toEqual(['m3']) + expect(call3.hits.map((h) => h.id)).toEqual(['m3']) + expect(events.map((e) => e.callIndex)).toEqual([1, 2, 3]) + expect(events.map((e) => e.droppedId)).toEqual(['m7', 'm7', null]) + expect(events.map((e) => e.sessionTargetId)).toEqual(['m7', 'm7', 'm7']) + expect(events.map((e) => e.taskId)).toEqual(['task-9', 'task-9', 'task-9']) + expect(events.every((e) => e.sessionHoldout)).toBe(true) + }) + + it('emits control-arm events on no-drop calls, including at epsilon 0', () => { + const { config, events } = collectingConfig({ + epsilon: 0, + watchlist: ['m1'], + }) + const hits = [hit('m1', 'alpha', 0.9), hit('m2', 'beta', 0.8)] + const { delivered, event } = applyRetrievalHoldout(hits, config, { sessionId: 's-3' }) + + expect(delivered).toBe(hits) + expect(events).toHaveLength(1) + expect(event.holdoutEligible).toBe(true) + expect(event.sessionHoldout).toBe(false) + expect(event.droppedId).toBeNull() + expect(event.sessionTargetId).toBeNull() + expect(event.pickPropensity).toBeNull() + expect(event.dropPropensity).toBeNull() + // The full eligibility set is still logged: control-arm membership is half the data. + expect(event.eligible.map((e) => e.id)).toEqual(['m1', 'm2']) + expect(event.deliveredIds).toEqual(['m1', 'm2']) + }) + + it('records pickPropensity 1 and dropPropensity epsilon when one watchlist item is eligible', () => { + const { config } = collectingConfig({ + epsilon: 0.2, + watchlist: ['m7', 'm-absent'], + rng: () => 0, + }) + const { event } = applyRetrievalHoldout( + [hit('m7', 'beta', 0.8), hit('m3', 'gamma', 0.7)], + config, + { + sessionId: 's-4', + }, + ) + expect(event.droppedId).toBe('m7') + expect(event.pickPropensity).toBe(1) + expect(event.dropPropensity).toBeCloseTo(0.2, 12) + }) + + it('marks calls without a sessionId as holdout-ineligible and never drops', () => { + const { config, events } = collectingConfig({ epsilon: 1, watchlist: ['m1'], rng: () => 0 }) + const hits = [hit('m1', 'alpha', 0.9)] + const { delivered, event } = applyRetrievalHoldout(hits, config, {}) + + expect(delivered).toBe(hits) + expect(events).toHaveLength(1) + expect(event.holdoutEligible).toBe(false) + expect(event.sessionHoldout).toBe(false) + expect(event.droppedId).toBeNull() + expect(event.callIndex).toBe(0) + }) + + it('threads explicit session state through the pure function identically to the sticky wrapper', () => { + const { config: pureConfig } = collectingConfig({ + epsilon: 1, + watchlist: ['m1', 'm7'], + rng: forcedRng, + }) + const { config: stickyConfig } = collectingConfig({ + epsilon: 1, + watchlist: ['m1', 'm7'], + rng: forcedRng, + }) + const call1Hits = [hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8)] + const call2Hits = [hit('m7', 'beta', 0.8), hit('m3', 'gamma', 0.7)] + + const pure1 = applyRetrievalHoldout(call1Hits, pureConfig, { sessionId: 's-5' }) + const pure2 = applyRetrievalHoldout(call2Hits, pureConfig, { + sessionId: 's-5', + session: pure1.session, + }) + const sticky1 = applySessionStickyRetrievalHoldout(call1Hits, stickyConfig, { + sessionId: 's-5', + }) + const sticky2 = applySessionStickyRetrievalHoldout(call2Hits, stickyConfig, { + sessionId: 's-5', + }) + + expect(pure1.event.droppedId).toBe(sticky1.event.droppedId) + expect(pure2.event.droppedId).toBe(sticky2.event.droppedId) + expect(pure2.event.droppedId).toBe('m7') + expect(pure2.event.callIndex).toBe(2) + }) +}) + +describe('retrieval holdout: empirical calibration of the default rng', () => { + it('drops at a rate matching epsilon over 10^4 simulated sessions', () => { + const epsilon = 0.2 + const { config, events } = collectingConfig({ epsilon, watchlist: ['m1', 'm7'] }) + const n = 10_000 + for (let i = 0; i < n; i += 1) { + applyRetrievalHoldout([hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8)], config, { + sessionId: `session-${i}`, + }) + } + const drops = events.filter((e) => e.droppedId !== null) + const dropRate = drops.length / n + // Binomial sd at n=1e4, p=0.2 is 0.004; ±0.015 is ~3.75 sd, and the default rng is + // deterministic so this either always passes or flags a real bias. + expect(dropRate).toBeGreaterThan(epsilon - 0.015) + expect(dropRate).toBeLessThan(epsilon + 0.015) + // The uniform pick over two sorted candidates should split drops roughly evenly. + const m1Share = drops.filter((e) => e.droppedId === 'm1').length / drops.length + expect(m1Share).toBeGreaterThan(0.4) + expect(m1Share).toBeLessThan(0.6) + // Replayability (design rule D5): the same sessionId reproduces the same decision. + expect(deterministicRng('session-0#holdout')).toBe(deterministicRng('session-0#holdout')) + }) +}) From 8647badd5dcc46cb1a3407a8efb6cc02496edf53 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Sat, 4 Jul 2026 16:10:51 -0600 Subject: [PATCH 2/4] fix(memory): log holdout bypass events on neo4j paths, reuse agent-eval off-policy vocabulary Addresses the two PR #46 value-audit findings: - integration: createNeo4jAgentMemoryAdapter.getContext's short-term and raw-string early returns previously skipped the holdout hook silently. With options.holdout configured they now emit one holdoutEligible:false event with bypassReason ('short-term-context' | 'raw-string-context') via the same onEvent sink, keeping the estimator's denominator honest. No dropping on these paths: suppression is only meaningful for retrieved memory hits. - better-architecture: toOffPolicyTrajectory() + retrievalHoldoutBehaviorProb() map holdout events onto agent-eval's OffPolicyTrajectory so EXP-007 consumes the substrate's IPS/SNIPS/DR estimators directly (behaviorProb = the event's logged pick/drop propensity; mapping documented on the function). deterministicRng now seeds agent-eval's mulberry32 from the sha256 key derivation instead of hand-rolling the uniform. Pre-registration stays in the research repo by design; the doc comment points at agent-eval's HypothesisManifest vocabulary. --- src/memory/holdout.ts | 163 +++++++++++++++++++++--- src/memory/neo4j.ts | 30 +++++ tests/memory-holdout.test.ts | 232 +++++++++++++++++++++++++++++++++++ 3 files changed, 405 insertions(+), 20 deletions(-) diff --git a/src/memory/holdout.ts b/src/memory/holdout.ts index c84a0c8..281adf3 100644 --- a/src/memory/holdout.ts +++ b/src/memory/holdout.ts @@ -1,4 +1,6 @@ import { randomUUID } from 'node:crypto' +import { mulberry32 } from '@tangle-network/agent-eval' +import type { OffPolicyTrajectory } from '@tangle-network/agent-eval/rl' import { sha256 } from '../ids' import type { AgentMemoryHit, AgentMemoryScope } from './types' @@ -52,8 +54,13 @@ export interface RetrievalHoldoutEvent { queryHash?: string scope?: AgentMemoryScope config: { epsilon: number; watchlist: string[]; configVersion?: string } - /** False when no sessionId is available, so the fraction-under-experiment denominator stays honest. */ + /** + * False when no sessionId is available or the adapter answered without retrieval + * (see bypassReason), so the fraction-under-experiment denominator stays honest. + */ holdoutEligible: boolean + /** Present only on adapter paths that bypassed retrieval, where no suppression could apply. */ + bypassReason?: RetrievalHoldoutBypassReason /** The full post-filter eligibility set E, logged on every call (control arm + interference probes). */ eligible: RetrievalHoldoutEligibleItem[] /** Ids in watchlist ∩ E, in eligibility order. */ @@ -97,9 +104,17 @@ export interface RetrievalHoldoutResult { session?: RetrievalHoldoutSessionState } -/** Deterministic uniform [0,1): 13 hex chars = 52 bits, exactly representable in a double. */ +/** Adapter context paths that answer without retrieval, so no holdout draw can happen. */ +export type RetrievalHoldoutBypassReason = 'short-term-context' | 'raw-string-context' + +/** + * Deterministic uniform [0,1) for assignment draws. The sha256 key derivation is ours — it makes + * every draw replayable from the logged keys alone (design rule D5) — while the generator core is + * the substrate's `mulberry32` (@tangle-network/agent-eval statistics vocabulary), seeded with the + * top 32 bits of the digest, so the statistical machinery is reused rather than forked. + */ export function deterministicRng(key: string): number { - return Number.parseInt(sha256(key).slice(0, 13), 16) / 16 ** 13 + return mulberry32(Number.parseInt(sha256(key).slice(0, 8), 16))() } /** @@ -113,9 +128,9 @@ export function applyRetrievalHoldout( ctx: RetrievalHoldoutCallContext = {}, ): RetrievalHoldoutResult { const rng = config.rng ?? deterministicRng - const watchlist = config.watchlist ?? [] const sessionId = ctx.sessionId ?? ctx.scope?.sessionId const holdoutEligible = typeof sessionId === 'string' && sessionId.length > 0 + const base = holdoutEventBase(hits, config, ctx) let session: RetrievalHoldoutSessionState | undefined if (holdoutEligible) { @@ -132,8 +147,7 @@ export function applyRetrievalHoldout( session = { ...session, callCount: session.callCount + 1 } } - const watchlistSet = new Set(watchlist) - const watchlistEligible = hits.filter((hit) => watchlistSet.has(hit.id)).map((hit) => hit.id) + const watchlistEligible = base.watchlistEligible if (session?.sessionHoldout && session.targetId === null && watchlistEligible.length > 0) { // Draw once, uniformly over watchlist ∩ E at the first intersecting call, then stay sticky. @@ -155,13 +169,39 @@ export function applyRetrievalHoldout( const pickPropensity = session?.sessionHoldout ? (session.pickPropensity ?? null) : null const event: RetrievalHoldoutEvent = { - v: 1, + ...base, + callIndex: session?.callCount ?? 0, + holdoutEligible, + sessionHoldout: session?.sessionHoldout ?? false, + sessionTargetId: targetId, + droppedId, + pickPropensity, + dropPropensity: pickPropensity !== null ? config.epsilon * pickPropensity : null, + deliveredIds: delivered.map((hit) => hit.id), + } + config.onEvent(event) + + return droppedId === null + ? { delivered: hits, event, ...(session !== undefined ? { session } : {}) } + : { delivered, event, ...(session !== undefined ? { session } : {}) } +} + +/** Fields shared by randomized and bypass events: identity, attribution, config echo, eligibility. */ +function holdoutEventBase( + hits: AgentMemoryHit[], + config: RetrievalHoldoutConfig, + ctx: Omit, +) { + const sessionId = ctx.sessionId ?? ctx.scope?.sessionId + const watchlist = config.watchlist ?? [] + const watchlistSet = new Set(watchlist) + return { + v: 1 as const, eventId: randomUUID(), ts: new Date().toISOString(), ...(config.adapterId !== undefined ? { adapterId: config.adapterId } : {}), ...(sessionId !== undefined ? { sessionId } : {}), ...(ctx.taskId !== undefined ? { taskId: ctx.taskId } : {}), - callIndex: session?.callCount ?? 0, ...(sessionId !== undefined ? { rngKey: sha256(sessionId).slice(0, 16) } : {}), ...(ctx.query !== undefined ? { queryHash: sha256(ctx.query).slice(0, 16) } : {}), ...(ctx.scope !== undefined ? { scope: ctx.scope } : {}), @@ -170,7 +210,6 @@ export function applyRetrievalHoldout( watchlist: [...watchlist], ...(config.configVersion !== undefined ? { configVersion: config.configVersion } : {}), }, - holdoutEligible, eligible: hits.map((hit, index) => ({ id: hit.id, rank: index + 1, @@ -180,20 +219,38 @@ export function applyRetrievalHoldout( kind: hit.kind, contentHash: sha256(hit.text).slice(0, 16), })), - watchlistEligible, - sessionHoldout: session?.sessionHoldout ?? false, - sessionTargetId: targetId, - droppedId, - pickPropensity, - dropPropensity: pickPropensity !== null ? config.epsilon * pickPropensity : null, - deliveredIds: delivered.map((hit) => hit.id), + watchlistEligible: hits.filter((hit) => watchlistSet.has(hit.id)).map((hit) => hit.id), ...(config.corpusVersion !== undefined ? { corpusVersion: config.corpusVersion } : {}), } - config.onEvent(event) +} - return droppedId === null - ? { delivered: hits, event, ...(session !== undefined ? { session } : {}) } - : { delivered, event, ...(session !== undefined ? { session } : {}) } +/** + * Logs a holdout event for adapter context paths that answer WITHOUT going through the + * search→render seam (Neo4j short-term conversation context, raw-string getContext results), + * so a consumer with a holdout configured still sees every call and the fraction-under-experiment + * denominator stays honest instead of silently losing these calls. No suppression is applied: + * dropping is only meaningful for retrieved memory hits, never for conversation context. + */ +export function emitRetrievalHoldoutBypass( + hits: AgentMemoryHit[], + config: RetrievalHoldoutConfig, + ctx: Omit, + bypassReason: RetrievalHoldoutBypassReason, +): RetrievalHoldoutEvent { + const event: RetrievalHoldoutEvent = { + ...holdoutEventBase(hits, config, ctx), + callIndex: 0, + holdoutEligible: false, + sessionHoldout: false, + sessionTargetId: null, + droppedId: null, + pickPropensity: null, + dropPropensity: null, + deliveredIds: hits.map((hit) => hit.id), + bypassReason, + } + config.onEvent(event) + return event } // Session states are tiny (5 scalar fields); the cap only guards unbounded long-lived processes. @@ -235,3 +292,69 @@ export function applySessionStickyRetrievalHoldout( } return result } + +export interface RetrievalHoldoutOffPolicyOptions { + /** Realized outcome of the call's session/task, joined by the caller — events carry no reward. */ + reward: (event: RetrievalHoldoutEvent) => number + /** + * Target-policy probability of the LOGGED delivery. Defaults to the "always deliver in full" + * policy: 1 when nothing was dropped, 0 when something was. + */ + targetProb?: (event: RetrievalHoldoutEvent) => number + /** Reward-model prediction enabling `doublyRobust`; when absent, entries carry qHat null. */ + qHat?: (event: RetrievalHoldoutEvent) => number | null +} + +/** + * Behavior-policy probability of the delivery an event records, reconstructed from the logged + * propensities alone: + * - drop events: `dropPropensity` (= epsilon × pickPropensity, recorded at draw time); + * - no-drop events whose watchlist intersected the eligibility set: 1 − epsilon × pick, where + * pick is the logged `pickPropensity` (holdout arm, sticky draw) or 1/|watchlist ∩ E| + * (control arm — the uniform draw that would have happened, exact at the session's first + * intersecting call); + * - calls where no drop was possible (bypass paths, holdout-ineligible calls, empty + * watchlist ∩ E): 1, because full delivery was the only action the behavior policy could take. + */ +export function retrievalHoldoutBehaviorProb(event: RetrievalHoldoutEvent): number { + if (event.droppedId !== null) { + if (event.dropPropensity === null) + throw new Error(`holdout event ${event.eventId} recorded a drop without dropPropensity`) + return event.dropPropensity + } + if (!event.holdoutEligible) return 1 + const pick = + event.pickPropensity ?? + (event.watchlistEligible.length > 0 ? 1 / event.watchlistEligible.length : null) + return pick === null ? 1 : 1 - event.config.epsilon * pick +} + +/** + * Maps holdout log events onto agent-eval's `OffPolicyTrajectory` so EXP-007's analysis consumes + * the substrate's `inverseProbabilityWeighting` / `selfNormalizedImportanceWeighting` / + * `doublyRobust` estimators directly instead of re-deriving them from raw events. + * + * Field mapping (one trajectory per event): + * - `runId` ← `eventId` + * - `reward` ← `options.reward(event)` — joined by the caller, events carry no outcome + * - `behaviorProb` ← the event's logged pick/drop propensity ({@link retrievalHoldoutBehaviorProb}) + * - `targetProb` ← `options.targetProb(event)`; default models the always-deliver-in-full policy + * - `qHat` ← `options.qHat(event)`, else null (IPS/SNIPS-only) + * + * The per-event mapping treats calls independently; for a strict session-level estimand, + * aggregate to one trajectory per session (its first watchlist-intersecting call) upstream. + * Pre-registration for EXP-007 stays in the research repo by design; the manifest vocabulary for + * it is agent-eval's `HypothesisManifest` / `signManifest` / `evaluateHypothesis` exports. + */ +export function toOffPolicyTrajectory( + events: RetrievalHoldoutEvent[], + options: RetrievalHoldoutOffPolicyOptions, +): OffPolicyTrajectory[] { + return events.map((event) => ({ + runId: event.eventId, + reward: options.reward(event), + behaviorProb: retrievalHoldoutBehaviorProb(event), + targetProb: options.targetProb?.(event) ?? (event.droppedId === null ? 1 : 0), + qHat: options.qHat?.(event) ?? null, + })) +} diff --git a/src/memory/neo4j.ts b/src/memory/neo4j.ts index 162e3ac..a6e73db 100644 --- a/src/memory/neo4j.ts +++ b/src/memory/neo4j.ts @@ -1,9 +1,11 @@ import { stableId } from '../ids' import { defaultGetMemoryContext } from './adapter' +import { emitRetrievalHoldoutBypass } from './holdout' import { memoryHitToSourceRecord, memoryWriteResultToSourceRecord } from './source-record' import type { AgentMemoryAdapter, AgentMemoryHit, + AgentMemoryScope, AgentMemorySearchOptions, AgentMemoryWriteInput, AgentMemoryWriteResult, @@ -58,6 +60,14 @@ export function createNeo4jAgentMemoryAdapter( : hits.length > 0 ? renderHits(hits) : '') + if (searchOptions.holdout) { + emitRetrievalHoldoutBypass( + hits, + searchOptions.holdout, + holdoutBypassContext(query, searchOptions), + 'short-term-context', + ) + } return { query, text, @@ -102,6 +112,14 @@ export function createNeo4jAgentMemoryAdapter( title: 'Memory context', normalizedScore: 1, } + if (searchOptions.holdout) { + emitRetrievalHoldoutBypass( + [hit], + searchOptions.holdout, + holdoutBypassContext(query, searchOptions), + 'raw-string-context', + ) + } return { query, text: result, @@ -285,6 +303,18 @@ async function searchNeo4jMemory( .slice(0, options.limit) } +// Mirrors defaultGetMemoryContext's holdout call context so bypass events join the same log stream. +function holdoutBypassContext( + query: string, + options: AgentMemorySearchOptions, +): { query: string; scope?: AgentMemoryScope; taskId?: string } { + return { + query, + ...(options.scope !== undefined ? { scope: options.scope } : {}), + ...(options.scope?.tags?.taskId !== undefined ? { taskId: options.scope.tags.taskId } : {}), + } +} + function neo4jOptions(options: AgentMemorySearchOptions): Record { return { limit: options.limit, diff --git a/tests/memory-holdout.test.ts b/tests/memory-holdout.test.ts index a33f302..356a99c 100644 --- a/tests/memory-holdout.test.ts +++ b/tests/memory-holdout.test.ts @@ -1,3 +1,7 @@ +import { + inverseProbabilityWeighting, + selfNormalizedImportanceWeighting, +} from '@tangle-network/agent-eval/rl' import { describe, expect, it } from 'vitest' import { sha256 } from '../src/ids' import type { @@ -8,9 +12,13 @@ import type { import { applyRetrievalHoldout, applySessionStickyRetrievalHoldout, + createNeo4jAgentMemoryAdapter, defaultGetMemoryContext, deterministicRng, + emitRetrievalHoldoutBypass, renderMemoryContext, + retrievalHoldoutBehaviorProb, + toOffPolicyTrajectory, } from '../src/memory/index' function hit(id: string, text: string, score: number): AgentMemoryHit { @@ -240,6 +248,230 @@ describe('retrieval holdout: enabled behavior and event schema', () => { }) }) +describe('retrieval holdout: adapter bypass paths still log the call', () => { + it('emits a short-term-context bypass event instead of silently skipping the hook', async () => { + const { config, events } = collectingConfig({ + epsilon: 1, + watchlist: ['obs-1'], + adapterId: 'neo4j-test', + rng: () => 0, + }) + const client = { + shortTerm: { + async getContext(conversationId: string) { + return { + observations: [{ id: 'obs-1', content: `Observation for ${conversationId}` }], + recentMessages: [{ id: 'msg-1', role: 'user', content: 'Keep it brief.' }], + } + }, + }, + } + const adapter = createNeo4jAgentMemoryAdapter({ client }) + + const context = await adapter.getContext('style', { + scope: { sessionId: 'conv-1', tags: { taskId: 'task-3' } }, + holdout: config, + }) + + // The bypass never suppresses: dropping is only meaningful for retrieved memory hits. + expect(context.hits.map((h) => h.id)).toEqual(['obs-1', 'msg-1']) + expect(context.text).toContain('Observation for conv-1') + expect(events).toHaveLength(1) + const event = events[0]! + expect(event.bypassReason).toBe('short-term-context') + expect(event.holdoutEligible).toBe(false) + expect(event.sessionHoldout).toBe(false) + expect(event.droppedId).toBeNull() + expect(event.sessionTargetId).toBeNull() + expect(event.pickPropensity).toBeNull() + expect(event.dropPropensity).toBeNull() + expect(event.callIndex).toBe(0) + expect(event.sessionId).toBe('conv-1') + expect(event.taskId).toBe('task-3') + expect(event.adapterId).toBe('neo4j-test') + expect(event.watchlistEligible).toEqual(['obs-1']) + expect(event.eligible.map((e) => e.id)).toEqual(['obs-1', 'msg-1']) + expect(event.deliveredIds).toEqual(['obs-1', 'msg-1']) + expect(event.queryHash).toBe(sha256('style').slice(0, 16)) + }) + + it('emits a raw-string-context bypass event for string getContext results', async () => { + const { config, events } = collectingConfig({ epsilon: 1, watchlist: ['w-1'], rng: () => 0 }) + const client = { + async getContext() { + return 'Use the private project namespace.' + }, + } + const adapter = createNeo4jAgentMemoryAdapter({ client, id: 'neo4j-private' }) + + const context = await adapter.getContext('project namespace', { holdout: config }) + + expect(context.text).toBe('Use the private project namespace.') + expect(context.hits).toHaveLength(1) + expect(events).toHaveLength(1) + const event = events[0]! + expect(event.bypassReason).toBe('raw-string-context') + expect(event.holdoutEligible).toBe(false) + expect(event.droppedId).toBeNull() + expect(event.eligible).toEqual([ + { + id: context.hits[0]!.id, + rank: 1, + score: 1, + kind: 'fact', + contentHash: sha256('Use the private project namespace.').slice(0, 16), + }, + ]) + expect(event.deliveredIds).toEqual([context.hits[0]!.id]) + expect(event.sessionId).toBeUndefined() + expect(event.rngKey).toBeUndefined() + }) +}) + +describe('retrieval holdout: off-policy conversion for agent-eval estimators', () => { + // Forces holdout (epsilon 1) and picks index floor(0.6 * |candidates|) deterministically. + const forcedRng = (key: string) => (key.endsWith('#pick') ? 0.6 : 0) + + it('maps drop, keep, and control events onto OffPolicyTrajectory propensities', () => { + const { config, events } = collectingConfig({ + epsilon: 1, + watchlist: ['m1', 'm7'], + rng: forcedRng, + }) + const drop = applyRetrievalHoldout([hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8)], config, { + sessionId: 's-ope', + }) + // Same session, sticky target m7 absent from E: nothing dropped, propensity still logged. + applyRetrievalHoldout([hit('m3', 'gamma', 0.7)], config, { + sessionId: 's-ope', + session: drop.session, + }) + const { config: controlConfig, events: controlEvents } = collectingConfig({ + epsilon: 0.2, + watchlist: ['m1', 'm7'], + // Coin 0.99 >= epsilon: control arm, no pick is ever drawn. + rng: () => 0.99, + }) + applyRetrievalHoldout( + [hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8), hit('m3', 'gamma', 0.7)], + controlConfig, + { sessionId: 's-ctl' }, + ) + + const all = [...events, ...controlEvents] + const trajectories = toOffPolicyTrajectory(all, { + reward: (event) => (event.droppedId === null ? 1 : 0), + }) + + // Drop event: behaviorProb = dropPropensity = epsilon * pick = 1 * (1/2). + expect(trajectories[0]).toEqual({ + runId: all[0]!.eventId, + reward: 0, + behaviorProb: 0.5, + targetProb: 0, + qHat: null, + }) + // Keep event in the holdout arm: 1 - epsilon * logged sticky pickPropensity. + expect(trajectories[1]).toEqual({ + runId: all[1]!.eventId, + reward: 1, + behaviorProb: 0.5, + targetProb: 1, + qHat: null, + }) + // Control event: no draw was logged; counterfactual uniform pick over |watchlist ∩ E| = 2. + expect(trajectories[2]!.behaviorProb).toBeCloseTo(1 - 0.2 * 0.5, 12) + expect(trajectories[2]!.targetProb).toBe(1) + expect(trajectories[2]!.reward).toBe(1) + }) + + it('assigns behaviorProb 1 whenever no drop was possible', () => { + // Holdout-ineligible call: no sessionId, no randomization happened. + const { config } = collectingConfig({ epsilon: 1, watchlist: ['m1'], rng: () => 0 }) + const ineligible = applyRetrievalHoldout([hit('m1', 'alpha', 0.9)], config, {}).event + expect(retrievalHoldoutBehaviorProb(ineligible)).toBe(1) + + // Control session whose eligibility set never intersects the watchlist. + const { config: missConfig } = collectingConfig({ + epsilon: 0.2, + watchlist: ['m-absent'], + rng: () => 0.99, + }) + const miss = applyRetrievalHoldout([hit('m3', 'gamma', 0.7)], missConfig, { + sessionId: 's-miss', + }).event + expect(retrievalHoldoutBehaviorProb(miss)).toBe(1) + + // Adapter bypass event: the call never reached retrieval. + const { config: bypassConfig } = collectingConfig({ epsilon: 1, watchlist: ['m1'] }) + const bypass = emitRetrievalHoldoutBypass( + [hit('m1', 'alpha', 0.9)], + bypassConfig, + { query: 'q', scope: { sessionId: 's-bypass' } }, + 'short-term-context', + ) + expect(retrievalHoldoutBehaviorProb(bypass)).toBe(1) + }) + + it('fails loud on a corrupt drop event missing its propensity', () => { + const { config } = collectingConfig({ epsilon: 1, watchlist: ['m1'], rng: () => 0 }) + const { event } = applyRetrievalHoldout([hit('m1', 'alpha', 0.9)], config, { + sessionId: 's-bad', + }) + expect(() => retrievalHoldoutBehaviorProb({ ...event, dropPropensity: null })).toThrow( + 'without dropPropensity', + ) + }) + + it('honors custom targetProb and qHat callbacks', () => { + const { config, events } = collectingConfig({ epsilon: 1, watchlist: ['m1'], rng: () => 0 }) + applyRetrievalHoldout([hit('m1', 'alpha', 0.9)], config, { sessionId: 's-custom' }) + + const [trajectory] = toOffPolicyTrajectory(events, { + reward: () => 0.75, + targetProb: () => 0.25, + qHat: () => 0.5, + }) + expect(trajectory).toEqual({ + runId: events[0]!.eventId, + reward: 0.75, + behaviorProb: 1, + targetProb: 0.25, + qHat: 0.5, + }) + }) + + it('feeds converted logs into agent-eval IPS/SNIPS end to end', () => { + // 10 holdout + 10 control sessions at epsilon 0.5 with a single watchlist item: + // drops have behaviorProb 0.5, keeps 0.5, so the always-deliver target policy + // (weight 0 on drops, 2 on keeps) recovers the delivered-arm mean reward exactly. + const { config, events } = collectingConfig({ + epsilon: 0.5, + watchlist: ['m1'], + rng: (key) => (key.includes('holdout-session') ? 0 : 0.99), + }) + for (let i = 0; i < 10; i += 1) { + applyRetrievalHoldout([hit('m1', 'alpha', 0.9), hit('m2', 'beta', 0.8)], config, { + sessionId: `holdout-session-${i}`, + }) + applyRetrievalHoldout([hit('m1', 'alpha', 0.9), hit('m2', 'beta', 0.8)], config, { + sessionId: `control-session-${i}`, + }) + } + const trajectories = toOffPolicyTrajectory(events, { + reward: (event) => (event.droppedId === null ? 1 : 0), + }) + + expect(events.filter((e) => e.droppedId !== null)).toHaveLength(10) + const ips = inverseProbabilityWeighting(trajectories) + const snips = selfNormalizedImportanceWeighting(trajectories) + expect(ips.n).toBe(20) + expect(ips.value).toBeCloseTo(1, 12) + expect(snips.value).toBeCloseTo(1, 12) + expect(ips.maxImportanceWeight).toBeCloseTo(2, 12) + }) +}) + describe('retrieval holdout: empirical calibration of the default rng', () => { it('drops at a rate matching epsilon over 10^4 simulated sessions', () => { const epsilon = 0.2 From e289c54ac3d7fdbb2a915e397b771a2fcfbe0ba8 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Sat, 4 Jul 2026 16:21:10 -0600 Subject: [PATCH 3/4] fix(memory): value-keyed sticky sessions, default-private event identifiers, fail-loud config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the PR #46 multi-shot audit (1 high, 1 medium, 4 low): - HIGH: the sticky-session registry was keyed by config OBJECT IDENTITY, so callers building options inline per call (the natural adapter pattern) lost stickiness silently — callIndex reset and the drop target flipped mid-session, logging one session under-treatment for two different items. The registry is now keyed by VALUE: configHash = sha256({epsilon, sorted watchlist}) prefix, sessions nested per configHash, oldest-first eviction kept. configHash is emitted on every event so the estimator can group by experiment config. Regression test reproduces the reviewer's exact flip scenario with a fresh config object per call. - MEDIUM: events logged plaintext sessionId and verbatim scope (tenantId/userId/tags) into a consumer-controlled sink while hashing the query. Default-private now: sessionIdHash (sha256 prefix; replaces the identically-derived rngKey) and scopeHash (canonical JSON) are the join keys; plaintext sessionId/scope require includePlaintextIdentifiers: true. Join-key contract documented on toOffPolicyTrajectory. - LOW: epsilon is validated in [0,1] and watchlist as string[] at first use (fail loud); empty/omitted watchlist, LRU eviction (via new maxTrackedSessions testability knob, default 10k) incl. the mixed-exposure eviction signature, and the score-without-normalizedScore fallback are all now tested. --- src/memory/holdout.ts | 123 ++++++++++++++++--- tests/memory-holdout.test.ts | 231 +++++++++++++++++++++++++++++++++-- 2 files changed, 330 insertions(+), 24 deletions(-) diff --git a/src/memory/holdout.ts b/src/memory/holdout.ts index 281adf3..6ce8439 100644 --- a/src/memory/holdout.ts +++ b/src/memory/holdout.ts @@ -21,6 +21,19 @@ export interface RetrievalHoldoutConfig { adapterId?: string /** Corpus/store version stamp; an edited item under the same id is a different treatment. */ corpusVersion?: string + /** + * Emit plaintext sessionId and scope on events. Default false: events carry only + * sessionIdHash/scopeHash, so PII-bearing identifiers (tenantId/userId/tags) never reach a + * consumer-controlled sink unless the consumer explicitly owns that decision. Note that + * replaying assignment draws from logs alone needs the plaintext sessionId, so + * privacy-default logs require the consumer's own sessionId mapping for replay audits. + */ + includePlaintextIdentifiers?: boolean + /** + * Cap on tracked sessions per experiment config in the sticky wrapper's registry. + * Exists so tests can exercise eviction; production should keep the default (10,000). + */ + maxTrackedSessions?: number /** * Uniform-[0,1) generator keyed by a string. Defaults to a sha256-derived deterministic * generator so every assignment is replayable from the logged keys alone (design rule D5). @@ -45,15 +58,28 @@ export interface RetrievalHoldoutEvent { eventId: string ts: string adapterId?: string + /** Plaintext session id — emitted ONLY when config.includePlaintextIdentifiers is true. */ sessionId?: string + /** Consumer-supplied experiment/outcome join id (scope.tags.taskId); deliberately plaintext. */ taskId?: string /** 1-based call counter within the session; 0 when the call is outside session randomization. */ callIndex: number - /** sha256(sessionId) prefix — the seed key for the replayable assignment draws. */ - rngKey?: string + /** + * sha256(sessionId) prefix — the default privacy-preserving session join key AND the seed-key + * reference for the assignment draws (previously named rngKey; identical derivation, deduped). + */ + sessionIdHash?: string queryHash?: string + /** Verbatim scope — emitted ONLY when config.includePlaintextIdentifiers is true (PII risk). */ scope?: AgentMemoryScope + /** sha256 prefix of the canonical-JSON scope (keys sorted, undefined stripped). */ + scopeHash?: string config: { epsilon: number; watchlist: string[]; configVersion?: string } + /** + * Value-hash of the experiment-defining knobs, sha256({epsilon, sorted watchlist}) prefix. + * The estimator groups events by it; the sticky-session registry is keyed by it. + */ + configHash: string /** * False when no sessionId is available or the adapter answered without retrieval * (see bypassReason), so the fraction-under-experiment denominator stays honest. @@ -117,6 +143,48 @@ export function deterministicRng(key: string): number { return mulberry32(Number.parseInt(sha256(key).slice(0, 8), 16))() } +// Deterministic serialization for hashing: keys sorted, undefined-valued entries stripped, so +// the same logical scope/config always produces the same hash regardless of construction order. +function canonicalJson(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(',')}]` + if (value !== null && typeof value === 'object') { + const entries = Object.entries(value as Record) + .filter(([, v]) => v !== undefined) + .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) + .map(([k, v]) => `${JSON.stringify(k)}:${canonicalJson(v)}`) + return `{${entries.join(',')}}` + } + return JSON.stringify(value) +} + +/** + * Value-hash of the knobs that DEFINE the experiment (epsilon + watchlist, order-independent). + * Everything else on the config (callbacks, attribution stamps, privacy flags) does not change + * which assignments are drawn, so it stays out of the hash. + */ +export function retrievalHoldoutConfigHash( + config: Pick, +): string { + return sha256( + canonicalJson({ epsilon: config.epsilon, watchlist: [...(config.watchlist ?? [])].sort() }), + ).slice(0, 16) +} + +// Malformed knobs silently corrupt propensities (epsilon>1 forces holdout with dropPropensity>1; +// epsilon<0 disables it while still logging a control-arm-shaped stream), so fail loud instead. +function assertValidHoldoutConfig(config: RetrievalHoldoutConfig): void { + const { epsilon, watchlist } = config + if (typeof epsilon !== 'number' || !Number.isFinite(epsilon) || epsilon < 0 || epsilon > 1) { + throw new Error(`retrieval holdout epsilon must be a number in [0, 1], got ${String(epsilon)}`) + } + if ( + watchlist !== undefined && + (!Array.isArray(watchlist) || watchlist.some((id) => typeof id !== 'string')) + ) { + throw new Error('retrieval holdout watchlist must be an array of item-id strings') + } +} + /** * Pure per-call holdout: takes post-filter hits, returns delivered hits plus the log event * and the next session state. Suppression only removes items (no backfill), and it happens @@ -127,6 +195,7 @@ export function applyRetrievalHoldout( config: RetrievalHoldoutConfig, ctx: RetrievalHoldoutCallContext = {}, ): RetrievalHoldoutResult { + assertValidHoldoutConfig(config) const rng = config.rng ?? deterministicRng const sessionId = ctx.sessionId ?? ctx.scope?.sessionId const holdoutEligible = typeof sessionId === 'string' && sessionId.length > 0 @@ -195,21 +264,26 @@ function holdoutEventBase( const sessionId = ctx.sessionId ?? ctx.scope?.sessionId const watchlist = config.watchlist ?? [] const watchlistSet = new Set(watchlist) + const plaintext = config.includePlaintextIdentifiers === true return { v: 1 as const, eventId: randomUUID(), ts: new Date().toISOString(), ...(config.adapterId !== undefined ? { adapterId: config.adapterId } : {}), - ...(sessionId !== undefined ? { sessionId } : {}), + ...(plaintext && sessionId !== undefined ? { sessionId } : {}), ...(ctx.taskId !== undefined ? { taskId: ctx.taskId } : {}), - ...(sessionId !== undefined ? { rngKey: sha256(sessionId).slice(0, 16) } : {}), + ...(sessionId !== undefined ? { sessionIdHash: sha256(sessionId).slice(0, 16) } : {}), ...(ctx.query !== undefined ? { queryHash: sha256(ctx.query).slice(0, 16) } : {}), - ...(ctx.scope !== undefined ? { scope: ctx.scope } : {}), + ...(plaintext && ctx.scope !== undefined ? { scope: ctx.scope } : {}), + ...(ctx.scope !== undefined + ? { scopeHash: sha256(canonicalJson(ctx.scope)).slice(0, 16) } + : {}), config: { epsilon: config.epsilon, watchlist: [...watchlist], ...(config.configVersion !== undefined ? { configVersion: config.configVersion } : {}), }, + configHash: retrievalHoldoutConfigHash(config), eligible: hits.map((hit, index) => ({ id: hit.id, rank: index + 1, @@ -237,6 +311,7 @@ export function emitRetrievalHoldoutBypass( ctx: Omit, bypassReason: RetrievalHoldoutBypassReason, ): RetrievalHoldoutEvent { + assertValidHoldoutConfig(config) const event: RetrievalHoldoutEvent = { ...holdoutEventBase(hits, config, ctx), callIndex: 0, @@ -255,28 +330,35 @@ export function emitRetrievalHoldoutBypass( // Session states are tiny (5 scalar fields); the cap only guards unbounded long-lived processes. // Evicting a live session degrades detectably, not silently: a re-drawn target that differs -// shows up in the log as a mixed-exposure session, which analysis excludes and counts. -const MAX_TRACKED_SESSIONS = 10_000 -const sessionRegistry = new WeakMap< - RetrievalHoldoutConfig, - Map ->() +// shows up in the log as a mixed-exposure session (same sessionIdHash, different sessionTargetId, +// callIndex restarting at 1), which analysis excludes and counts. +const DEFAULT_MAX_TRACKED_SESSIONS = 10_000 +// Keyed by configHash VALUE, not config object identity: callers building options inline pass a +// fresh config object per call (the natural adapter pattern), and identity-keying would silently +// reset callIndex and re-draw the "sticky" target mid-session — one session logged as +// under-treatment for two different items, invalidating the per-item estimate. The outer map is +// bounded by the number of distinct experiment configs, in practice a handful. +const sessionRegistry = new Map>() /** - * Convenience wrapper that threads session state internally, keyed by config object identity. - * Reuse ONE config object across all calls of a session (normally one per harness) or - * stickiness falls back to the deterministic per-session draws alone. + * Convenience wrapper that threads session state internally, keyed by the VALUE of the + * experiment-defining knobs (configHash = epsilon + sorted watchlist) plus sessionId, so a fresh + * config object per call — the natural pattern when options are built inline — keeps full + * stickiness. Distinct experiments never share session state; two config objects with the same + * epsilon/watchlist are the same experiment by definition. */ export function applySessionStickyRetrievalHoldout( hits: AgentMemoryHit[], config: RetrievalHoldoutConfig, ctx: Omit = {}, ): RetrievalHoldoutResult { + assertValidHoldoutConfig(config) const sessionId = ctx.sessionId ?? ctx.scope?.sessionId - let sessions = sessionRegistry.get(config) + const configHash = retrievalHoldoutConfigHash(config) + let sessions = sessionRegistry.get(configHash) if (!sessions) { sessions = new Map() - sessionRegistry.set(config, sessions) + sessionRegistry.set(configHash, sessions) } const prior = sessionId !== undefined ? sessions.get(sessionId) : undefined const result = applyRetrievalHoldout(hits, config, { @@ -284,7 +366,9 @@ export function applySessionStickyRetrievalHoldout( ...(prior ? { session: prior } : {}), }) if (sessionId !== undefined && result.session) { - if (!sessions.has(sessionId) && sessions.size >= MAX_TRACKED_SESSIONS) { + const cap = config.maxTrackedSessions ?? DEFAULT_MAX_TRACKED_SESSIONS + if (!sessions.has(sessionId) && sessions.size >= cap) { + // Oldest-inserted-first: Map preserves insertion order and updates keep position. const oldest = sessions.keys().next().value if (oldest !== undefined) sessions.delete(oldest) } @@ -341,6 +425,11 @@ export function retrievalHoldoutBehaviorProb(event: RetrievalHoldoutEvent): numb * - `targetProb` ← `options.targetProb(event)`; default models the always-deliver-in-full policy * - `qHat` ← `options.qHat(event)`, else null (IPS/SNIPS-only) * + * Join keys: events carry `sessionIdHash` (= sha256(sessionId) first 16 hex) by default — compute + * the same prefix over the outcome table's session ids to join rewards, or set + * `includePlaintextIdentifiers: true` on the config to join on raw ids. Group and filter by + * `configHash` so trajectories from different experiment configs are never pooled. + * * The per-event mapping treats calls independently; for a strict session-level estimand, * aggregate to one trajectory per session (its first watchlist-intersecting call) upstream. * Pre-registration for EXP-007 stays in the research repo by design; the manifest vocabulary for diff --git a/tests/memory-holdout.test.ts b/tests/memory-holdout.test.ts index 356a99c..acf879b 100644 --- a/tests/memory-holdout.test.ts +++ b/tests/memory-holdout.test.ts @@ -18,6 +18,7 @@ import { emitRetrievalHoldoutBypass, renderMemoryContext, retrievalHoldoutBehaviorProb, + retrievalHoldoutConfigHash, toOffPolicyTrajectory, } from '../src/memory/index' @@ -75,6 +76,7 @@ describe('retrieval holdout: enabled behavior and event schema', () => { 'adapterId', 'callIndex', 'config', + 'configHash', 'corpusVersion', 'deliveredIds', 'droppedId', @@ -84,10 +86,9 @@ describe('retrieval holdout: enabled behavior and event schema', () => { 'holdoutEligible', 'pickPropensity', 'queryHash', - 'rngKey', - 'scope', + 'scopeHash', 'sessionHoldout', - 'sessionId', + 'sessionIdHash', 'sessionTargetId', 'taskId', 'ts', @@ -96,7 +97,9 @@ describe('retrieval holdout: enabled behavior and event schema', () => { ].sort(), ) expect(event.v).toBe(1) - expect(event.sessionId).toBe('s-1') + // Default-private: plaintext sessionId/scope are opt-in via includePlaintextIdentifiers. + expect(event.sessionId).toBeUndefined() + expect(event.scope).toBeUndefined() expect(event.taskId).toBe('task-17') expect(event.callIndex).toBe(1) expect(event.holdoutEligible).toBe(true) @@ -113,13 +116,17 @@ describe('retrieval holdout: enabled behavior and event schema', () => { { id: 'm7', rank: 2, score: 0.8, kind: 'fact', contentHash: sha256('beta').slice(0, 16) }, { id: 'm3', rank: 3, score: 0.7, kind: 'fact', contentHash: sha256('gamma').slice(0, 16) }, ]) - expect(event.rngKey).toBe(sha256('s-1').slice(0, 16)) + expect(event.sessionIdHash).toBe(sha256('s-1').slice(0, 16)) + expect(event.scopeHash).toBe(sha256('{"namespace":"ns","sessionId":"s-1"}').slice(0, 16)) expect(event.queryHash).toBe(sha256('the query').slice(0, 16)) expect(event.config).toEqual({ epsilon: 1, watchlist: ['m1', 'm7'], configVersion: '2026-07-04a', }) + expect(event.configHash).toBe( + retrievalHoldoutConfigHash({ epsilon: 1, watchlist: ['m1', 'm7'] }), + ) expect(event.adapterId).toBe('test-adapter') expect(event.corpusVersion).toBe('kb-test-1') // The manipulation is invisible: the rendered context renumbers with no gap (design rule D6). @@ -286,7 +293,8 @@ describe('retrieval holdout: adapter bypass paths still log the call', () => { expect(event.pickPropensity).toBeNull() expect(event.dropPropensity).toBeNull() expect(event.callIndex).toBe(0) - expect(event.sessionId).toBe('conv-1') + expect(event.sessionId).toBeUndefined() + expect(event.sessionIdHash).toBe(sha256('conv-1').slice(0, 16)) expect(event.taskId).toBe('task-3') expect(event.adapterId).toBe('neo4j-test') expect(event.watchlistEligible).toEqual(['obs-1']) @@ -324,7 +332,216 @@ describe('retrieval holdout: adapter bypass paths still log the call', () => { ]) expect(event.deliveredIds).toEqual([context.hits[0]!.id]) expect(event.sessionId).toBeUndefined() - expect(event.rngKey).toBeUndefined() + expect(event.sessionIdHash).toBeUndefined() + }) +}) + +describe('retrieval holdout: value-keyed session registry', () => { + const forcedRng = (key: string) => (key.endsWith('#pick') ? 0.6 : 0) + + it('keeps stickiness when a FRESH config object is built per call', async () => { + const events: RetrievalHoldoutEvent[] = [] + // The natural adapter pattern: options (and the holdout config inside them) built inline + // per retrieval. Identity-keyed state would miss on every call. + const freshConfig = (): RetrievalHoldoutConfig => ({ + epsilon: 1, + watchlist: ['m1', 'm3', 'm7'], + rng: forcedRng, + onEvent: (event) => events.push(event), + }) + + // Reviewer repro: call1 candidates sorted(watchlist ∩ E) = [m1, m7] → floor(0.6·2) = 1 → m7. + // A registry miss on call2 would re-draw over [m1, m3, m7] (floor(0.6·3) = 1 → m3), + // logging one session as under-treatment for two different items. + const call1 = await defaultGetMemoryContext( + { search: async () => [hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8)] }, + 'q1', + { scope: { sessionId: 's-fresh' }, holdout: freshConfig() }, + ) + const call2 = await defaultGetMemoryContext( + { + search: async () => [ + hit('m1', 'alpha', 0.9), + hit('m3', 'gamma', 0.7), + hit('m7', 'beta', 0.8), + ], + }, + 'q2', + { scope: { sessionId: 's-fresh' }, holdout: freshConfig() }, + ) + + expect(call1.hits.map((h) => h.id)).toEqual(['m1']) + expect(call2.hits.map((h) => h.id)).toEqual(['m1', 'm3']) + expect(events.map((e) => e.callIndex)).toEqual([1, 2]) + expect(events.map((e) => e.droppedId)).toEqual(['m7', 'm7']) + expect(events.map((e) => e.sessionTargetId)).toEqual(['m7', 'm7']) + expect(new Set(events.map((e) => e.configHash)).size).toBe(1) + }) + + it('computes configHash order-independently and only from experiment-defining knobs', () => { + expect(retrievalHoldoutConfigHash({ epsilon: 0.2, watchlist: ['b', 'a'] })).toBe( + retrievalHoldoutConfigHash({ epsilon: 0.2, watchlist: ['a', 'b'] }), + ) + expect(retrievalHoldoutConfigHash({ epsilon: 0.2, watchlist: ['a'] })).not.toBe( + retrievalHoldoutConfigHash({ epsilon: 0.3, watchlist: ['a'] }), + ) + expect(retrievalHoldoutConfigHash({ epsilon: 0.2 })).toBe( + retrievalHoldoutConfigHash({ epsilon: 0.2, watchlist: [] }), + ) + }) + + it('evicts oldest-first at the cap and re-drawn targets surface as mixed exposure', () => { + const events: RetrievalHoldoutEvent[] = [] + const config: RetrievalHoldoutConfig = { + epsilon: 1, + watchlist: ['e1', 'e5', 'e9'], + maxTrackedSessions: 2, + rng: (key) => (key.endsWith('#pick') ? 0.6 : 0), + onEvent: (event) => events.push(event), + } + // Candidates sorted(watchlist ∩ E) = [e1, e9] → floor(0.6·2) = 1 → e9. + const richHits = [hit('e1', 'one', 0.9), hit('e9', 'nine', 0.8)] + + const a1 = applySessionStickyRetrievalHoldout(richHits, config, { sessionId: 'evict-a' }) + expect(a1.event.sessionTargetId).toBe('e9') + applySessionStickyRetrievalHoldout(richHits, config, { sessionId: 'evict-b' }) + // Inserting C at cap 2 evicts A, the oldest-inserted session. + applySessionStickyRetrievalHoldout(richHits, config, { sessionId: 'evict-c' }) + + // B survived: its second call continues callIndex 2 with the same sticky target. + const b2 = applySessionStickyRetrievalHoldout(richHits, config, { sessionId: 'evict-b' }) + expect(b2.event.callIndex).toBe(2) + expect(b2.event.sessionTargetId).toBe('e9') + + // A returns after eviction with a narrower eligibility set: state was lost, so the target + // is re-drawn over [e1] alone — a different item than A's first exposure. + const a2 = applySessionStickyRetrievalHoldout([hit('e1', 'one', 0.9)], config, { + sessionId: 'evict-a', + }) + expect(a2.event.sessionTargetId).toBe('e1') + expect(a2.event.droppedId).toBe('e1') + + // The detectable mixed-exposure signature the analysis excludes and counts: + // same session hash, two different targets, callIndex reset to 1. + const aEvents = events.filter((e) => e.sessionIdHash === sha256('evict-a').slice(0, 16)) + expect(aEvents.map((e) => e.sessionTargetId)).toEqual(['e9', 'e1']) + expect(aEvents.map((e) => e.callIndex)).toEqual([1, 1]) + }) +}) + +describe('retrieval holdout: privacy defaults on event identifiers', () => { + it('emits plaintext sessionId and scope only with includePlaintextIdentifiers', () => { + const scope = { sessionId: 's-pii', userId: 'user-9', tags: { team: 'gtm' } } + const { config, events } = collectingConfig({ epsilon: 0, watchlist: ['m1'] }) + applyRetrievalHoldout([hit('m1', 'alpha', 0.9)], config, { scope }) + const { config: openConfig, events: openEvents } = collectingConfig({ + epsilon: 0, + watchlist: ['m1'], + includePlaintextIdentifiers: true, + }) + applyRetrievalHoldout([hit('m1', 'alpha', 0.9)], openConfig, { scope }) + + const priv = events[0]! + expect(priv.sessionId).toBeUndefined() + expect(priv.scope).toBeUndefined() + expect(priv.sessionIdHash).toBe(sha256('s-pii').slice(0, 16)) + expect(priv.scopeHash).toBe( + sha256('{"sessionId":"s-pii","tags":{"team":"gtm"},"userId":"user-9"}').slice(0, 16), + ) + + const open = openEvents[0]! + expect(open.sessionId).toBe('s-pii') + expect(open.scope).toEqual(scope) + // Hashes stay present either way, so both log shapes share the same join keys. + expect(open.sessionIdHash).toBe(priv.sessionIdHash) + expect(open.scopeHash).toBe(priv.scopeHash) + }) +}) + +describe('retrieval holdout: config validation fails loud', () => { + const onEvent = () => {} + + it('rejects epsilon outside [0, 1]', () => { + expect(() => + applyRetrievalHoldout( + [hit('m1', 'alpha', 0.9)], + { epsilon: 1.5, watchlist: ['m1'], onEvent }, + { sessionId: 's-v1' }, + ), + ).toThrow('epsilon must be a number in [0, 1]') + expect(() => applyRetrievalHoldout([], { epsilon: -0.1, onEvent })).toThrow( + 'epsilon must be a number in [0, 1]', + ) + expect(() => applyRetrievalHoldout([], { epsilon: Number.NaN, onEvent })).toThrow('epsilon') + }) + + it('rejects a watchlist that is not an array of strings', () => { + expect(() => + applyRetrievalHoldout([], { epsilon: 0.5, watchlist: [7 as unknown as string], onEvent }), + ).toThrow('watchlist must be an array of item-id strings') + expect(() => + emitRetrievalHoldoutBypass( + [], + { epsilon: 0.5, watchlist: 'm1' as unknown as string[], onEvent }, + {}, + 'raw-string-context', + ), + ).toThrow('watchlist must be an array of item-id strings') + }) + + it('validates before the sticky wrapper touches the registry', () => { + expect(() => + applySessionStickyRetrievalHoldout([], { epsilon: 2, onEvent }, { sessionId: 's-v2' }), + ).toThrow('epsilon must be a number in [0, 1]') + }) +}) + +describe('retrieval holdout: watchlist and score edge cases', () => { + it('with an empty or omitted watchlist the event is still emitted and nothing is ever dropped', () => { + const { config, events } = collectingConfig({ epsilon: 1, watchlist: [], rng: () => 0 }) + const hits = [hit('m1', 'alpha', 0.9), hit('m2', 'beta', 0.8)] + const { delivered, event } = applyRetrievalHoldout(hits, config, { sessionId: 's-empty' }) + + // Delivery identity is preserved on the no-drop path even in a holdout-arm session. + expect(delivered).toBe(hits) + expect(events).toHaveLength(1) + // The epsilon coin is independent of the watchlist, but no target can ever be drawn. + expect(event.sessionHoldout).toBe(true) + expect(event.sessionTargetId).toBeNull() + expect(event.droppedId).toBeNull() + expect(event.pickPropensity).toBeNull() + expect(event.watchlistEligible).toEqual([]) + expect(event.eligible.map((e) => e.id)).toEqual(['m1', 'm2']) + expect(event.deliveredIds).toEqual(['m1', 'm2']) + + const { config: omitted, events: omittedEvents } = collectingConfig({ + epsilon: 1, + rng: () => 0, + }) + const result = applyRetrievalHoldout(hits, omitted, { sessionId: 's-omitted' }) + expect(result.delivered).toBe(hits) + expect(omittedEvents).toHaveLength(1) + expect(omittedEvents[0]!.droppedId).toBeNull() + expect(omittedEvents[0]!.config.watchlist).toEqual([]) + }) + + it('falls back to the raw score when normalizedScore is absent', () => { + const { config } = collectingConfig({ epsilon: 0 }) + const scoreOnly: AgentMemoryHit = { + id: 'm-s', + uri: 'memory://test/m-s', + kind: 'fact', + text: 'raw', + score: 0.42, + } + const { event } = applyRetrievalHoldout([scoreOnly], config, { sessionId: 's-score' }) + expect(event.eligible[0]).toEqual({ + id: 'm-s', + rank: 1, + score: 0.42, + kind: 'fact', + contentHash: sha256('raw').slice(0, 16), + }) }) }) From e5cde50f43996bfd76646ee6f9796de35695f196 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Sat, 4 Jul 2026 17:05:10 -0600 Subject: [PATCH 4/4] =?UTF-8?q?fix(memory):=20session-level=20off-policy?= =?UTF-8?q?=20conversion=20=E2=80=94=20the=20session=20is=20the=20randomiz?= =?UTF-8?q?ation=20unit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-call converter treated each retrieval call as an independent draw, which is the wrong randomization unit for this design (arm coin and sticky target are drawn once per SESSION). Per-call propensities implied an action distribution summing past 1 (1.25 at epsilon 0.5, k=2) and biased IPS 33% low in the audit's 100-session sim. - toOffPolicyTrajectory now emits ONE OffPolicyTrajectory per (configHash, sessionIdHash): behaviorProb = 1−epsilon for observed control sessions, epsilon/|W∩E_first| for drop sessions (the draw event's logged dropPropensity), 1 when no call could drop; probabilities sum to 1 by construction (property-tested over epsilon×k grid). Rewards join per session via a rewards-by-sessionIdHash map; t>1 target-absent calls and bypass calls fold into the session summary instead of generating bogus independent propensities. Mixed-exposure sessions are excluded and surfaced, unattributable events counted. The invalid per-call retrievalHoldoutBehaviorProb export is removed. - Regression test reproduces the audit scenario: 50 control + 50 holdout at epsilon 0.5, k=2 — IPS now 1.000 (per-call baseline measured 0.667). - onEvent sink failures are logged and swallowed on BOTH the randomized and bypass paths — deliberately identical: observability must never break or alter retrieval. - canonicalJson: bigint fails loud (TypeError); NaN→null limitation documented. resetRetrievalHoldoutRegistry() exported for test isolation. - New tests: IPS-vs-SNIPS divergence under arm imbalance with mixed candidate counts, doublyRobust with per-session qHat, partial-batch and missing-reward fail-loud paths. --- src/memory/holdout.ts | 269 ++++++++++++++++++++----- tests/memory-holdout.test.ts | 371 ++++++++++++++++++++++++++--------- 2 files changed, 489 insertions(+), 151 deletions(-) diff --git a/src/memory/holdout.ts b/src/memory/holdout.ts index 6ce8439..f142e8b 100644 --- a/src/memory/holdout.ts +++ b/src/memory/holdout.ts @@ -145,7 +145,13 @@ export function deterministicRng(key: string): number { // Deterministic serialization for hashing: keys sorted, undefined-valued entries stripped, so // the same logical scope/config always produces the same hash regardless of construction order. +// Known limitation: NaN and Infinity serialize as null (JSON.stringify semantics) — acceptable +// for hashing because it is deterministic; bigint has no JSON form at all, so it fails loud. function canonicalJson(value: unknown): string { + if (typeof value === 'bigint') { + throw new TypeError('canonicalJson: bigint values are not supported in holdout hashing') + } + if (value === undefined) return 'null' if (Array.isArray(value)) return `[${value.map(canonicalJson).join(',')}]` if (value !== null && typeof value === 'object') { const entries = Object.entries(value as Record) @@ -185,6 +191,20 @@ function assertValidHoldoutConfig(config: RetrievalHoldoutConfig): void { } } +// Observability must never break retrieval: a throwing onEvent sink is logged and swallowed, +// deliberately identical on the randomized and bypass paths. The cost is a lost event — a hole +// the analysis can see as a callIndex gap — never a lost or altered retrieval result. +function safeEmit(config: RetrievalHoldoutConfig, event: RetrievalHoldoutEvent): void { + try { + config.onEvent(event) + } catch (error) { + console.error( + '[agent-knowledge] retrieval holdout onEvent sink threw; retrieval continues', + error, + ) + } +} + /** * Pure per-call holdout: takes post-filter hits, returns delivered hits plus the log event * and the next session state. Suppression only removes items (no backfill), and it happens @@ -248,7 +268,7 @@ export function applyRetrievalHoldout( dropPropensity: pickPropensity !== null ? config.epsilon * pickPropensity : null, deliveredIds: delivered.map((hit) => hit.id), } - config.onEvent(event) + safeEmit(config, event) return droppedId === null ? { delivered: hits, event, ...(session !== undefined ? { session } : {}) } @@ -324,7 +344,7 @@ export function emitRetrievalHoldoutBypass( deliveredIds: hits.map((hit) => hit.id), bypassReason, } - config.onEvent(event) + safeEmit(config, event) return event } @@ -340,6 +360,15 @@ const DEFAULT_MAX_TRACKED_SESSIONS = 10_000 // bounded by the number of distinct experiment configs, in practice a handful. const sessionRegistry = new Map>() +/** + * Clears all sticky-session state. For test isolation and experiment-epoch boundaries; + * sessions in flight afterwards re-draw deterministically (same rng keys), so with the default + * rng a reset is invisible in the logs unless eligibility sets changed in between. + */ +export function resetRetrievalHoldoutRegistry(): void { + sessionRegistry.clear() +} + /** * Convenience wrapper that threads session state internally, keyed by the VALUE of the * experiment-defining knobs (configHash = epsilon + sorted watchlist) plus sessionId, so a fresh @@ -377,73 +406,205 @@ export function applySessionStickyRetrievalHoldout( return result } +/** Aggregated view of one session's holdout exposure — the randomization unit of this design. */ +export interface RetrievalHoldoutSessionSummary { + /** `${configHash}:${sessionIdHash}` — the trajectory's runId. */ + runId: string + configHash: string + sessionIdHash: string + sessionHoldout: boolean + sessionTargetId: string | null + /** Item actually suppressed in this session (equals sessionTargetId when a drop occurred). */ + droppedId: string | null + /** |watchlist ∩ E| at the session's first intersecting randomized call; 0 when never intersecting. */ + firstCandidateCount: number + /** Randomized retrieval calls observed for this session. */ + callCount: number + /** Adapter bypass calls observed for this session (logged, never randomized). */ + bypassCallCount: number + /** Session-level probability of the OBSERVED assignment under the behavior policy. */ + behaviorProb: number + /** True when the session's events disagree on arm or target (e.g. registry eviction re-draw). */ + mixedExposure: boolean + /** Present only on sessions surfaced in `excluded` rather than converted. */ + exclusionReason?: 'mixed-exposure' | 'no-randomized-calls' +} + export interface RetrievalHoldoutOffPolicyOptions { - /** Realized outcome of the call's session/task, joined by the caller — events carry no reward. */ - reward: (event: RetrievalHoldoutEvent) => number /** - * Target-policy probability of the LOGGED delivery. Defaults to the "always deliver in full" - * policy: 1 when nothing was dropped, 0 when something was. + * Realized outcome PER SESSION, keyed by sessionIdHash — the session is the randomization + * unit, so rewards are per session, never per call. A missing entry for an included session + * throws: silently dropping unscored sessions would corrupt the estimator's denominator, so + * filter events to scored sessions upstream instead. + */ + rewards: Record + /** + * Target-policy probability of the session's observed assignment. Default models the + * always-deliver-in-full policy: 1 for full-delivery sessions, 0 for drop sessions. */ - targetProb?: (event: RetrievalHoldoutEvent) => number - /** Reward-model prediction enabling `doublyRobust`; when absent, entries carry qHat null. */ - qHat?: (event: RetrievalHoldoutEvent) => number | null + targetProb?: (session: RetrievalHoldoutSessionSummary) => number + /** Per-session reward-model prediction enabling `doublyRobust`; when absent, qHat is null. */ + qHat?: (session: RetrievalHoldoutSessionSummary) => number | null } -/** - * Behavior-policy probability of the delivery an event records, reconstructed from the logged - * propensities alone: - * - drop events: `dropPropensity` (= epsilon × pickPropensity, recorded at draw time); - * - no-drop events whose watchlist intersected the eligibility set: 1 − epsilon × pick, where - * pick is the logged `pickPropensity` (holdout arm, sticky draw) or 1/|watchlist ∩ E| - * (control arm — the uniform draw that would have happened, exact at the session's first - * intersecting call); - * - calls where no drop was possible (bypass paths, holdout-ineligible calls, empty - * watchlist ∩ E): 1, because full delivery was the only action the behavior policy could take. - */ -export function retrievalHoldoutBehaviorProb(event: RetrievalHoldoutEvent): number { - if (event.droppedId !== null) { - if (event.dropPropensity === null) - throw new Error(`holdout event ${event.eventId} recorded a drop without dropPropensity`) - return event.dropPropensity - } - if (!event.holdoutEligible) return 1 - const pick = - event.pickPropensity ?? - (event.watchlistEligible.length > 0 ? 1 / event.watchlistEligible.length : null) - return pick === null ? 1 : 1 - event.config.epsilon * pick +export interface RetrievalHoldoutOffPolicyResult { + /** One trajectory per (configHash, sessionIdHash) — never per call. */ + trajectories: OffPolicyTrajectory[] + /** 1:1 with `trajectories` (same order): the diagnostic view of each converted session. */ + sessions: RetrievalHoldoutSessionSummary[] + /** Sessions surfaced but NOT converted (mixed exposure, bypass-only) — counted, never hidden. */ + excluded: RetrievalHoldoutSessionSummary[] + /** Events with no sessionIdHash: outside session randomization, cannot join a reward. */ + unattributableEvents: number } /** - * Maps holdout log events onto agent-eval's `OffPolicyTrajectory` so EXP-007's analysis consumes - * the substrate's `inverseProbabilityWeighting` / `selfNormalizedImportanceWeighting` / - * `doublyRobust` estimators directly instead of re-deriving them from raw events. + * Maps holdout logs onto agent-eval's `OffPolicyTrajectory`, ONE TRAJECTORY PER SESSION, so + * EXP-007's analysis consumes the substrate's `inverseProbabilityWeighting` / + * `selfNormalizedImportanceWeighting` / `doublyRobust` estimators directly. This matches the + * PREREG estimator (session-sticky self-normalized IPW): the design randomizes per SESSION — + * the arm coin is flipped once (P(holdout) = epsilon) and the sticky target is drawn once, + * uniformly over watchlist ∩ E at the session's first intersecting call. * - * Field mapping (one trajectory per event): - * - `runId` ← `eventId` - * - `reward` ← `options.reward(event)` — joined by the caller, events carry no outcome - * - `behaviorProb` ← the event's logged pick/drop propensity ({@link retrievalHoldoutBehaviorProb}) - * - `targetProb` ← `options.targetProb(event)`; default models the always-deliver-in-full policy - * - `qHat` ← `options.qHat(event)`, else null (IPS/SNIPS-only) + * Session-level action space and behavior probabilities (they sum to 1 by construction): + * - full delivery (control arm observed): `1 − epsilon`; + * - drop candidate i of k = |watchlist ∩ E_first|: `epsilon / k` each (the draw event's logged + * `dropPropensity`); + * - sessions whose eligibility sets never intersect the watchlist: probability 1 — full + * delivery was certain in either arm. * - * Join keys: events carry `sessionIdHash` (= sha256(sessionId) first 16 hex) by default — compute - * the same prefix over the outcome table's session ids to join rewards, or set - * `includePlaintextIdentifiers: true` on the config to join on raw ids. Group and filter by - * `configHash` so trajectories from different experiment configs are never pooled. + * Per-call events are repeated observations WITHIN one session-level randomization; per-call + * IPW is statistically invalid for this design (per-call "propensities" imply an action + * distribution summing to more than 1 and bias IPS downward). Calls where the sticky target is + * absent from E, and adapter bypass calls, fold into the session summary (callCount / + * bypassCallCount) instead of generating independent propensities. * - * The per-event mapping treats calls independently; for a strict session-level estimand, - * aggregate to one trajectory per session (its first watchlist-intersecting call) upstream. + * Join contract: `rewards` is keyed by `sessionIdHash` (sha256(sessionId) first 16 hex — compute + * the same prefix over the outcome table's session ids, or log with + * `includePlaintextIdentifiers: true` and join on raw ids upstream). Events are grouped per + * (configHash, sessionIdHash) so different experiment configs are never pooled. Mixed-exposure + * sessions (arm or target disagreement, e.g. after registry eviction) are excluded from the + * trajectories and surfaced in `excluded` for the analysis to count. * Pre-registration for EXP-007 stays in the research repo by design; the manifest vocabulary for * it is agent-eval's `HypothesisManifest` / `signManifest` / `evaluateHypothesis` exports. */ export function toOffPolicyTrajectory( events: RetrievalHoldoutEvent[], options: RetrievalHoldoutOffPolicyOptions, -): OffPolicyTrajectory[] { - return events.map((event) => ({ - runId: event.eventId, - reward: options.reward(event), - behaviorProb: retrievalHoldoutBehaviorProb(event), - targetProb: options.targetProb?.(event) ?? (event.droppedId === null ? 1 : 0), - qHat: options.qHat?.(event) ?? null, - })) +): RetrievalHoldoutOffPolicyResult { + const groups = new Map() + let unattributableEvents = 0 + for (const event of events) { + if (event.sessionIdHash === undefined) { + unattributableEvents += 1 + continue + } + const key = `${event.configHash}:${event.sessionIdHash}` + const group = groups.get(key) + if (group) group.push(event) + else groups.set(key, [event]) + } + + const trajectories: OffPolicyTrajectory[] = [] + const sessions: RetrievalHoldoutSessionSummary[] = [] + const excluded: RetrievalHoldoutSessionSummary[] = [] + + for (const [runId, group] of groups) { + const ordered = [...group].sort((a, b) => a.callIndex - b.callIndex || a.ts.localeCompare(b.ts)) + const randomized = ordered.filter((event) => event.holdoutEligible) + const firstEvent = ordered[0] + if (firstEvent?.sessionIdHash === undefined) continue + const base = { + runId, + configHash: firstEvent.configHash, + sessionIdHash: firstEvent.sessionIdHash, + callCount: randomized.length, + bypassCallCount: ordered.length - randomized.length, + } + const first = randomized[0] + if (first === undefined) { + excluded.push({ + ...base, + sessionHoldout: false, + sessionTargetId: null, + droppedId: null, + firstCandidateCount: 0, + behaviorProb: 1, + mixedExposure: false, + exclusionReason: 'no-randomized-calls', + }) + continue + } + + const targets = new Set( + randomized + .map((event) => event.sessionTargetId) + .filter((target): target is string => target !== null), + ) + const arms = new Set(randomized.map((event) => event.sessionHoldout)) + const mixedExposure = targets.size > 1 || arms.size > 1 + const dropEvent = randomized.find((event) => event.droppedId !== null) + const firstIntersecting = randomized.find((event) => event.watchlistEligible.length > 0) + const sessionTargetId = dropEvent?.sessionTargetId ?? [...targets][0] ?? null + + let behaviorProb: number + if (dropEvent !== undefined) { + if (dropEvent.dropPropensity === null) { + throw new Error(`holdout event ${dropEvent.eventId} recorded a drop without dropPropensity`) + } + // epsilon / |watchlist ∩ E_first|, logged at draw time. + behaviorProb = dropEvent.dropPropensity + } else if (!mixedExposure && sessionTargetId !== null) { + // A drawn target implies its draw call dropped it, so a target with no drop event means + // the batch is missing that session's calls. Weighting partial sessions biases the + // estimate, so fail loud instead. + throw new Error( + `holdout session ${base.sessionIdHash} has a drawn target but no drop event: ` + + 'the event batch is missing this session draw call', + ) + } else if (firstIntersecting === undefined) { + // No call could ever drop anything: full delivery was certain in either arm. + behaviorProb = 1 + } else { + // Control arm observed with a drop possible: the session-level assignment probability + // is 1 − epsilon (NOT 1 − epsilon·pick, which is a per-item survival probability and + // makes the implied action distribution sum past 1). + behaviorProb = 1 - first.config.epsilon + } + + const summary: RetrievalHoldoutSessionSummary = { + ...base, + sessionHoldout: first.sessionHoldout, + sessionTargetId, + droppedId: dropEvent?.droppedId ?? null, + firstCandidateCount: firstIntersecting + ? new Set(firstIntersecting.watchlistEligible).size + : 0, + behaviorProb, + mixedExposure, + ...(mixedExposure ? { exclusionReason: 'mixed-exposure' as const } : {}), + } + if (mixedExposure) { + excluded.push(summary) + continue + } + + const reward = options.rewards[summary.sessionIdHash] + if (reward === undefined) { + throw new Error( + `no reward for session ${summary.sessionIdHash}: ` + + 'filter events to scored sessions before converting', + ) + } + trajectories.push({ + runId, + reward, + behaviorProb, + targetProb: options.targetProb?.(summary) ?? (summary.droppedId === null ? 1 : 0), + qHat: options.qHat?.(summary) ?? null, + }) + sessions.push(summary) + } + + return { trajectories, sessions, excluded, unattributableEvents } } diff --git a/tests/memory-holdout.test.ts b/tests/memory-holdout.test.ts index acf879b..c6f6756 100644 --- a/tests/memory-holdout.test.ts +++ b/tests/memory-holdout.test.ts @@ -1,8 +1,9 @@ import { + doublyRobust, inverseProbabilityWeighting, selfNormalizedImportanceWeighting, } from '@tangle-network/agent-eval/rl' -import { describe, expect, it } from 'vitest' +import { beforeEach, describe, expect, it, vi } from 'vitest' import { sha256 } from '../src/ids' import type { AgentMemoryHit, @@ -17,11 +18,16 @@ import { deterministicRng, emitRetrievalHoldoutBypass, renderMemoryContext, - retrievalHoldoutBehaviorProb, + resetRetrievalHoldoutRegistry, retrievalHoldoutConfigHash, toOffPolicyTrajectory, } from '../src/memory/index' +/** sessionIdHash as events carry it: sha256(sessionId) first 16 hex. */ +function sid(sessionId: string): string { + return sha256(sessionId).slice(0, 16) +} + function hit(id: string, text: string, score: number): AgentMemoryHit { return { id, uri: `memory://test/${id}`, kind: 'fact', text, normalizedScore: score } } @@ -339,6 +345,10 @@ describe('retrieval holdout: adapter bypass paths still log the call', () => { describe('retrieval holdout: value-keyed session registry', () => { const forcedRng = (key: string) => (key.endsWith('#pick') ? 0.6 : 0) + beforeEach(() => { + resetRetrievalHoldoutRegistry() + }) + it('keeps stickiness when a FRESH config object is built per call', async () => { const events: RetrievalHoldoutEvent[] = [] // The natural adapter pattern: options (and the holdout config inside them) built inline @@ -456,6 +466,41 @@ describe('retrieval holdout: privacy defaults on event identifiers', () => { expect(open.sessionIdHash).toBe(priv.sessionIdHash) expect(open.scopeHash).toBe(priv.scopeHash) }) + + it('canonical hashing rejects bigint values instead of hashing a lossy form', () => { + const { config } = collectingConfig({ epsilon: 0 }) + expect(() => + applyRetrievalHoldout([], config, { + scope: { tags: { n: 1n as unknown as string } }, + }), + ).toThrow(TypeError) + }) +}) + +describe('retrieval holdout: onEvent sink failures never break retrieval', () => { + it('logs and continues when the sink throws, identically on randomized and bypass paths', () => { + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}) + try { + const config: RetrievalHoldoutConfig = { + epsilon: 1, + watchlist: ['m1'], + rng: () => 0, + onEvent: () => { + throw new Error('sink down') + }, + } + const hits = [hit('m1', 'alpha', 0.9), hit('m2', 'beta', 0.8)] + const { delivered, event } = applyRetrievalHoldout(hits, config, { sessionId: 's-sink' }) + // Suppression still applied and the event still returned: only persistence was lost. + expect(delivered.map((h) => h.id)).toEqual(['m2']) + expect(event.droppedId).toBe('m1') + const bypass = emitRetrievalHoldoutBypass(hits, config, { query: 'q' }, 'raw-string-context') + expect(bypass.bypassReason).toBe('raw-string-context') + expect(errorSpy).toHaveBeenCalledTimes(2) + } finally { + errorSpy.mockRestore() + } + }) }) describe('retrieval holdout: config validation fails loud', () => { @@ -545,11 +590,11 @@ describe('retrieval holdout: watchlist and score edge cases', () => { }) }) -describe('retrieval holdout: off-policy conversion for agent-eval estimators', () => { +describe('retrieval holdout: session-level off-policy conversion', () => { // Forces holdout (epsilon 1) and picks index floor(0.6 * |candidates|) deterministically. const forcedRng = (key: string) => (key.endsWith('#pick') ? 0.6 : 0) - it('maps drop, keep, and control events onto OffPolicyTrajectory propensities', () => { + it('emits ONE trajectory per session; t>1 target-absent calls fold in without a propensity', () => { const { config, events } = collectingConfig({ epsilon: 1, watchlist: ['m1', 'm7'], @@ -558,134 +603,266 @@ describe('retrieval holdout: off-policy conversion for agent-eval estimators', ( const drop = applyRetrievalHoldout([hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8)], config, { sessionId: 's-ope', }) - // Same session, sticky target m7 absent from E: nothing dropped, propensity still logged. + // Second call of the SAME session with the sticky target absent from E: this is a repeated + // observation within one randomization, never an independent draw. applyRetrievalHoldout([hit('m3', 'gamma', 0.7)], config, { sessionId: 's-ope', session: drop.session, }) - const { config: controlConfig, events: controlEvents } = collectingConfig({ - epsilon: 0.2, - watchlist: ['m1', 'm7'], - // Coin 0.99 >= epsilon: control arm, no pick is ever drawn. - rng: () => 0.99, - }) - applyRetrievalHoldout( - [hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8), hit('m3', 'gamma', 0.7)], - controlConfig, - { sessionId: 's-ctl' }, - ) - - const all = [...events, ...controlEvents] - const trajectories = toOffPolicyTrajectory(all, { - reward: (event) => (event.droppedId === null ? 1 : 0), - }) - // Drop event: behaviorProb = dropPropensity = epsilon * pick = 1 * (1/2). - expect(trajectories[0]).toEqual({ - runId: all[0]!.eventId, + const result = toOffPolicyTrajectory(events, { rewards: { [sid('s-ope')]: 0 } }) + expect(result.trajectories).toHaveLength(1) + expect(result.trajectories[0]).toEqual({ + runId: `${events[0]!.configHash}:${sid('s-ope')}`, reward: 0, + // Session-level P(drop m7) = epsilon / |watchlist ∩ E_first| = 1/2, the logged + // draw-time dropPropensity. behaviorProb: 0.5, targetProb: 0, qHat: null, }) - // Keep event in the holdout arm: 1 - epsilon * logged sticky pickPropensity. - expect(trajectories[1]).toEqual({ - runId: all[1]!.eventId, - reward: 1, - behaviorProb: 0.5, - targetProb: 1, - qHat: null, + expect(result.sessions[0]).toMatchObject({ + callCount: 2, + bypassCallCount: 0, + droppedId: 'm7', + sessionTargetId: 'm7', + firstCandidateCount: 2, + mixedExposure: false, }) - // Control event: no draw was logged; counterfactual uniform pick over |watchlist ∩ E| = 2. - expect(trajectories[2]!.behaviorProb).toBeCloseTo(1 - 0.2 * 0.5, 12) - expect(trajectories[2]!.targetProb).toBe(1) - expect(trajectories[2]!.reward).toBe(1) + expect(result.excluded).toHaveLength(0) + expect(result.unattributableEvents).toBe(0) }) - it('assigns behaviorProb 1 whenever no drop was possible', () => { - // Holdout-ineligible call: no sessionId, no randomization happened. - const { config } = collectingConfig({ epsilon: 1, watchlist: ['m1'], rng: () => 0 }) - const ineligible = applyRetrievalHoldout([hit('m1', 'alpha', 0.9)], config, {}).event - expect(retrievalHoldoutBehaviorProb(ineligible)).toBe(1) - - // Control session whose eligibility set never intersects the watchlist. - const { config: missConfig } = collectingConfig({ + it('assigns control sessions behaviorProb 1 − epsilon, the session-level assignment probability', () => { + const { config, events } = collectingConfig({ epsilon: 0.2, - watchlist: ['m-absent'], + watchlist: ['m1', 'm7'], + // Coin 0.99 >= epsilon: control arm. rng: () => 0.99, }) - const miss = applyRetrievalHoldout([hit('m3', 'gamma', 0.7)], missConfig, { - sessionId: 's-miss', - }).event - expect(retrievalHoldoutBehaviorProb(miss)).toBe(1) - - // Adapter bypass event: the call never reached retrieval. - const { config: bypassConfig } = collectingConfig({ epsilon: 1, watchlist: ['m1'] }) - const bypass = emitRetrievalHoldoutBypass( - [hit('m1', 'alpha', 0.9)], - bypassConfig, - { query: 'q', scope: { sessionId: 's-bypass' } }, - 'short-term-context', + applyRetrievalHoldout( + [hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8), hit('m3', 'gamma', 0.7)], + config, + { sessionId: 's-ctl' }, ) - expect(retrievalHoldoutBehaviorProb(bypass)).toBe(1) + const { trajectories } = toOffPolicyTrajectory(events, { rewards: { [sid('s-ctl')]: 1 } }) + // NOT 1 − epsilon·pick (a per-item survival probability): P(control) = 1 − epsilon. + expect(trajectories[0]!.behaviorProb).toBeCloseTo(0.8, 12) + expect(trajectories[0]!.targetProb).toBe(1) + expect(trajectories[0]!.reward).toBe(1) + }) + + it('property: session-level action probabilities sum to 1 over the (epsilon, k) grid', () => { + for (const epsilon of [0.1, 0.5, 0.9]) { + for (const k of [1, 2, 5]) { + const ids = Array.from({ length: k }, (_, i) => `w${i}`) + const hits = ids.map((id, i) => hit(id, `text-${id}`, 1 - i * 0.01)) + const rewards: Record = {} + const events: RetrievalHoldoutEvent[] = [] + const mkConfig = (rig: (key: string) => number): RetrievalHoldoutConfig => ({ + epsilon, + watchlist: ids, + rng: rig, + onEvent: (event) => events.push(event), + }) + // One control session plus k holdout sessions realizing each drop action once + // (the pick rig lands on candidate i exactly: floor(((i + 0.5) / k) · k) = i). + const ctlSession = `grid-ctl-${epsilon}-${k}` + applyRetrievalHoldout( + hits, + mkConfig(() => 0.999999), + { sessionId: ctlSession }, + ) + rewards[sid(ctlSession)] = 1 + for (let i = 0; i < k; i += 1) { + const holdoutSession = `grid-h-${epsilon}-${k}-${i}` + applyRetrievalHoldout( + hits, + mkConfig((key) => (key.endsWith('#pick') ? (i + 0.5) / k : 0)), + { sessionId: holdoutSession }, + ) + rewards[sid(holdoutSession)] = 1 + } + + const { trajectories, sessions } = toOffPolicyTrajectory(events, { rewards }) + expect(trajectories).toHaveLength(k + 1) + const drops = sessions.filter((s) => s.droppedId !== null) + // All k drop actions enumerated, each with probability epsilon / k. + expect(new Set(drops.map((s) => s.droppedId)).size).toBe(k) + for (const s of drops) expect(s.behaviorProb).toBeCloseTo(epsilon / k, 12) + const control = sessions.find((s) => s.droppedId === null) + expect(control!.behaviorProb).toBeCloseTo(1 - epsilon, 12) + const total = control!.behaviorProb + drops.reduce((acc, s) => acc + s.behaviorProb, 0) + expect(total).toBeCloseTo(1, 12) + } + } }) - it('fails loud on a corrupt drop event missing its propensity', () => { - const { config } = collectingConfig({ epsilon: 1, watchlist: ['m1'], rng: () => 0 }) - const { event } = applyRetrievalHoldout([hit('m1', 'alpha', 0.9)], config, { - sessionId: 's-bad', + it('regression: IPS recovers 1.000 on the audit 100-session scenario (per-call gave 0.667)', () => { + const rewards: Record = {} + const { config, events } = collectingConfig({ + epsilon: 0.5, + watchlist: ['m1', 'm7'], + rng: (key) => (key.includes(':h:') ? (key.endsWith('#pick') ? 0.6 : 0) : 0.99), }) - expect(() => retrievalHoldoutBehaviorProb({ ...event, dropPropensity: null })).toThrow( - 'without dropPropensity', - ) + for (let i = 0; i < 50; i += 1) { + const holdoutSession = `sim:h:${i}` + const controlSession = `sim:c:${i}` + applyRetrievalHoldout([hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8)], config, { + sessionId: holdoutSession, + }) + applyRetrievalHoldout([hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8)], config, { + sessionId: controlSession, + }) + rewards[sid(holdoutSession)] = 0 + rewards[sid(controlSession)] = 1 + } + + const { trajectories, sessions } = toOffPolicyTrajectory(events, { rewards }) + expect(trajectories).toHaveLength(100) + expect(sessions.filter((s) => s.droppedId !== null)).toHaveLength(50) + const ips = inverseProbabilityWeighting(trajectories) + const snips = selfNormalizedImportanceWeighting(trajectories) + expect(ips.n).toBe(100) + // True always-deliver value is 1: controls weigh 1/(1−ε) = 2, drops weigh 0. + // The per-call converter weighed controls 1/(1−ε·pick) = 4/3 and measured 0.667 here. + expect(ips.value).toBeCloseTo(1, 12) + expect(snips.value).toBeCloseTo(1, 12) + expect(ips.maxImportanceWeight).toBeCloseTo(2, 12) }) - it('honors custom targetProb and qHat callbacks', () => { - const { config, events } = collectingConfig({ epsilon: 1, watchlist: ['m1'], rng: () => 0 }) - applyRetrievalHoldout([hit('m1', 'alpha', 0.9)], config, { sessionId: 's-custom' }) + it('IPS and SNIPS diverge under arm imbalance with mixed candidate counts; DR recovers with a truthful qHat', () => { + const rewards: Record = {} + const events: RetrievalHoldoutEvent[] = [] + const cfg = (rig: (key: string) => number): RetrievalHoldoutConfig => ({ + epsilon: 0.5, + watchlist: ['m1', 'm7'], + rng: rig, + onEvent: (event) => events.push(event), + }) + for (let i = 0; i < 4; i += 1) { + const s = `div-c-${i}` + applyRetrievalHoldout( + [hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8)], + cfg(() => 0.99), + { sessionId: s }, + ) + rewards[sid(s)] = 1 + } + // Two holdout sessions with DIFFERENT candidate-set sizes: k=2 → ε/k = 0.25, k=1 → 0.5. + applyRetrievalHoldout( + [hit('m1', 'alpha', 0.9), hit('m7', 'beta', 0.8)], + cfg((key) => (key.endsWith('#pick') ? 0.6 : 0)), + { sessionId: 'div-h-k2' }, + ) + rewards[sid('div-h-k2')] = 0 + applyRetrievalHoldout( + [hit('m7', 'beta', 0.8)], + cfg(() => 0), + { sessionId: 'div-h-k1' }, + ) + rewards[sid('div-h-k1')] = 0 + + const { trajectories, sessions } = toOffPolicyTrajectory(events, { rewards }) + expect( + sessions + .filter((s) => s.droppedId !== null) + .map((s) => s.behaviorProb) + .sort((a, b) => a - b), + ).toEqual([0.25, 0.5]) + const ips = inverseProbabilityWeighting(trajectories) + const snips = selfNormalizedImportanceWeighting(trajectories) + // 4 controls at weight 2, 2 drops at weight 0: Σw·r = 8 over n = 6 vs Σw = 8. + expect(ips.value).toBeCloseTo(8 / 6, 12) + expect(snips.value).toBeCloseTo(1, 12) + expect(Math.abs(ips.value - snips.value)).toBeGreaterThan(0.2) + // Doubly-robust with a truthful per-session qHat is exact despite the imbalance. + const dr = doublyRobust(toOffPolicyTrajectory(events, { rewards, qHat: () => 1 }).trajectories) + expect(dr.value).toBeCloseTo(1, 12) + }) - const [trajectory] = toOffPolicyTrajectory(events, { - reward: () => 0.75, - targetProb: () => 0.25, + it('honors custom targetProb and qHat callbacks per session', () => { + const { config, events } = collectingConfig({ + epsilon: 0.2, + watchlist: ['m1'], + rng: () => 0.99, + }) + applyRetrievalHoldout([hit('m1', 'alpha', 0.9)], config, { sessionId: 's-cb' }) + const { trajectories } = toOffPolicyTrajectory(events, { + rewards: { [sid('s-cb')]: 0.75 }, + targetProb: (session) => (session.droppedId === null ? 0.25 : 0), qHat: () => 0.5, }) - expect(trajectory).toEqual({ - runId: events[0]!.eventId, + expect(trajectories[0]).toMatchObject({ reward: 0.75, - behaviorProb: 1, + behaviorProb: 0.8, targetProb: 0.25, qHat: 0.5, }) }) - it('feeds converted logs into agent-eval IPS/SNIPS end to end', () => { - // 10 holdout + 10 control sessions at epsilon 0.5 with a single watchlist item: - // drops have behaviorProb 0.5, keeps 0.5, so the always-deliver target policy - // (weight 0 on drops, 2 on keeps) recovers the delivered-arm mean reward exactly. + it('excludes mixed-exposure and bypass-only sessions, surfacing them with reasons', () => { const { config, events } = collectingConfig({ - epsilon: 0.5, - watchlist: ['m1'], - rng: (key) => (key.includes('holdout-session') ? 0 : 0.99), + epsilon: 1, + watchlist: ['e1', 'e9'], + rng: (key) => (key.endsWith('#pick') ? 0.6 : 0), }) - for (let i = 0; i < 10; i += 1) { - applyRetrievalHoldout([hit('m1', 'alpha', 0.9), hit('m2', 'beta', 0.8)], config, { - sessionId: `holdout-session-${i}`, - }) - applyRetrievalHoldout([hit('m1', 'alpha', 0.9), hit('m2', 'beta', 0.8)], config, { - sessionId: `control-session-${i}`, - }) - } - const trajectories = toOffPolicyTrajectory(events, { - reward: (event) => (event.droppedId === null ? 1 : 0), + // Two calls of one session WITHOUT threaded state and with different eligibility sets: + // fresh draws land on different targets — the mixed-exposure signature. + applyRetrievalHoldout([hit('e1', 'one', 0.9), hit('e9', 'nine', 0.8)], config, { + sessionId: 's-mixed', }) + applyRetrievalHoldout([hit('e1', 'one', 0.9)], config, { sessionId: 's-mixed' }) + // A session that only ever hit adapter bypass paths. + emitRetrievalHoldoutBypass( + [hit('m1', 'alpha', 0.9)], + config, + { query: 'q', scope: { sessionId: 's-bypass-only' } }, + 'short-term-context', + ) + // An event with no session at all. + applyRetrievalHoldout([hit('e1', 'one', 0.9)], config, {}) + + const result = toOffPolicyTrajectory(events, { rewards: {} }) + expect(result.trajectories).toHaveLength(0) + expect(result.excluded.map((s) => s.exclusionReason).sort()).toEqual([ + 'mixed-exposure', + 'no-randomized-calls', + ]) + const mixed = result.excluded.find((s) => s.exclusionReason === 'mixed-exposure') + expect(mixed).toMatchObject({ mixedExposure: true, sessionIdHash: sid('s-mixed') }) + const bypassOnly = result.excluded.find((s) => s.exclusionReason === 'no-randomized-calls') + expect(bypassOnly).toMatchObject({ callCount: 0, bypassCallCount: 1 }) + expect(result.unattributableEvents).toBe(1) + }) - expect(events.filter((e) => e.droppedId !== null)).toHaveLength(10) - const ips = inverseProbabilityWeighting(trajectories) - const snips = selfNormalizedImportanceWeighting(trajectories) - expect(ips.n).toBe(20) - expect(ips.value).toBeCloseTo(1, 12) - expect(snips.value).toBeCloseTo(1, 12) - expect(ips.maxImportanceWeight).toBeCloseTo(2, 12) + it('fails loud when an included session has no reward', () => { + const { config, events } = collectingConfig({ epsilon: 0, watchlist: ['m1'] }) + applyRetrievalHoldout([hit('m1', 'alpha', 0.9)], config, { sessionId: 's-unscored' }) + expect(() => toOffPolicyTrajectory(events, { rewards: {} })).toThrow('no reward for session') + }) + + it('fails loud on a corrupt drop event and on a batch missing the draw call', () => { + const { config, events } = collectingConfig({ epsilon: 1, watchlist: ['m1'], rng: () => 0 }) + const drop = applyRetrievalHoldout([hit('m1', 'alpha', 0.9)], config, { + sessionId: 's-corrupt', + }) + const corrupted = events.map((event) => ({ ...event, dropPropensity: null })) + expect(() => toOffPolicyTrajectory(corrupted, { rewards: { [sid('s-corrupt')]: 0 } })).toThrow( + 'without dropPropensity', + ) + + // Batch containing only the t=2 target-absent call: the drawn target proves the draw call + // is missing, and weighting the partial session would bias the estimate. + const { event: absentCall } = applyRetrievalHoldout([hit('m3', 'gamma', 0.7)], config, { + sessionId: 's-corrupt', + session: drop.session, + }) + expect(absentCall.sessionTargetId).toBe('m1') + expect(absentCall.droppedId).toBeNull() + expect(() => + toOffPolicyTrajectory([absentCall], { rewards: { [sid('s-corrupt')]: 0 } }), + ).toThrow('missing this session draw call') }) })