diff --git a/docs-site/src/content/docs/guides/codex-integration.md b/docs-site/src/content/docs/guides/codex-integration.md index 57373b06ec..f960e459ae 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/docs-site/src/content/docs/guides/codex-integration.md @@ -939,13 +939,41 @@ or remove it to disable capture. The file is owner-only (`0600`) in the hardened and rolls after 200 lines, retaining the newest 100. Each JSONL record contains the protocol, routed provider/model, cache-counter presence and -provenance, process-local equality tags for the account, prompt-cache key, and allowlisted session -headers, plus ordered fingerprints for instructions, tools, and message/input blocks. Prefix -sections retain at most 128 tags and identify only the first divergent section/index. The -diagnostic never stores prompt or message text, tool names, raw headers, raw cache/session/account -identifiers, or a durable tag derived from them. Its random HMAC key is created at process start, -separate from other debug keys, and is never persisted; tags therefore compare values only within -one proxy process. +provenance, and process-local equality tags for the account, prompt-cache key, session headers, +and ordered instruction/tool/message blocks. It never stores prompt or message text, tool names, +raw headers, or raw cache/session/account identifiers. The random HMAC key is separate from other +debug keys, created at process start and never persisted, so tags compare only within one process. + +- `session.inboundHeaders` and `outboundHeaders` record `session_id`, `session-id`, `thread-id`, + and `x-session-id` independently. Identical values have identical tags across these names and + directions. An unobserved side is omitted, distinct from observed absent headers. These headers + are diagnostic correlation inputs, not proof of account affinity. +- `accountAffinity` records a separately tagged key from the shared Codex conversation-identity + derivation for each observed header set. `derived-key-not-binding` means exactly that: a key + is not evidence that a binding was stored, selected, or reused. Reported route labels are separate. +- `prefix.comparison` compares the observed inbound and outbound projections. `continuation` + compares each side with the previous finalized request having the same complete nonempty inbound + session-header identity, authenticated caller scope, protocol, model, and config directory. The + caller scope comes from the authenticated admission captured before Chat/Messages bridging, not + from rewritten upstream headers. It is used only in private in-memory lookup keys, never serialized. + The same admitted key has the same scope across supported authentication headers; different keys + remain isolated. Without an authenticated admission identity (including unauthenticated loopback), + continuation comparisons are `uncomparable`; per-request inbound/outbound comparisons remain available. + Provider changes do not reset this + comparison; prompt-key and explicit account-label changes are reported separately. Outbound-only + session identifiers and prompt-cache keys never supply a missing correlation identity. +- Comparisons report `first`, `equal`, `append-only`, `different`, `truncated`, or `uncomparable`. + Only message-tail growth is append-only. Top-level instructions and system content precede tools; + input/message rows, including system/developer updates, stay in their original order. Missing + observations and unsupported shapes are uncomparable, never assumed equal to an empty prefix. +- Each section retains at most 128 tags. An unobserved common tail reports `truncated`, not equality + or an exact first divergence. At most 512 recent correlation histories are retained in memory; + an evicted history is `first` on its next observation. + +These are parsed-body observations at existing request hooks, not captured network bytes or proof +of upstream cache reuse. Concurrent requests compare in finalization order, not send order. The +switch changes diagnostics only, not affinity selection or cache behavior; the existing debug-mode +bridge requirement for otherwise eligible direct Chat/Messages lanes still applies. ### Catalog troubleshooting diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index e2ec2449df..eec02af2d1 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -294,7 +294,7 @@ "bun-runtime.test.ts": "ci-workflows", "bun-transpiler-cache-scope.test.ts": "cli", "bun-stream-caps.test.ts": "lib", - "cache-diagnostic.test.ts": "usage", + "cache-diagnostic.test.ts": "usage", "cache-diagnostic-caller-isolation.test.ts": "usage", "cache-diagnostic-hook-compat.test.ts": "usage", "cancel-body-on-abort.test.ts": "server", "catalog-auto-refresh-client-fanout.test.ts": "codex-integration", "catalog-auto-refresh-scheduler.test.ts": "codex-integration", "catalog-cursor-search.test.ts": "codex-integration", diff --git a/src/server/request-log.ts b/src/server/request-log.ts index cf8feddc5b..a908b162e9 100644 --- a/src/server/request-log.ts +++ b/src/server/request-log.ts @@ -98,7 +98,7 @@ const CACHE_DIAGNOSTIC_HOOK = Symbol.for("opencodex.cache-diagnostic.v1"); interface CacheDiagnosticHooks { /** Whether inbound bodies are actually recorded; the hook is registered at import either way. */ enabled(): boolean; - observeInbound(body: unknown, headers: Headers, source: PromptCacheKeySource): CacheDiagnosticDraft; + observeInbound(body: unknown, headers: Headers, source: PromptCacheKeySource, admissionPrincipal?: string): CacheDiagnosticDraft; rebind(body: unknown, draft: CacheDiagnosticDraft | undefined): void; finalize(facts: CacheDiagnosticFinalFacts): void; } @@ -280,8 +280,9 @@ export function observeCacheDiagnosticInbound( body: unknown, headers: Headers, source: PromptCacheKeySource, + admissionPrincipal?: string, ): void { - const draft = cacheDiagnosticHooks()?.observeInbound(body, headers, source); + const draft = cacheDiagnosticHooks()?.observeInbound(body, headers, source, admissionPrincipal); if (draft) logCtx.cacheDiagnosticDraft = draft; } diff --git a/src/server/responses/request-prepare.ts b/src/server/responses/request-prepare.ts index 9830fef7e0..9fd471481c 100644 --- a/src/server/responses/request-prepare.ts +++ b/src/server/responses/request-prepare.ts @@ -36,7 +36,7 @@ import { sessionIdHeaderFromRequest, reasoningReplayConversationIdFromResponsesRequest, } from "../request-log-conversation"; -import { resolveContextPrincipal } from "../auth-cors"; +import { contextPrincipalIdOf, resolveContextPrincipal } from "../auth-cors"; import { resolveSkillsSnapshotScopeKey, snapshotSkillsCatalogInBody } from "./skills-snapshot"; import { isShadowSourceModel, @@ -249,6 +249,7 @@ export async function prepareResponsesRequest( options.promptCacheKeyIsSharedCohort === true ? "system-derived" : options.promptCacheKeyIsSharedCohort === false ? "metadata-derived" : "caller", + contextPrincipalIdOf(options.admission), ); if (!concreteSelection && !options.compactionRoutingOverride && inboundWire === "responses") { options.compactionRoutingOverride = applyCompactionRoutingOverride(body, req.headers, config, { diff --git a/src/usage/cache-diagnostic.ts b/src/usage/cache-diagnostic.ts index 14918c6056..7824301183 100644 --- a/src/usage/cache-diagnostic.ts +++ b/src/usage/cache-diagnostic.ts @@ -2,7 +2,7 @@ import { createHmac, randomBytes } from "node:crypto"; import { appendFileSync, chmodSync, existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; import { join } from "node:path"; -import { CODEX_AFFINITY_DEBUG_SAFE_HEADERS } from "../codex/affinity-debug"; +import { codexConversationIdentity } from "../codex/lineage"; import { getConfigDir } from "../config"; import { recordOwnedConfigPath } from "../lib/config-ownership"; import type { CacheTelemetryProvenance } from "./log"; @@ -10,6 +10,8 @@ import type { CacheTelemetryProvenance } from "./log"; const CACHE_DEBUG_KEY = randomBytes(32); const MAX_BLOCKS = 128; const MAX_DRAFTS = 512; +const SESSION_HEADERS = ["session_id", "session-id", "thread-id", "x-session-id"] as const; +type SessionHeaders = Record; export const CACHE_DEBUG_MAX_LINES = 200; export const CACHE_DEBUG_KEEP_LINES = 100; @@ -20,11 +22,24 @@ export interface PrefixFingerprint { instructions: TaggedSequence; tools: TaggedSequence; messages: TaggedSequence; + unsupported?: true; } +type PrefixSection = "instructions" | "tools" | "messages"; +export type PrefixComparison = + | { status: "first" | "equal" } + | { status: "append-only" | "different" | "truncated"; section: PrefixSection; index: number } + | { status: "uncomparable"; reason: "missing-session" | "missing-observation" | "unsupported-shape" }; export interface CacheDiagnosticDraft { promptCacheKey?: { inbound?: TaggedPresence; outbound?: TaggedPresence }; - session?: { inboundHeader?: TaggedPresence; outboundHeader?: TaggedPresence }; + session?: { + inboundHeader?: TaggedPresence; + outboundHeader?: TaggedPresence; + inboundHeaders?: SessionHeaders; + outboundHeaders?: SessionHeaders; + }; prefix?: { inbound?: PrefixFingerprint; outbound?: PrefixFingerprint }; + /** The shared Pool key derivation's result, not a binding lookup or selection decision. */ + accountAffinity?: { inbound?: TaggedPresence; outbound?: TaggedPresence }; } export interface CacheDiagnosticFinalFacts { @@ -49,6 +64,16 @@ export interface CacheDiagnosticFinalFacts { const drafts = new Map(); const bodyDrafts = new WeakMap(); +// Internal only: neither the caller principal nor its tag belongs in the diagnostic record. +const callerScopes = new WeakMap(); +interface PreviousObservation { + prefix?: CacheDiagnosticDraft["prefix"]; + promptCacheKey?: CacheDiagnosticDraft["promptCacheKey"]; + account: TaggedPresence; +} +// Finalization order, process-local only. Keys include the diagnostic destination so changing +// OPENCODEX_HOME cannot join histories. Neither keys nor values retain raw session/account data. +const previousObservations = new Map(); export function isCacheDiagnosticEnabled(): boolean { return process.env.OPENCODEX_CACHE_DEBUG === "1"; @@ -79,14 +104,27 @@ function taggedPresence(value: unknown, domain: string, source?: PromptCacheKeyS return { present: true, tag: tag(domain, value), ...(source ? { source } : {}) }; } -function sessionPresence(headers: Headers | HeadersInit): TaggedPresence { +function accountAffinityKey(headers: HeadersInit): TaggedPresence { + // codexPoolAffinityKey is exactly this read-only identity projection. Import its leaf + // derivation, not auth-context (which brings credential resolution and routing with it). + // Parent-only inputs may consult existing TTL-bounded lineage; this call records nothing. + const normalized = headers instanceof Headers ? headers : new Headers(headers); + return taggedPresence(codexConversationIdentity(normalized)?.conversationKey, "cache-debug:codex-pool-affinity-key"); +} + +function sessionHeaders(headers: HeadersInit): SessionHeaders { const normalized = headers instanceof Headers ? headers : new Headers(headers); - const values = CODEX_AFFINITY_DEBUG_SAFE_HEADERS.flatMap(name => { + return Object.fromEntries(SESSION_HEADERS.map(name => { const value = normalized.get(name); - return value === null ? [] : [[name, value] as const]; - }); - return values.length === 0 - ? { present: false } + return [name, value === null ? { present: false } : { + present: true, tag: tag("cache-debug:session-header", value), + }]; + })) as SessionHeaders; +} + +function sessionPresence(headers: SessionHeaders): TaggedPresence { + const values = SESSION_HEADERS.filter(name => headers[name].present).map(name => [name, headers[name].tag]); + return values.length === 0 ? { present: false } : { present: true, tag: tag("cache-debug:session-headers", canonicalJson(values)) }; } @@ -102,25 +140,21 @@ function sequence(blocks: unknown[], domain: string): TaggedSequence { function requestBlocks(body: Record): { instructions: unknown[]; messages: unknown[] } { const instructions = body.instructions; - // The spread is load-bearing: an array-valued instructions field must be copied, never - // aliased, because the pushes below would otherwise mutate the live request body that - // the adapter is about to serialize upstream. - const instructionRows = instructions === undefined || instructions === null - ? [] : Array.isArray(instructions) ? [...instructions] : [instructions]; - const input = Array.isArray(body.input) ? body.input : Array.isArray(body.messages) ? body.messages : []; - const messages: unknown[] = []; - for (const block of input) { - if (block && typeof block === "object" && !Array.isArray(block)) { - const row = block as Record; - if (row.role === "system" || row.role === "developer") { - const content = row.content; - instructionRows.push(...(Array.isArray(content) ? content : [content])); - continue; - } - } - messages.push(block); + // Build fresh rows: pushes below must never mutate an array-valued instructions field + // in the live request body. Include the source so moving text between fields is visible. + const instructionRows: unknown[] = instructions === undefined || instructions === null + ? [] : (Array.isArray(instructions) ? instructions : [instructions]) + .map(content => ({ source: "instructions", content })); + if (body.system !== undefined && body.system !== null) { + instructionRows.push(...(Array.isArray(body.system) ? body.system : [body.system]) + .map(content => ({ source: "system", content }))); } - return { instructions: instructionRows, messages }; + const input = typeof body.input === "string" ? [body.input] + : Array.isArray(body.input) ? body.input : Array.isArray(body.messages) ? body.messages : []; + // Keep every input/messages row in wire order, including system/developer updates. + // Extracting those roles into instructions misclassifies a genuine tail append as a + // top-level prefix rewrite. Whole-row tags still detect role, position and content edits. + return { instructions: instructionRows, messages: input }; } export function prefixFingerprint(body: unknown): PrefixFingerprint { @@ -128,30 +162,91 @@ export function prefixFingerprint(body: unknown): PrefixFingerprint { ? body as Record : {}; const tools = Array.isArray(record.tools) ? record.tools : []; const blocks = requestBlocks(record); + const unsupported = !body || typeof body !== "object" || Array.isArray(body) + || (record.input !== undefined && typeof record.input !== "string" && !Array.isArray(record.input)) + || (record.messages !== undefined && !Array.isArray(record.messages)) + || (record.tools !== undefined && !Array.isArray(record.tools)); return { instructions: sequence(blocks.instructions, "cache-debug:prefix:instructions"), tools: sequence(tools, "cache-debug:prefix:tools"), messages: sequence(blocks.messages, "cache-debug:prefix:messages"), + ...(unsupported ? { unsupported: true as const } : {}), }; } -function firstDivergence(inbound: PrefixFingerprint, outbound: PrefixFingerprint): - { section: "instructions" | "tools" | "messages"; index: number } | undefined { +function comparePrefixes(before?: PrefixFingerprint, after?: PrefixFingerprint): PrefixComparison { + if (!before || !after) return { status: "uncomparable", reason: "missing-observation" }; + if (before.unsupported || after.unsupported) return { status: "uncomparable", reason: "unsupported-shape" }; for (const section of ["instructions", "tools", "messages"] as const) { - const before = inbound[section].tags; - const after = outbound[section].tags; - const compared = Math.min(before.length, after.length); + const left = before[section]; + const right = after[section]; + const compared = Math.min(left.tags.length, right.tags.length); for (let index = 0; index < compared; index += 1) { - if (before[index] !== after[index]) return { section, index }; + if (left.tags[index] !== right.tags[index]) return { status: "different", section, index }; + } + // A hidden difference here could precede any difference in later sections. Do not + // infer equality, append-only, or an exact first divergence from an unobserved tail. + if (compared < Math.min(left.count, right.count)) return { status: "truncated", section, index: compared }; + if (left.count !== right.count) { + // Only message-tail growth is append-only. Growing instructions/tools changes the + // cache prefix before the conversation even when all existing section tags match. + return { status: section === "messages" && right.count > left.count ? "append-only" : "different", section, index: compared }; } - if (inbound[section].count !== outbound[section].count) return { section, index: compared }; } - return undefined; + return { status: "equal" }; +} + +type PresenceComparison = "first" | "equal" | "changed" | "uncomparable"; +function comparePresence(before?: TaggedPresence, after?: TaggedPresence): PresenceComparison { + if (!before || !after) return "uncomparable"; + return before.present === after.present && before.tag === after.tag ? "equal" : "changed"; +} + +function continuation(facts: CacheDiagnosticFinalFacts, draft: CacheDiagnosticDraft, account: TaggedPresence) { + const headers = draft.session?.inboundHeaders; + // Diagnostic correlation ONLY, not the account affinity identity or proof of a binding. + // All available inbound identifiers participate. Do not use prompt_cache_key (which may + // legitimately rotate), synthesized outbound IDs, or unrelated affinity headers as identity. + const identity = headers && SESSION_HEADERS.filter(name => headers[name].present + && headers[name].tag !== tag("cache-debug:session-header", "")) + .map(name => [name, headers[name].tag]); + const callerScope = callerScopes.get(draft); + const key = identity?.length && callerScope ? tag("cache-debug:history", canonicalJson([ + getConfigDir(), callerScope, facts.protocol, facts.model, identity, + ])) : undefined; + const previous = key ? previousObservations.get(key) : undefined; + const compare = (side: "inbound" | "outbound"): PrefixComparison => { + if (!identity?.length) return { status: "uncomparable", reason: "missing-session" }; + if (!callerScope) return { status: "uncomparable", reason: "missing-observation" }; + const current = draft.prefix?.[side]; + if (!current) return { status: "uncomparable", reason: "missing-observation" }; + if (current.unsupported) return { status: "uncomparable", reason: "unsupported-shape" }; + return previous ? comparePrefixes(previous.prefix?.[side], current) : { status: "first" }; + }; + const presence = (side: "inbound" | "outbound"): PresenceComparison => !key || !draft.promptCacheKey?.[side] ? "uncomparable" + : !previous ? "first" : comparePresence(previous.promptCacheKey?.[side], draft.promptCacheKey?.[side]); + const result = { + inbound: compare("inbound"), + outbound: compare("outbound"), + promptCacheKey: { inbound: presence("inbound"), outbound: presence("outbound") }, + account: !key ? "uncomparable" : !previous ? "first" : comparePresence(previous.account, account), + }; + if (key) { + previousObservations.delete(key); + previousObservations.set(key, { + prefix: { ...draft.prefix }, promptCacheKey: { ...draft.promptCacheKey }, account, + }); + while (previousObservations.size > MAX_DRAFTS) previousObservations.delete(previousObservations.keys().next().value!); + } + return result; } export function observe(requestId: string, observation: Partial): CacheDiagnosticDraft { + if (!isCacheDiagnosticEnabled()) return {}; const draft = drafts.get(requestId) ?? {}; Object.assign(draft, observation); + const callerScope = callerScopes.get(observation); + if (callerScope) callerScopes.set(draft, callerScope); drafts.delete(requestId); drafts.set(requestId, draft); while (drafts.size > MAX_DRAFTS) drafts.delete(drafts.keys().next().value!); @@ -162,16 +257,22 @@ export function observeInbound( body: unknown, headers: Headers, source: PromptCacheKeySource = "caller", + admissionPrincipal?: string, ): CacheDiagnosticDraft { if (!isCacheDiagnosticEnabled()) return {}; try { const record = body && typeof body === "object" && !Array.isArray(body) ? body as Record : {}; + const inboundHeaders = sessionHeaders(headers); const draft: CacheDiagnosticDraft = { promptCacheKey: { inbound: taggedPresence(record.prompt_cache_key, "cache-debug:prompt-cache-key", source) }, - session: { inboundHeader: sessionPresence(headers) }, + session: { inboundHeader: sessionPresence(inboundHeaders), inboundHeaders }, prefix: { inbound: prefixFingerprint(body) }, + accountAffinity: { inbound: accountAffinityKey(headers) }, }; + // Admission owns caller identity; bridges may remove or replace Authorization. + // No trusted principal (including unauthenticated loopback) means no history join. + if (admissionPrincipal) callerScopes.set(draft, tag("cache-debug:caller-scope", admissionPrincipal)); if (body && typeof body === "object") bodyDrafts.set(body, draft); return draft; } catch { @@ -208,8 +309,10 @@ export function observeOutbound( outbound.source = inboundKey.source; } (draft.promptCacheKey ??= {}).outbound = outbound; - (draft.session ??= {}).outboundHeader = sessionPresence(headers); + const outboundHeaders = sessionHeaders(headers); + Object.assign(draft.session ??= {}, { outboundHeader: sessionPresence(outboundHeaders), outboundHeaders }); (draft.prefix ??= {}).outbound = prefixFingerprint(outboundBody); + (draft.accountAffinity ??= {}).outbound = accountAffinityKey(headers); } catch { /* diagnostics must never affect request handling */ } @@ -236,8 +339,11 @@ export function appendFinalCacheDiagnostic(facts: CacheDiagnosticFinalFacts): vo drafts.delete(facts.requestId); const inboundKey = draft.promptCacheKey?.inbound; const outboundKey = draft.promptCacheKey?.outbound; - const inboundPrefix = draft.prefix?.inbound ?? prefixFingerprint(undefined); - const outboundPrefix = draft.prefix?.outbound ?? prefixFingerprint(undefined); + const inboundPrefix = draft.prefix?.inbound; + const outboundPrefix = draft.prefix?.outbound; + const comparison = comparePrefixes(inboundPrefix, outboundPrefix); + // Only explicit account facts are evidence; never infer an account from provider text. + const account = taggedPresence(facts.accountLogLabel, "cache-debug:account-log-label"); const record = { version: 1 as const, ts: Date.now(), @@ -251,23 +357,33 @@ export function appendFinalCacheDiagnostic(facts: CacheDiagnosticFinalFacts): vo outbound: outboundKey ?? { present: false }, ...(inboundKey?.present && outboundKey?.present ? { equal: inboundKey.tag === outboundKey.tag } : {}), }, + accountAffinity: { + derivation: "codexConversationIdentity", + evidence: "derived-key-not-binding", + // Missing observation stays omitted, distinct from an observed absent key. + ...draft.accountAffinity, + }, session: { + purpose: "diagnostic-correlation-only", inboundHeader: draft.session?.inboundHeader ?? { present: false }, outboundHeader: draft.session?.outboundHeader ?? { present: false }, + inboundHeaders: draft.session?.inboundHeaders, + outboundHeaders: draft.session?.outboundHeaders, }, prefix: { inbound: inboundPrefix, outbound: outboundPrefix, - ...(firstDivergence(inboundPrefix, outboundPrefix) - ? { firstDivergentBlock: firstDivergence(inboundPrefix, outboundPrefix) } - : {}), + comparison, + ...(comparison.status === "different" || comparison.status === "append-only" + ? { firstDivergentBlock: { section: comparison.section, index: comparison.index } } : {}), }, + continuation: continuation(facts, draft, account), route: { provider: facts.provider, model: facts.model, - ...(facts.accountLogLabel - ? { accountTag: tag("cache-debug:account-log-label", facts.accountLogLabel) } - : {}), + ...(account.present ? { accountTag: account.tag } : {}), + // Preserve reported route labels, never infer key presence or an actual binding + // from them: even `new_bind` can be reported when no affinity key was derived. ...(facts.affinityMove ? { affinityMove: facts.affinityMove } : {}), ...(facts.affinityReason ? { affinityReason: facts.affinityReason } : {}), }, @@ -295,7 +411,7 @@ export function appendFinalCacheDiagnostic(facts: CacheDiagnosticFinalFacts): vo const CACHE_DIAGNOSTIC_HOOK = Symbol.for("opencodex.cache-diagnostic.v1"); interface CacheDiagnosticHooks { enabled(): boolean; - observeInbound(body: unknown, headers: Headers, source: PromptCacheKeySource): CacheDiagnosticDraft; + observeInbound(body: unknown, headers: Headers, source: PromptCacheKeySource, admissionPrincipal?: string): CacheDiagnosticDraft; rebind(body: unknown, draft: CacheDiagnosticDraft | undefined): void; finalize(facts: CacheDiagnosticFinalFacts): void; } diff --git a/structure/dashboard-and-usage.md b/structure/dashboard-and-usage.md index e38d47e0fa..1e040e180b 100644 --- a/structure/dashboard-and-usage.md +++ b/structure/dashboard-and-usage.md @@ -395,15 +395,15 @@ once it exceeds 200) with the upstream content-type, body kind (`sse / json / ot body sample, and the extracted usage. Off by default; the hot path is guarded so production stays untouched. -For diagnosing cache-read instability without capturing content, set `OPENCODEX_CACHE_DEBUG=1` -before start. `src/usage/cache-diagnostic.ts` then writes one record per finalized request to -`~/.opencodex/cache-debug.jsonl` (same `0o600` file, same 200-to-100 rolling bound) holding only -presence booleans, counts, closed enums, the raw upstream cache counter before defaulting, and -process-local HMAC equality tags for the prompt-cache key, allowlisted session headers, the account -log label, and ordered instruction/tool/message blocks (capped at 128 per section, first divergent -section/index only). The signing key is created at process start and never persisted, so tags -compare values within one proxy process and never become a durable correlation key; no prompt -text, tool name, raw identifier, or header value is recorded. Off by default. +For content-free cache-read diagnostics, `OPENCODEX_CACHE_DEBUG=1` enables `src/usage/cache-diagnostic.ts`; otherwise hooks return before body/header inspection, hashing or file access. Its live `enabled()` hook remains registered. +It writes one finalized request to `~/.opencodex/cache-debug.jsonl` (mode `0o600`, 200-to-100 rolling lines), preserving raw cache-counter evidence. +Process-local HMAC tags cover prompt keys, explicit account labels and ordered instructions/tools/messages; no prompt, tool name or raw identifier is retained, and keys never persist. +Session-header tags cover `session_id`, `session-id`, `thread-id` and `x-session-id`, sharing a value domain but not merging header identities; an unobserved side is omitted. `accountAffinity` uses read-only `codexConversationIdentity` and states `derived-key-not-binding`; neither a key nor a route label proves a binding. +`prefix.comparison` compares observed sides; `continuation` compares their previous finalization under the complete nonempty inbound session-header set, authenticated admission principal, protocol, model and config directory. `src/server/responses/request-prepare.ts` carries the admission identity through the request-log hook; only its private diagnostic HMAC is retained, never serialized or reconstructed from bridge headers. +Provider changes preserve correlation; prompt-key and explicit account-label changes are separate. Synthesized outbound headers and prompt keys never replace missing inbound identity. +Comparisons distinguish first/equal/message-tail append/different/truncated/uncomparable; instructions and tools grow before messages, while system/developer input rows retain wire order. +Missing or unsupported observations are uncomparable. Missing admission identity, including unauthenticated loopback, disables cross-request history while preserving per-request comparisons. Sections retain 128 tags and never claim equality across hidden tails; history evicts beyond 512 entries. +These are parsed hook projections in completion order, not exact network bytes or proof of cache reuse; `tests/usage/cache-diagnostic.test.ts`, `tests/usage/cache-diagnostic-hook-compat.test.ts` and `tests/usage/cache-diagnostic-caller-isolation.test.ts` pin the contract. ## Z.ai quota destination ownership diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 02f6eb0a8b..45ed8abe0e 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -12,7 +12,7 @@ When a successful streamed native response has a missing or unrecognized non-JSO ## Responses HTTP/SSE HTTP ingress applies the [caller header contract](../data-planes/inbound-compat.md#http-caller-conversation-identity) before admission and dispatch. Optional Codex memory selection enters `src/server/responses/request-prepare.ts` for HTTP and WebSocket frames, as specified in [memory phase routing](responses-failover.md#memory-phase-routing); an unset phase keeps its existing route and starts no background work. Responses request preparation stabilizes incoming `` under `skills.catalog_refresh`: `per_session` (default) reuses the first received catalog for a conversation; `per_turn` leaves the supplied catalog unchanged. Other instruction sections and user/tool content remain untouched. Requests without a reliable conversation identity bypass snapshots; shared prompt-cache cohorts are not conversation identities. Only a body with exactly one catalog block across its instructions and developer/system content takes part; two or more pass through unchanged. A known snapshot is substituted before parsing, but a new catalog is stored only when preparation reaches its success return, so a request rejected by parsing or admission pins nothing. Without a named principal, snapshots are shared by conversation id only on a server that requires no data-plane auth. Snapshots are process-local, expire after four idle hours, and use bounded LRU retention; oversized blocks bypass caching. The dashboard's `src/codex/prompt-layers.ts` and `src/codex/prompt-text-probe.ts` continue observing current files for previews and do not own session snapshots. -`/v1/responses` is the main Codex-facing endpoint. The server parses Responses input, routes to a +Cache continuation diagnostics carry the authenticated admission principal out of band through Chat/Messages bridges under the [private diagnostic contract](../dashboard-and-usage.md#opt-in-aggregate-request-metrics), never from rewritten headers. `/v1/responses` is the main Codex-facing endpoint. The server parses Responses input, routes to a provider, lets the selected adapter speak the upstream protocol, then bridges adapter events back to Responses-compatible streaming output. Every Responses serializer projects unreferenced hosted-search history when the final body declares no hosted search tool; with search declared, the canonical Codex ordinary serializer retains structural hosted history by default, and only selected-provider `projectResponsesLiteHostedSearchHistory === true` plus exact canonical `/responses` and explicit Lite intent enables model-independent projection under the [hosted-search history boundary](../providers-and-adapters.md#hosted-search-continuation-binding), shared by HTTP/SSE and WS. The routed provider name follows every adapter build, including retries and continuations; only `github-copilot` injects a configured `modelContextTiers` value as upstream `contextTier`, while unconfigured passthrough retains the caller field. For an opted-in key-auth provider, a hosted-search continuation stays bound to the API-key selection that served the first leg; the contract is the [hosted-search continuation binding](../providers-and-adapters.md#hosted-search-continuation-binding). The native passthrough undeclared-tool guard fails closed on the Responses inbound wire in SSE and buffered JSON before continuation state is stored. Chat and Anthropic inbounds defer to the caller's own registry except an explicit empty catalog; the [declared-tool contract](responses-wire-shapes.md#declared-tool-membership-by-inbound-wire) also covers managed native, translated, directly encoded, and final media output while preserving internal hosted operations. GitHub Copilot Auto admission rechecks the negotiated provider/model against the key scope captured at request admission (a key edited or removed mid-request cannot widen it) before route mutation or dispatch, on initial and renewal negotiation; denial returns the standard `model_not_allowed_for_key` 403 naming the caller's inbound selector without disclosing the destination. The web-search and image sidecars answer a denial during their credential rotation with that 403 instead of the original 429, and a combo child's denial keeps the same envelope. Local model-scope refusals are non-replayable. A denial during Copilot dispatch renewal propagates as an admission exception to the request boundary, never as a provider fetch response; account recovery and combo/policy fallback cannot continue after that denial. A committed continuation, including OAuth account rotation after a 429, emits the same refusal type in-stream without dispatching another inference; ordinary negotiation failures still preserve the original provider refusal. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index fed4bb7ebc..6ce4a52cfc 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -295,7 +295,7 @@ "build-desktop-icon-set.test.ts": "ci-workflows", "build-release-changelog.test.ts": "ci-workflows", "ci-gui-typecheck-gate.test.ts": "ci-workflows", "bump-dev-version.test.ts": "ci-workflows", "bun-runtime.test.ts": "ci-workflows", "bun-transpiler-cache-scope.test.ts": "cli", - "bun-stream-caps.test.ts": "lib", "cache-diagnostic.test.ts": "usage", + "bun-stream-caps.test.ts": "lib", "cache-diagnostic.test.ts": "usage", "cache-diagnostic-caller-isolation.test.ts": "usage", "cache-diagnostic-hook-compat.test.ts": "usage", "cancel-body-on-abort.test.ts": "server", "catalog-auto-refresh-client-fanout.test.ts": "codex-integration", "catalog-auto-refresh-scheduler.test.ts": "codex-integration", "catalog-cursor-search.test.ts": "codex-integration", "catalog-custom-ordering.test.ts": "codex-integration", "catalog-duplicate-slug-dedup.test.ts": "codex-integration", diff --git a/tests/usage/cache-diagnostic-caller-isolation.test.ts b/tests/usage/cache-diagnostic-caller-isolation.test.ts new file mode 100644 index 0000000000..c305b0eff7 --- /dev/null +++ b/tests/usage/cache-diagnostic-caller-isolation.test.ts @@ -0,0 +1,274 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, readFileSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { contextPrincipalIdOf, requireResponsesApiAuth, resolveApiAuth, resolveResponsesApiAuth } from "../../src/server/auth-cors"; +import { handleChatCompletions } from "../../src/server/chat-completions"; +import { handleClaudeMessages } from "../../src/server/claude-messages"; +import { handleResponses } from "../../src/server/responses/core"; +import { tryAdmitTurn } from "../../src/server/lifecycle"; +import { addFinalRequestLog, type RequestLogContext } from "../../src/server/request-log"; +import { closeRequestHistoryIndex } from "../../src/routing/history/indexer"; +import { clearResponseStateForTests, flushResponseState } from "../../src/responses/state"; +import { cacheDiagnosticPath, observeInbound, tagCacheDiagnosticValue, type PrefixComparison } from "../../src/usage/cache-diagnostic"; +import type { OcxConfig } from "../../src/types"; +import { responsesSuccess } from "../helpers/combo-failover-upstream"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +type Protocol = "responses" | "chat" | "messages"; +type CredentialHeader = "authorization" | "x-opencodex-api-key" | "x-api-key"; +interface DiagnosticRow { + protocol: Protocol; + continuation: { inbound: PrefixComparison; outbound: PrefixComparison }; +} +const CALLER_A = "synthetic-diagnostic-caller-a"; +const CALLER_B = "synthetic-diagnostic-caller-b"; +const PROVIDER_KEY = "synthetic-diagnostic-provider"; +const UNRELATED_BEARERS = ["synthetic-unrelated-bearer-one", "synthetic-unrelated-bearer-two"]; +const FORGED_PRINCIPAL = "synthetic-forged-principal"; +const HOOK = Symbol.for("opencodex.cache-diagnostic.v1"); +type DiagnosticHook = { observeInbound: typeof observeInbound } & Record; +const hookRegistry = globalThis as unknown as Record; +let registeredHook: DiagnosticHook; +let home: string; +let codex: IsolatedCodexHome; +let releaseSpend: () => void; +let previousHome: string | undefined; +let previousDebug: string | undefined; +let originalFetch: typeof fetch; +let networkCalls: number; +let inboundHeaders: Headers[]; +let wireHeaders: Headers[]; +let wireBodies: unknown[]; +let principals: Set; +let serializedContexts: string[]; + +beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + previousDebug = process.env.OPENCODEX_CACHE_DEBUG; + home = mkdtempSync(join(tmpdir(), "ocx-cache-caller-")); + process.env.OPENCODEX_HOME = home; + process.env.OPENCODEX_CACHE_DEBUG = "1"; + codex = installIsolatedCodexHome("ocx-cache-caller-codex-"); + releaseSpend = acquireOwnedSpendHome(); + networkCalls = 0; + inboundHeaders = []; + wireHeaders = []; + wireBodies = []; + principals = new Set(); + serializedContexts = []; + originalFetch = globalThis.fetch; + const rejectNetwork = () => { + networkCalls++; + throw new Error("Network forbidden in cache caller isolation regression"); + }; + globalThis.fetch = Object.assign(async () => rejectNetwork(), { preconnect: rejectNetwork }); + // Observe the real bridge seam without replacing auth, handlers, adapters, or diagnostics. + registeredHook = hookRegistry[HOOK]!; + hookRegistry[HOOK] = { ...registeredHook, observeInbound(...args) { + inboundHeaders.push(new Headers(args[1])); + return registeredHook.observeInbound(...args); + } }; +}); + +afterEach(async () => { + releaseSpend(); + closeRequestHistoryIndex(); + await flushResponseState(); + clearResponseStateForTests(); + hookRegistry[HOOK] = registeredHook; + globalThis.fetch = originalFetch; + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (previousDebug === undefined) delete process.env.OPENCODEX_CACHE_DEBUG; + else process.env.OPENCODEX_CACHE_DEBUG = previousDebug; + codex.restore(); + removeTreeWithRetry(home); +}); + +function config(): OcxConfig { + return { + port: 0, hostname: "0.0.0.0", defaultProvider: "fixture", + apiKeys: [CALLER_A, CALLER_B].map((key, index) => ({ + id: `fixture-caller-${index}`, name: `Fixture caller ${index}`, key, createdAt: "2026-01-01", + })), + providers: { fixture: { + adapter: "openai-responses", baseUrl: "https://cache-fixture.invalid/v1", authMode: "key", + apiKey: PROVIDER_KEY, models: ["model"], liveModels: false, + fetch: (async (_url: RequestInfo | URL, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)); + wireHeaders.push(new Headers(init?.headers)); + wireBodies.push(body); + return Response.json(responsesSuccess("synthetic cache answer", "model")); + }) as typeof fetch, + } }, + claudeCode: { enabled: true }, + }; +} + +function rows(): DiagnosticRow[] { + return readFileSync(cacheDiagnosticPath(), "utf8").trim().split("\n").map(line => JSON.parse(line)); +} + +async function send(protocol: Protocol, cfg: OcxConfig, key: string, header: CredentialHeader, + appended = false, extraHeaders: Record = {}): Promise { + const messages = [{ role: "user", content: "synthetic shared prefix" }, + ...(appended ? [{ role: "assistant", content: "synthetic first answer" }, + { role: "user", content: "synthetic next turn" }] : [])]; + const body = { model: "fixture/model", stream: false, max_tokens: 32, + ...(protocol === "responses" ? { input: messages } : { messages }) }; + const req = new Request(`http://localhost/v1/${protocol === "chat" ? "chat/completions" : protocol}`, { + method: "POST", headers: { "content-type": "application/json", session_id: "synthetic-shared-session", + [header]: header === "authorization" ? `Bearer ${key}` : key, ...extraHeaders }, body: JSON.stringify(body), + }); + const admission = (protocol === "messages" ? resolveApiAuth : resolveResponsesApiAuth)(req, cfg); + expect(admission).not.toBeNull(); + const principal = contextPrincipalIdOf(admission!); + if (principal) principals.add(principal); + const requestId = crypto.randomUUID(); + const start = Date.now(); + // These are the same log fields the HTTP ingress owns before invoking each handler. + const log: RequestLogContext = { model: "", provider: "", inboundProtocol: protocol }; + const before = wireBodies.length; + const turnAdmissionLease = tryAdmitTurn()!; + expect(turnAdmissionLease).not.toBeNull(); + try { + const response = protocol === "responses" + ? await handleResponses(req, cfg, log, { admission: admission!, abortSignal: req.signal, turnAdmissionLease }) + : await (protocol === "chat" ? handleChatCompletions : handleClaudeMessages)( + req, cfg, log, { requestId, start, admission: admission!, turnAdmissionLease }); + const text = await response.text(); + expect(response.status).toBe(200); + expect(text).toContain("synthetic cache answer"); + // Responses' HTTP route owns finalization; Chat/Messages finalize their own bridge log. + if (protocol === "responses") addFinalRequestLog(requestId, start, log, response.status); + expect(wireBodies).toHaveLength(before + 1); + expect(rows()).toHaveLength(before + 1); + serializedContexts.push(JSON.stringify(log)); + return principal; + } finally { + turnAdmissionLease.release(); + } +} + +function expectPrivacy(): void { + const persisted = readFileSync(cacheDiagnosticPath(), "utf8"); + const contexts = serializedContexts.join("\n"); + const forwarded = JSON.stringify(wireHeaders.map(headers => Object.fromEntries(headers))); + for (const secret of [CALLER_A, CALLER_B, ...UNRELATED_BEARERS, FORGED_PRINCIPAL, ...principals]) { + expect(persisted).not.toContain(secret); + expect(contexts).not.toContain(secret); + expect(forwarded).not.toContain(secret); + expect(JSON.stringify(wireBodies)).not.toContain(secret); + } + for (const principal of principals) { + expect(persisted).not.toContain(tagCacheDiagnosticValue("caller-scope", principal)); + expect(contexts).not.toContain(tagCacheDiagnosticValue("caller-scope", principal)); + expect(forwarded).not.toContain(tagCacheDiagnosticValue("caller-scope", principal)); + } + for (const headers of wireHeaders) { + expect(headers.get("authorization")).toBe(`Bearer ${PROVIDER_KEY}`); + expect(headers.has("x-opencodex-api-key")).toBe(false); + expect(headers.has("x-api-key")).toBe(false); + expect([...headers.keys()].some(name => /principal|caller|diagnostic/i.test(name))).toBe(false); + } + expect(networkCalls).toBe(0); +} + +describe("cache diagnostic authenticated caller isolation", () => { + for (const protocol of ["responses", "chat", "messages"] as const) { + const acceptedHeaders: CredentialHeader[] = protocol === "messages" + ? ["authorization", "x-opencodex-api-key", "x-api-key"] : ["authorization", "x-opencodex-api-key"]; + for (const header of acceptedHeaders) { + test(`${protocol}: ${header} isolates interleaved callers and preserves identity across header aliases`, async () => { + const cfg = config(); + const a = await send(protocol, cfg, CALLER_A, header); + const b = await send(protocol, cfg, CALLER_B, header); + expect(a).toBeDefined(); + expect(b).toBeDefined(); + expect(a).not.toBe(b); + const alias = header === "authorization" ? "x-opencodex-api-key" : "authorization"; + expect(await send(protocol, cfg, CALLER_A, alias)).toBe(a); + expect(await send(protocol, cfg, CALLER_A, header, true)).toBe(a); + expect(await send(protocol, cfg, CALLER_B, alias)).toBe(b); + for (const side of ["inbound", "outbound"] as const) { + expect(rows().map(row => row.continuation[side].status)).toEqual([ + "first", "first", "equal", "append-only", "equal", + ]); + } + expect(rows().map(row => row.protocol)).toEqual(Array(5).fill(protocol)); + if (protocol !== "responses") { + expect(inboundHeaders).toHaveLength(5); + for (const headers of inboundHeaders) expect(headers.has("authorization")).toBe(false); + } + expectPrivacy(); + }); + } + + test(`${protocol}: unrelated bearer and forged identity headers cannot override the admitted dedicated key`, async () => { + const cfg = config(); + const forged = { authorization: `Bearer ${UNRELATED_BEARERS[0]}`, + "x-opencodex-context-principal": FORGED_PRINCIPAL }; + const a = await send(protocol, cfg, CALLER_A, "x-opencodex-api-key", false, forged); + const b = await send(protocol, cfg, CALLER_B, "x-opencodex-api-key", false, forged); + expect(a).not.toBe(b); + expect(await send(protocol, cfg, CALLER_A, "x-opencodex-api-key", false, { + ...forged, authorization: `Bearer ${UNRELATED_BEARERS[1]}`, + })).toBe(a); + for (const side of ["inbound", "outbound"] as const) { + expect(rows().map(row => row.continuation[side].status)).toEqual(["first", "first", "equal"]); + } + expectPrivacy(); + }); + + if (protocol !== "messages") { + test(`${protocol}: x-api-key alone remains rejected by its real ingress resolver`, () => { + const req = new Request(`http://localhost/v1/${protocol === "chat" ? "chat/completions" : protocol}`, { + method: "POST", headers: { "x-api-key": CALLER_A, session_id: "synthetic-shared-session" }, + }); + expect(resolveResponsesApiAuth(req, config())).toBeNull(); + expect(requireResponsesApiAuth(req, config())?.status).toBe(401); + expect(wireBodies).toHaveLength(0); + expect(inboundHeaders).toHaveLength(0); + expect(networkCalls).toBe(0); + }); + } + + test(`${protocol}: loopback admission never treats unverified credential headers as caller identity`, async () => { + const cfg = { ...config(), hostname: "127.0.0.1" }; + for (const header of ["authorization", "x-opencodex-api-key", "x-api-key"] as const) { + expect(await send(protocol, cfg, CALLER_A, header)).toBeUndefined(); + expect(await send(protocol, cfg, CALLER_A, header)).toBeUndefined(); + } + for (const row of rows()) { + expect(row.continuation.inbound.status).toBe("uncomparable"); + expect(row.continuation.outbound.status).toBe("uncomparable"); + } + expectPrivacy(); + }); + } + + for (const protocol of ["chat", "messages"] as const) { + test(`${protocol}: changing synthetic stored-main Authorization cannot replace the admission principal`, async () => { + const cfg = config(); + const first = await send(protocol, cfg, CALLER_A, "authorization"); + const replacements = ["synthetic-bridge-main-one", "synthetic-bridge-main-two"]; + for (const access of replacements) { + writeFileSync(join(codex.path, "auth.json"), JSON.stringify({ + tokens: { access_token: access, account_id: "synthetic-bridge-main-account" }, + })); + expect(await send(protocol, cfg, CALLER_A, "authorization")).toBe(first); + } + expect(inboundHeaders.map(headers => headers.get("authorization"))).toEqual([ + null, ...replacements.map(access => `Bearer ${access}`), + ]); + for (const side of ["inbound", "outbound"] as const) { + expect(rows().map(row => row.continuation[side].status)).toEqual(["first", "equal", "equal"]); + } + for (const access of replacements) expect(readFileSync(cacheDiagnosticPath(), "utf8")).not.toContain(access); + expectPrivacy(); + }); + } +}); diff --git a/tests/usage/cache-diagnostic-hook-compat.test.ts b/tests/usage/cache-diagnostic-hook-compat.test.ts new file mode 100644 index 0000000000..d1e34b27d9 --- /dev/null +++ b/tests/usage/cache-diagnostic-hook-compat.test.ts @@ -0,0 +1,119 @@ +import { afterAll, beforeAll, describe, expect, mock, test } from "bun:test"; +import { mkdtempSync, mkdirSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { repoRoot } from "../helpers/repo-root"; + +const root = repoRoot(); +const originalFetch = globalThis.fetch; +const saved = Object.fromEntries(["OPENCODEX_HOME", "CODEX_HOME", "OPENCODEX_CACHE_DEBUG"] + .map(key => [key, process.env[key]])); +const home = mkdtempSync(join(tmpdir(), "ocx-hook-compat-")); +process.env.OPENCODEX_HOME = home; +process.env.CODEX_HOME = join(home, "codex"); +mkdirSync(process.env.CODEX_HOME); +let networkCalls = 0; +let coreCalls = 0; +const rejectNetwork = () => { + networkCalls++; + throw new Error("Network forbidden in hook compatibility regression"); +}; +globalThis.fetch = Object.assign(async () => rejectNetwork(), { preconnect: rejectNetwork }); + +const fixtureResponse = async () => { + coreCalls++; + return Response.json({ + id: "resp_hook_fixture", object: "response", created_at: 1, status: "completed", + model: "fixture/model", error: null, incomplete_details: null, + output: [{ id: "msg_hook_fixture", type: "message", role: "assistant", status: "completed", + content: [{ type: "output_text", text: "hook fixture answer", annotations: [] }] }], + usage: { input_tokens: 2, output_tokens: 3, total_tokens: 5 }, + }); +}; +// Stop after the real ingress routing/eligibility checks; no provider, auth or inference runs. +mock.module(join(root, "src/server/responses/policy-fallback.ts"), () => ({ + handleResponses: fixtureResponse, + handleResponsesWithPolicyFallback: fixtureResponse, + rankPolicyFallbackCandidates: () => { throw new Error("Unexpected policy fallback"); }, +})); + +let observing: () => boolean; +let chat: typeof import("../../src/server/chat-completions").handleChatCompletions; +let messages: typeof import("../../src/server/claude-messages").handleClaudeMessages; +let messagesDirectRouteHolds: typeof import("../../src/server/messages-antigravity-direct").messagesDirectRouteHolds; +let chatDecline: typeof import("../../src/server/antigravity-direct-eligibility").antigravityDirectDeclineReason; + +beforeAll(async () => { + await import(join(root, "src/usage/cache-diagnostic.ts")); + observing = (await import(join(root, "src/server/request-log.ts"))).cacheDiagnosticsObserving; + chat = (await import(join(root, "src/server/chat-completions.ts"))).handleChatCompletions; + messages = (await import(join(root, "src/server/claude-messages.ts"))).handleClaudeMessages; + messagesDirectRouteHolds = (await import(join(root, "src/server/messages-antigravity-direct.ts"))).messagesDirectRouteHolds; + chatDecline = (await import(join(root, "src/server/antigravity-direct-eligibility.ts"))).antigravityDirectDeclineReason; +}); + +afterAll(() => { + globalThis.fetch = originalFetch; + mock.restore(); + for (const [key, value] of Object.entries(saved)) { + if (value === undefined) delete process.env[key]; + else process.env[key] = value; + } + rmSync(home, { recursive: true, force: true }); +}); + +function debug(value: string | undefined) { + if (value === undefined) delete process.env.OPENCODEX_CACHE_DEBUG; + else process.env.OPENCODEX_CACHE_DEBUG = value; +} + +function config() { + return { + port: 0, defaultProvider: "fixture", + providers: { fixture: { adapter: "openai-responses", baseUrl: "https://fixture.invalid/v1", + apiKey: "synthetic-test-key", models: ["model"] } }, + claudeCode: { enabled: true }, + } as unknown as import("../../src/types").OcxConfig; +} + +describe("registered cache diagnostic hook compatibility", () => { + for (const value of [undefined, "0", "1"]) { + test(`enabled follows environment ${String(value)} after module registration`, () => { + debug(value); + expect(observing()).toBe(value === "1"); + }); + for (const ingress of ["chat", "messages"] as const) { + test(`${ingress} ingress remains usable with diagnostic ${String(value)}`, async () => { + debug(value); + const before = coreCalls; + const body = { model: "fixture/model", stream: false, max_tokens: 16, + messages: [{ role: "user", content: "synthetic hook fixture" }] }; + const req = new Request(`http://localhost/v1/${ingress === "chat" ? "chat/completions" : "messages"}`, { + method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body), + }); + const response = await (ingress === "chat" ? chat : messages)(req, config(), { model: "", provider: "" }); + expect(response.status).toBe(200); + expect(await response.text()).toContain("hook fixture answer"); + expect(coreCalls).toBe(before + 1); + expect(networkCalls).toBe(0); + }); + } + } + + test("debug toggles preserve direct-lane eligibility for Chat and Messages", () => { + const model = "google-antigravity/gemini-3.8-flash"; + const route = { providerName: "google-antigravity", modelId: "gemini-3.8-flash", + provider: { adapter: "google", googleMode: "cloud-code-assist", baseUrl: "https://fixture.invalid" } } as never; + const cfg = { port: 0, providers: {}, protocols: { rollout: { antigravityDirect: true } } } as never; + const body = { model, messages: [{ role: "user", content: "synthetic fixture" }] }; + debug("0"); + expect(chatDecline(route, body, cfg, { cacheDiagnostics: observing() })).toBeUndefined(); + expect(messagesDirectRouteHolds(route, body, model, cfg)).toBe(true); + debug("1"); + expect(chatDecline(route, body, cfg, { cacheDiagnostics: observing() })).toBe("bridge-only-policy"); + expect(messagesDirectRouteHolds(route, body, model, cfg)).toBe(false); + debug("0"); + expect(messagesDirectRouteHolds(route, body, model, cfg)).toBe(true); + expect(networkCalls).toBe(0); + }); +}); diff --git a/tests/usage/cache-diagnostic.test.ts b/tests/usage/cache-diagnostic.test.ts index 4e61d7976d..c7a00e2f28 100644 --- a/tests/usage/cache-diagnostic.test.ts +++ b/tests/usage/cache-diagnostic.test.ts @@ -3,6 +3,7 @@ import { existsSync, mkdtempSync, readFileSync, statSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { captureCodexAffinityDiagnostic } from "../../src/codex/affinity-debug"; +import { codexPoolAffinityKey } from "../../src/codex/auth-context"; import { getDebugLogEntries, resetDebugLogBufferForTests } from "../../src/lib/debug-log-buffer"; import { resetDebugSettingsForTests, setDebugSettings } from "../../src/lib/debug-settings"; import { @@ -10,16 +11,21 @@ import { cacheDiagnosticPath, CACHE_DEBUG_KEEP_LINES, CACHE_DEBUG_MAX_LINES, + observe, observeInbound, observeOutbound, rebindCacheDiagnosticBodyAlias, tagCacheDiagnosticValue, + type CacheDiagnosticFinalFacts, + type TaggedPresence, + type PrefixComparison, } from "../../src/usage/cache-diagnostic"; import { removeTreeWithRetry } from "../helpers/remove-tree"; let testDir = ""; let previousHome: string | undefined; let previousDebug: string | undefined; +const TEST_PRINCIPAL = "synthetic-authenticated-admission-principal"; beforeEach(() => { previousHome = process.env.OPENCODEX_HOME; @@ -64,11 +70,30 @@ function records(): Array> { interface DiagnosticRecord { requestId: string; promptCacheKey: { inbound: { present: boolean; source?: string; tag?: string } }; - session: { inboundHeader: { tag?: string } }; + route: { affinityMove?: string; affinityReason?: string; accountTag?: string }; + accountAffinity: { + derivation: string; + evidence: string; + inbound?: TaggedPresence; + outbound?: TaggedPresence; + }; + session: { + purpose: string; + inboundHeader: TaggedPresence; + inboundHeaders?: Record; + outboundHeaders?: Record; + }; + continuation: { + inbound: PrefixComparison; + outbound: PrefixComparison; + promptCacheKey: { inbound: string; outbound: string }; + account: string; + }; prefix: { inbound: { messages: { tags: string[] } }; outbound: { messages: { tags: string[] } }; firstDivergentBlock?: { section: string; index: number }; + comparison: PrefixComparison; }; cache: { rawUpstream: { present: boolean; value?: number; field?: string }; @@ -80,13 +105,542 @@ function diagnosticRecords(): DiagnosticRecord[] { return records() as unknown as DiagnosticRecord[]; } +// Exercise the real adapter-observation/finalization chain, not fingerprint internals. +function turn( + inbound: object, + outbound: object = inbound, + options: { + session?: string | null; + headers?: HeadersInit; + outboundHeaders?: HeadersInit; + model?: string; + provider?: string; + account?: string; + affinityMove?: string; + affinityReason?: string; + protocol?: CacheDiagnosticFinalFacts["protocol"]; + principal?: string | null; + } = {}, +): void { + const headers = new Headers(options.headers ?? (options.session === null ? {} : { + session_id: options.session ?? "test-session", + })); + const draft = observeInbound(inbound, headers, "caller", options.principal === null ? undefined : options.principal ?? TEST_PRINCIPAL); + observeOutbound(inbound, outbound, options.outboundHeaders ?? headers); + appendFinalCacheDiagnostic({ + requestId: `turn-${recordsIfPresent()}`, + protocol: options.protocol ?? "responses", + provider: options.provider ?? "openai", + model: options.model ?? "gpt-test", + accountLogLabel: options.account, + affinityMove: options.affinityMove, + affinityReason: options.affinityReason, + cacheProvenance: "unknown", + draft, + }); +} + +function recordsIfPresent(): number { + return existsSync(cacheDiagnosticPath()) ? records().length : 0; +} + +const message = (content: string) => ({ role: "user", content }); + describe("cache diagnostic", () => { + test("stable session_id and prompt key correlate diagnostics without claiming pool affinity identity", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const headers = { session_id: "capture-session-sentinel" }; + const body = { input: "same", prompt_cache_key: "capture-prompt-key-sentinel" }; + expect(codexPoolAffinityKey(new Headers(headers))).toBeUndefined(); + turn(body, body, { headers }); + turn(body, body, { headers }); + for (const record of diagnosticRecords()) { + expect(record.session.purpose).toBe("diagnostic-correlation-only"); + expect(record.session.inboundHeaders!.session_id.present).toBe(true); + expect(record.accountAffinity).toEqual({ + derivation: "codexConversationIdentity", + evidence: "derived-key-not-binding", + inbound: { present: false }, + outbound: { present: false }, + }); + } + expect(diagnosticRecords()[1].continuation.inbound.status).toBe("equal"); + const text = readFileSync(cacheDiagnosticPath(), "utf8"); + expect(text).not.toContain("capture-session-sentinel"); + expect(text).not.toContain("capture-prompt-key-sentinel"); + }); + + test("route labels never substitute for the derived pool key or claim a binding", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const body = { input: "same", prompt_cache_key: "label-probe-key" }; + const incomplete = { session_id: "label-probe-session" }; + const complete = { ...incomplete, "session-id": "label-probe-session", "thread-id": "label-probe-thread" }; + for (const headers of [incomplete, complete]) { + const derived = codexPoolAffinityKey(new Headers(headers)); + for (const affinityMove of ["new_bind", "reused"]) { + // Supply finalized labels as facts: this tests diagnostic interpretation, not routing. + turn(body, body, { headers, affinityMove, affinityReason: "label-probe" }); + const record = diagnosticRecords().at(-1)!; + expect(record.route).toMatchObject({ affinityMove, affinityReason: "label-probe" }); + expect(record.accountAffinity.evidence).toBe("derived-key-not-binding"); + const expected = derived === undefined ? { present: false } : { + present: true, tag: tagCacheDiagnosticValue("codex-pool-affinity-key", derived), + }; + expect(record.accountAffinity.inbound).toEqual(expected); + expect(record.accountAffinity.outbound).toEqual(expected); + expect(record.session.purpose).toBe("diagnostic-correlation-only"); + } + } + const rows = diagnosticRecords(); + expect(rows[1].continuation.inbound.status).toBe("equal"); + expect(rows[0].accountAffinity.inbound).toEqual({ present: false }); + expect(rows[2].accountAffinity.inbound?.present).toBe(true); + expect(rows[2].accountAffinity.inbound).toEqual(rows[3].accountAffinity.inbound); + }); + + test("reports the real pool-key derivation separately for each observed header set", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const base = { session_id: "raw-capture-session", "session-id": "raw-affinity-session" }; + const full = { ...base, "thread-id": "raw-affinity-thread" }; + const rawKey = codexPoolAffinityKey(new Headers(full)); + expect(rawKey).toBeDefined(); + for (const headers of [base, { "thread-id": "raw-affinity-thread" }, { "x-session-id": "raw-capture-session" }]) { + expect(codexPoolAffinityKey(new Headers(headers))).toBeUndefined(); + turn({ input: "a" }, undefined, { headers, outboundHeaders: full }); + const record = diagnosticRecords().at(-1)!; + expect(record.accountAffinity.inbound).toEqual({ present: false }); + expect(record.accountAffinity.outbound).toEqual({ + present: true, tag: tagCacheDiagnosticValue("codex-pool-affinity-key", rawKey!), + }); + expect(record.accountAffinity.evidence).toBe("derived-key-not-binding"); + } + turn({ input: "a" }, undefined, { headers: full, outboundHeaders: full }); + const record = diagnosticRecords().at(-1)!; + expect(record.accountAffinity.inbound).toEqual(record.accountAffinity.outbound); + expect(record.accountAffinity.inbound?.tag).not.toBe(record.session.inboundHeader.tag); + const text = readFileSync(cacheDiagnosticPath(), "utf8"); + for (const value of [rawKey!, ...Object.values(full)]) expect(text).not.toContain(value); + }); + + test("does not turn a missing outbound observation into an absent affinity key", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const draft = observeInbound({ input: "a" }, new Headers({ "session-id": "s", "thread-id": "t" })); + append("inbound-only-affinity", draft); + const record = diagnosticRecords()[0]; + expect(record.accountAffinity.inbound?.present).toBe(true); + expect(record.accountAffinity.outbound).toBeUndefined(); + }); + + test("compares consecutive inbound and outbound prefixes independently", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + turn({ input: [message("a")] }); + turn({ input: [message("a"), message("b")] }, { input: [message("rewritten"), message("b")] }); + turn({ input: [message("a"), message("b")] }, { input: [message("rewritten"), message("b")] }); + const [first, second, same] = diagnosticRecords(); + expect(first.continuation.inbound).toEqual({ status: "first" }); + expect(first.continuation.outbound).toEqual({ status: "first" }); + expect(second.continuation.inbound).toEqual({ status: "append-only", section: "messages", index: 1 }); + expect(second.continuation.outbound).toEqual({ status: "different", section: "messages", index: 0 }); + expect(same.continuation.inbound).toEqual({ status: "equal" }); + expect(same.continuation.outbound).toEqual({ status: "equal" }); + }); + + test("seven wire-shaped turns remain append-only when skill and permission updates append input roles", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const content = (text: string) => [{ type: "input_text", text }]; + const additions = [ + [{ role: "developer", content: content("initial-policy-sentinel") }, message("task-sentinel")], + [{ role: "assistant", content: [{ type: "output_text", text: "reply-sentinel" }] }], + [{ role: "system", content: content("skill-update-sentinel") }, message("continue-sentinel")], + [{ type: "function_call", call_id: "call-sentinel", name: "tool-sentinel", arguments: "{}" }], + [{ type: "function_call_output", call_id: "call-sentinel", output: "result-sentinel" }], + [{ role: "developer", content: content("permission-update-sentinel") }, message("next-sentinel")], + [{ role: "system", content: content("second-update-sentinel") }], + ]; + let input: object[] = []; + for (const rows of additions) { + const previousLength = input.length; + input = [...input, ...rows]; + const body = { + instructions: "standing-instruction-sentinel", + system: [{ type: "text", text: "top-system-sentinel" }], + tools: [{ type: "function", name: "tool-sentinel", parameters: { type: "object" } }], + input, + }; + const before = JSON.stringify(body); + // Distinct parsed objects model the actual inbound and final outbound observations. + turn(body, JSON.parse(before)); + expect(JSON.stringify(body)).toBe(before); + const record = diagnosticRecords().at(-1)!; + for (const side of ["inbound", "outbound"] as const) { + expect(record.continuation[side]).toEqual(previousLength === 0 ? { status: "first" } + : { status: "append-only", section: "messages", index: previousLength }); + } + expect(record.prefix.comparison).toEqual({ status: "equal" }); + expect(record.prefix.inbound.messages.tags).toHaveLength(input.length); + } + expect(readFileSync(cacheDiagnosticPath(), "utf8")).not.toContain("sentinel"); + }); + + test("existing system/developer input rows still detect role, position and content rewrites", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + for (const field of ["input", "messages"]) { + for (const role of ["system", "developer"]) { + const policy = { role, content: [{ type: "input_text", text: "original-policy" }] }; + const before = [message("first"), policy, message("last")]; + const appendSession = `${field}-${role}-append`; + turn({ [field]: before }, undefined, { session: appendSession }); + turn({ [field]: [...before, policy] }, undefined, { session: appendSession }); + for (const side of ["inbound", "outbound"] as const) { + expect(diagnosticRecords().at(-1)!.continuation[side]).toEqual({ status: "append-only", section: "messages", index: 3 }); + } + const rewrites = [ + { rows: [before[0], { ...policy, role: role === "system" ? "developer" : "system" }, before[2]], index: 1 }, + { rows: [policy, before[0], before[2]], index: 0 }, + { rows: [before[0], { ...policy, content: [{ type: "input_text", text: "changed-policy" }] }, before[2]], index: 1 }, + ]; + for (const [index, rewrite] of rewrites.entries()) { + const session = `${field}-${role}-${index}`; + turn({ [field]: before }, undefined, { session }); + turn({ [field]: rewrite.rows }, undefined, { session }); + const record = diagnosticRecords().at(-1)!; + for (const side of ["inbound", "outbound"] as const) { + expect(record.continuation[side]).toEqual({ status: "different", section: "messages", index: rewrite.index }); + } + } + } + } + }); + + test("appended policy rows obey the ordered input truncation boundary", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const input = Array.from({ length: 128 }, (_, index) => ({ + role: index % 2 ? "developer" : "system", content: `policy-${index}`, + })); + turn({ input }); + const extended = [...input, { role: "developer", content: "permission-update" }]; + turn({ input: extended }); + for (const side of ["inbound", "outbound"] as const) { + expect(diagnosticRecords().at(-1)!.continuation[side]).toEqual({ status: "append-only", section: "messages", index: 128 }); + } + turn({ input: [...extended, { role: "system", content: "skill-update" }] }); + for (const side of ["inbound", "outbound"] as const) { + expect(diagnosticRecords().at(-1)!.continuation[side]).toEqual({ status: "truncated", section: "messages", index: 128 }); + } + expect(diagnosticRecords().at(-1)!.prefix.firstDivergentBlock).toBeUndefined(); + }); + + test("reports tool, system, instruction and message rewrites and removals", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const cases = [ + [{ tools: [{ type: "function", name: "a" }] }, { tools: [{ type: "function", name: "b" }] }, "tools"], + [{ system: "a" }, { system: "b" }, "instructions"], + [{ instructions: "a" }, { instructions: "b" }, "instructions"], + [{ input: [{ role: "system", content: "a" }] }, { input: [{ role: "developer", content: "a" }] }, "messages"], + [{ instructions: "a" }, { system: "a" }, "instructions"], + [{ input: [{ role: "system", content: "a" }, message("b")] }, { input: [message("b"), { role: "system", content: "a" }] }, "messages"], + [{ input: "a" }, { input: "b" }, "messages"], + [{ input: [message("a")] }, { input: [] }, "messages"], + ] as const; + for (const [before, after, section] of cases) { + const session = `case-${section}-${recordsIfPresent()}`; + turn(before, before, { session }); + turn(after, after, { session }); + const record = diagnosticRecords().at(-1)!; + for (const side of ["inbound", "outbound"] as const) { + expect(record.continuation[side]).toEqual({ status: "different", section, index: 0 }); + } + } + }); + + test("isolates interleaved sessions, models and protocols; tracks key and account changes", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const first = { input: [message("a")], prompt_cache_key: "key-a" }; + turn(first, first, { session: "a", account: "account-a" }); + turn({ input: [message("unrelated")] }, undefined, { session: "b" }); + turn(first, first, { session: "a", model: "another-model" }); + turn(first, first, { session: "a", protocol: "chat" }); + const next = { input: [message("a"), message("b")], prompt_cache_key: "key-b" }; + turn(next, { ...next, prompt_cache_key: "key-a" }, { session: "a", account: "account-b" }); + const rows = diagnosticRecords(); + for (const row of rows.slice(1, 4)) expect(row.continuation.inbound.status).toBe("first"); + expect(rows[4].continuation.inbound.status).toBe("append-only"); + expect(rows[4].continuation.promptCacheKey).toEqual({ inbound: "changed", outbound: "equal" }); + expect(rows[4].continuation.account).toBe("changed"); + const text = readFileSync(cacheDiagnosticPath(), "utf8"); + for (const secret of ["key-a", "key-b", "account-a", "account-b", "unrelated"]) expect(text).not.toContain(secret); + }); + + test("isolates identical sessions by trusted admission, never by mutable caller headers", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const headers = new Headers({ session_id: "shared-session", authorization: "Bearer test-upstream" }); + const a = "synthetic-admission-a"; + const b = "synthetic-admission-b"; + const body = { input: [message("same")], prompt_cache_key: "shared-key" }; + for (const principal of [a, b, null, a, b, null]) { + turn(body, body, { headers, principal, account: "same-upstream-account" }); + } + const rows = diagnosticRecords(); + for (const row of [rows[0], rows[1]]) { + expect(row.continuation.inbound.status).toBe("first"); + expect(row.continuation.outbound.status).toBe("first"); + expect(row.continuation.account).toBe("first"); + } + for (const row of [rows[3], rows[4]]) { + expect(row.continuation.inbound.status).toBe("equal"); + expect(row.continuation.outbound.status).toBe("equal"); + expect(row.continuation.account).toBe("equal"); + } + for (const row of [rows[2], rows[5]]) { + expect(row.continuation.inbound).toEqual({ status: "uncomparable", reason: "missing-observation" }); + expect(row.continuation.outbound).toEqual({ status: "uncomparable", reason: "missing-observation" }); + expect(row.continuation.account).toBe("uncomparable"); + expect(row.prefix.comparison.status).toBe("equal"); + } + const text = readFileSync(cacheDiagnosticPath(), "utf8"); + expect(text).not.toContain(headers.get("authorization")!); + for (const scope of [a, b]) { + expect(text).not.toContain(scope); + expect(text).not.toContain(tagCacheDiagnosticValue("caller-scope", scope)); + } + }); + + test("provider changes preserve continuity and account evidence comes only from explicit facts", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + turn({ input: "same" }, undefined, { provider: "openai-p111111", account: "p111111" }); + turn({ input: "same" }, undefined, { provider: "openai", account: "main" }); + const rotated = diagnosticRecords().at(-1)!; + expect(rotated.continuation.inbound.status).toBe("equal"); + expect(rotated.continuation.outbound.status).toBe("equal"); + expect(rotated.continuation.account).toBe("changed"); + turn({ input: "same" }, undefined, { provider: "anthropic-p111111" }); + turn({ input: "same" }, undefined, { provider: "anthropic-p222222" }); + const noAccount = diagnosticRecords().at(-1)!; + expect(noAccount.continuation.outbound.status).toBe("equal"); + expect(noAccount.continuation.account).toBe("equal"); + expect(noAccount.route.accountTag).toBeUndefined(); + expect(diagnosticRecords().at(-2)!.route.accountTag).toBeUndefined(); + turn({ input: "same" }, undefined, { provider: "unrelated-provider" }); + expect(diagnosticRecords().at(-1)!.continuation.inbound.status).toBe("equal"); + }); + + test("session header values share a tag domain across names and directions without merging identities", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const names = ["session_id", "session-id", "thread-id", "x-session-id"]; + const headers = Object.fromEntries(names.map(name => [name, "shared-header-value"])); + turn({ input: "same" }, undefined, { headers, outboundHeaders: headers }); + const record = diagnosticRecords()[0]; + const expected = record.session.inboundHeaders!.session_id; + for (const name of names) { + expect(record.session.inboundHeaders![name]).toEqual(expected); + expect(record.session.outboundHeaders![name]).toEqual(expected); + } + turn({ input: "same" }, undefined, { headers: { session_id: "shared-header-value" } }); + turn({ input: "same" }, undefined, { headers: { "x-session-id": "shared-header-value" } }); + const rows = diagnosticRecords(); + expect(rows[1].continuation.inbound.status).toBe("first"); + expect(rows[2].continuation.inbound.status).toBe("first"); + expect(rows[1].session.inboundHeaders!.session_id.tag).toBe(rows[2].session.inboundHeaders!["x-session-id"].tag); + }); + + test("records each session header independently and ignores unrelated rotating headers", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const names = ["session_id", "session-id", "thread-id", "x-session-id"]; + for (const name of names) { + const headers = { [name]: `private-${name}`, "x-client-request-id": "request-a" }; + turn({ input: "a" }, undefined, { headers, outboundHeaders: {} }); + turn({ input: "a" }, undefined, { headers: { ...headers, "x-client-request-id": "request-b" } }); + const record = diagnosticRecords().at(-1)!; + expect(record.continuation.inbound.status).toBe("equal"); + expect(record.session.inboundHeaders![name]).toMatchObject({ present: true, tag: expect.stringMatching(/^[0-9a-f]{12}$/) }); + for (const other of names.filter(value => value !== name)) { + expect(record.session.inboundHeaders![other]).toEqual({ present: false }); + } + const previous = diagnosticRecords().at(-2)!; + expect(previous.session.outboundHeaders![name]).toEqual({ present: false }); + } + expect(readFileSync(cacheDiagnosticPath(), "utf8")).not.toContain("private-"); + }); + + test("never associates headerless requests using prompt keys or incidental headers", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const body = { prompt_cache_key: "shared-key", input: "same" }; + for (let index = 0; index < 2; index++) turn(body, body, { + headers: { originator: "client", "x-client-request-id": "incidental", session_id: "" }, + outboundHeaders: { session_id: "synthesized" }, + }); + for (const record of diagnosticRecords()) { + expect(record.continuation.inbound).toEqual({ status: "uncomparable", reason: "missing-session" }); + expect(record.continuation.outbound).toEqual({ status: "uncomparable", reason: "missing-session" }); + } + }); + + test("does not claim equality or an exact divergence beyond the 128-block bound", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const input = Array.from({ length: 129 }, (_, index) => message(`block-${index}`)); + turn({ input }); + turn({ input }); + turn({ input: [...input.slice(0, 128), message("hidden-rewrite")] }); + turn({ input: [...input, message("extra")] }); + for (const record of diagnosticRecords().slice(1)) { + expect(record.continuation.inbound).toEqual({ status: "truncated", section: "messages", index: 128 }); + expect(record.continuation.outbound.status).toBe("truncated"); + expect(record.prefix.comparison.status).toBe("truncated"); + expect(record.prefix.firstDivergentBlock).toBeUndefined(); + expect(record.prefix.inbound.messages.tags).toHaveLength(128); + } + turn({ input: [message("visible-rewrite"), ...input.slice(1)] }); + expect(diagnosticRecords().at(-1)!.continuation.inbound).toEqual({ status: "different", section: "messages", index: 0 }); + }); + + test("proves a 128-block prefix append, but treats growth before messages as a rewrite", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const input = Array.from({ length: 128 }, (_, index) => message(`message-${index}`)); + turn({ input }); + turn({ input: [...input, message("tail")] }); + expect(diagnosticRecords().at(-1)!.continuation.inbound).toEqual({ status: "append-only", section: "messages", index: 128 }); + for (const section of ["instructions", "tools"] as const) { + turn({ [section]: ["one"], input }, undefined, { session: section }); + turn({ [section]: ["one", "two"], input }, undefined, { session: section }); + expect(diagnosticRecords().at(-1)!.continuation.inbound).toEqual({ status: "different", section, index: 1 }); + } + const instructions = Array.from({ length: 129 }, (_, index) => `instruction-${index}`); + turn({ instructions, input: "a" }, undefined, { session: "truncated-instructions" }); + turn({ instructions, input: "b" }, undefined, { session: "truncated-instructions" }); + expect(diagnosticRecords().at(-1)!.continuation.inbound).toEqual({ status: "truncated", section: "instructions", index: 128 }); + }); + + test("uses the complete inbound header identity, not one matching header or the outbound identity", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + for (const thread of ["a", "b", "a"]) turn({ input: "a" }, undefined, { + headers: { session_id: "shared-session", "thread-id": thread }, + outboundHeaders: { session_id: "same-upstream-session" }, + }); + const rows = diagnosticRecords(); + expect(rows[0].continuation.inbound.status).toBe("first"); + expect(rows[1].continuation.inbound.status).toBe("first"); + expect(rows[2].continuation.inbound.status).toBe("equal"); + expect(rows[2].continuation.outbound.status).toBe("equal"); + }); + + test("detects prompt-key and account presence changes without coupling content comparison to them", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + turn({ input: "same", prompt_cache_key: "private-key" }, undefined, { account: "private-account" }); + turn({ input: "same" }); + turn({ input: "same" }); + const [, removed, unchanged] = diagnosticRecords(); + expect(removed.continuation.inbound.status).toBe("equal"); + expect(removed.continuation.promptCacheKey).toEqual({ inbound: "changed", outbound: "changed" }); + expect(removed.continuation.account).toBe("changed"); + expect(unchanged.continuation.account).toBe("equal"); + expect(unchanged.continuation.promptCacheKey).toEqual({ inbound: "equal", outbound: "equal" }); + }); + + test("bounds previous-session records and treats an evicted session as first", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + turn({ input: "first" }, undefined, { session: "oldest" }); + for (let index = 0; index < 512; index++) turn({ input: "other" }, undefined, { session: `other-${index}` }); + turn({ input: "first" }, undefined, { session: "oldest" }); + expect(diagnosticRecords().at(-1)!.continuation.inbound.status).toBe("first"); + expect(diagnosticRecords().length).toBeLessThanOrEqual(CACHE_DEBUG_MAX_LINES); + }); + + test("missing and unsupported observations remain uncomparable, not empty equal prefixes", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const body = { input: "a" }; + const draft = observeInbound(body, new Headers({ session_id: "missing-outbound" })); + append("missing", draft); + expect(diagnosticRecords()[0].prefix.comparison).toEqual({ status: "uncomparable", reason: "missing-observation" }); + turn({ input: 123 }); + turn({ input: 123 }); + expect(diagnosticRecords().at(-1)!.continuation.inbound).toEqual({ status: "uncomparable", reason: "unsupported-shape" }); + }); + test("is disabled by default", () => { delete process.env.OPENCODEX_CACHE_DEBUG; append("disabled"); expect(existsSync(cacheDiagnosticPath())).toBe(false); }); + test("disabled entry points never inspect bodies, headers, observations or final facts", () => { + for (const value of [undefined, "0", "true"]) { + if (value === undefined) delete process.env.OPENCODEX_CACHE_DEBUG; + else process.env.OPENCODEX_CACHE_DEBUG = value; + let reads = 0; + const unreadable = new Proxy({}, { + get() { reads++; throw new Error("disabled diagnostics inspected input"); }, + ownKeys() { reads++; throw new Error("disabled diagnostics enumerated input"); }, + }); + expect(observeInbound(unreadable, unreadable as Headers)).toEqual({}); + expect(observe("disabled", unreadable)).toEqual({}); + expect(() => rebindCacheDiagnosticBodyAlias(unreadable, unreadable)).not.toThrow(); + expect(() => observeOutbound(unreadable, unreadable, unreadable as Headers)).not.toThrow(); + expect(() => appendFinalCacheDiagnostic(unreadable as CacheDiagnosticFinalFacts)).not.toThrow(); + expect(reads).toBe(0); + expect(existsSync(cacheDiagnosticPath())).toBe(false); + } + }); + + test("missing observations stay unknown on later turns and unsupported shapes do not compare equal", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const headers = new Headers({ session_id: "partial-observation" }); + const draft = observeInbound({ input: "same" }, headers, "caller", TEST_PRINCIPAL); + append("inbound-only", draft); + turn({ input: "same" }, undefined, { headers }); + const next = diagnosticRecords().at(-1)!; + expect(next.continuation.inbound).toEqual({ status: "equal" }); + expect(next.continuation.outbound).toEqual({ status: "uncomparable", reason: "missing-observation" }); + expect(next.continuation.promptCacheKey.outbound).toBe("uncomparable"); + for (const body of [{ input: null }, { messages: "unknown" }, { tools: {} }]) { + turn(body, body, { session: JSON.stringify(body) }); + turn(body, body, { session: JSON.stringify(body) }); + const record = diagnosticRecords().at(-1)!; + expect(record.prefix.comparison).toEqual({ status: "uncomparable", reason: "unsupported-shape" }); + expect(record.continuation.inbound).toEqual({ status: "uncomparable", reason: "unsupported-shape" }); + expect(record.prefix.firstDivergentBlock).toBeUndefined(); + } + }); + + test("missing session-header and prompt-key observations stay distinct from observed empty values", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + append("no-observation", {}); + const missing = diagnosticRecords().at(-1)!; + expect(missing.session.inboundHeaders).toBeUndefined(); + expect(missing.session.outboundHeaders).toBeUndefined(); + expect(missing.continuation.promptCacheKey).toEqual({ inbound: "uncomparable", outbound: "uncomparable" }); + const headers = new Headers({ session_id: "inbound-only" }); + const body = { input: "same" }; + append("inbound-only", observeInbound(body, headers, "caller", TEST_PRINCIPAL)); + const partial = diagnosticRecords().at(-1)!; + expect(partial.session.inboundHeaders?.session_id.present).toBe(true); + expect(partial.session.outboundHeaders).toBeUndefined(); + expect(partial.continuation.promptCacheKey).toEqual({ inbound: "first", outbound: "uncomparable" }); + turn(body, body, { headers, outboundHeaders: {} }); + const observed = diagnosticRecords().at(-1)!; + expect(observed.session.outboundHeaders).toEqual(Object.fromEntries( + ["session_id", "session-id", "thread-id", "x-session-id"].map(name => [name, { present: false }]), + )); + expect(observed.continuation.promptCacheKey).toEqual({ inbound: "equal", outbound: "uncomparable" }); + turn(body, body, { headers, outboundHeaders: {} }); + expect(diagnosticRecords().at(-1)!.continuation.promptCacheKey.outbound).toBe("equal"); + }); + + test("continuation compares finalization order, not ingress or request-id order", () => { + process.env.OPENCODEX_CACHE_DEBUG = "1"; + const headers = new Headers({ session_id: "overlapping-turns" }); + const early = { input: [message("first")] }; + const later = { input: [message("first"), message("second")] }; + const earlyDraft = observeInbound(early, headers, "caller", TEST_PRINCIPAL); + const laterDraft = observeInbound(later, headers, "caller", TEST_PRINCIPAL); + observeOutbound(early, early, headers); + observeOutbound(later, later, headers); + append("later-request", laterDraft); + append("early-request", earlyDraft); + const [first, second] = diagnosticRecords(); + expect(first.continuation.inbound).toEqual({ status: "first" }); + expect(second.continuation.inbound).toEqual({ status: "different", section: "messages", index: 1 }); + }); + test("writes only presence, process tags, counts, and bounded metadata", () => { process.env.OPENCODEX_CACHE_DEBUG = "1"; const cacheKey = "raw-cache-key-sentinel"; @@ -163,7 +717,7 @@ describe("cache diagnostic", () => { expect(JSON.stringify(body)).toBe(before); const [record] = diagnosticRecords(); - expect(record.prefix.inbound.messages.tags).toHaveLength(1); + expect(record.prefix.inbound.messages.tags).toHaveLength(2); }); test("keeps an observed upstream zero distinct from an absent counter", () => { @@ -207,3 +761,17 @@ describe("cache diagnostic", () => { expect(tagCacheDiagnosticValue("session_id", value)).not.toBe(payload.inbound[0].tag); }); }); + + +test("registered cache diagnostic hook retains its live enabled contract", () => { + const hook = (globalThis as Record)[ + Symbol.for("opencodex.cache-diagnostic.v1") + ]; + expect(typeof hook?.enabled).toBe("function"); + process.env.OPENCODEX_CACHE_DEBUG = "0"; + expect(hook?.enabled()).toBe(false); + process.env.OPENCODEX_CACHE_DEBUG = "1"; + expect(hook?.enabled()).toBe(true); + delete process.env.OPENCODEX_CACHE_DEBUG; + expect(hook?.enabled()).toBe(false); +});