From 7235920ffd994acf5396207661c4ce8a50515002 Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Thu, 20 Aug 2026 19:21:40 +0200 Subject: [PATCH] feat(opencode): capture cache-diagnostics and usage cache accounting per request The plugin has sent cache-diagnosis-2026-04-07 in its beta list on every request without enabling the feature: the beta requires a request-body opt-in (diagnostics.previous_message_id), and nothing sent it or read the response. Turn the dormant channel on, measure-only. Request side: eligible OAuth requests gain the diagnostics opt-in inside the fail-closed rewriteRequestBody pipeline. The id sent is only ever one captured from a prior Anthropic response (bounded per-session tracker) - opencode mints its own msg_01-shaped ids for every provider, and a foreign id fails silently as previous_message_not_found. Response side: the existing SSE wrapper exposes message_start.message through a typed callback; per valid eligible response one MC-CACHE-DIAG single-line JSON record (schema v:2) is emitted via the logger with verbatim usage, TTL-bucket accounting, diag_state (absent|server_null|pending|populated, always a string), populated cache_miss_reason.type, and attribution fields: source (open set) + synthetic (closed machinery/traffic split with published mapping), account_id (opaque persisted identifier; consumer timelines key on (account_id, prefix) because sticky routing migrates sessions across account-scoped caches), betas_hash (xxh64 of the sorted sent beta list, resolved by a once-per-hash MC-CACHE-DIAG-BETAS side-channel line), and requested_model present only on request/served divergence. Dumps: responses gain artifacts (status/id/model/usage/diagnostics, never content); CacheKeep prewarms are dumped tagged -prewarm-cachekeep- and emit records through the same chain, since any write resets the server TTL clock. Canary: short-gap previous_message_not_found with a genuinely-sent id logs a warn - the id capture broke, not a fingerprint expiry. Every per-chunk stateful consumer in the response wrapper is bounded at 8 MiB with drain-before-cap semantics; complete frames are processed before overflow passthrough engages. Verified live: null(write=24831) -> null(read=24831, hit) -> system_changed(cache_missed_input_tokens=23963) on a forced system change against a warm prefix. Comparison engages only on cacheable requests; consumers classify hit-first (documented in README, which is the record contract). Closes #157 --- CHANGELOG.md | 2 + packages/core/src/cachekeep.ts | 84 ++- packages/core/src/cch.ts | 6 + packages/core/src/claude-code.ts | 9 +- packages/core/src/dump.ts | 73 +- packages/core/src/relay.ts | 7 +- packages/core/src/tests/dump.test.ts | 140 +++- packages/opencode/README.md | 68 ++ packages/opencode/src/cache-diagnostics.ts | 322 +++++++++ packages/opencode/src/index.ts | 415 ++++++++++-- packages/opencode/src/server-fallback.ts | 25 +- .../src/tests/cache-diagnostics.test.ts | 386 +++++++++++ packages/opencode/src/tests/cachekeep.test.ts | 137 ++++ packages/opencode/src/tests/index.test.ts | 641 ++++++++++++++++++ .../opencode/src/tests/plugin-exports.test.ts | 8 +- packages/opencode/src/tests/relay.test.ts | 46 ++ packages/opencode/src/tests/transform.test.ts | 238 +++++++ packages/opencode/src/transform.ts | 156 ++++- 18 files changed, 2672 insertions(+), 91 deletions(-) create mode 100644 packages/opencode/src/cache-diagnostics.ts create mode 100644 packages/opencode/src/tests/cache-diagnostics.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 63c54772..172d35a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ This repo is a CortexKit-maintained Anthropic auth monorepo for OpenCode and Pi. ## Unreleased +- Capture Anthropic cache diagnostics in versioned `MC-CACHE-DIAG ` records, preserve provider response IDs across requests and cachekeep prewarms, and write response/request dump artifacts without response content. Document the beta states and known fingerprint, organization, workspace, and beta-set limitations. + ## 1.19.1 ### Patch Changes diff --git a/packages/core/src/cachekeep.ts b/packages/core/src/cachekeep.ts index 3d9621ad..8707a368 100644 --- a/packages/core/src/cachekeep.ts +++ b/packages/core/src/cachekeep.ts @@ -2,6 +2,7 @@ import type { AccountStorage } from './accounts.ts' import type { CacheKeepTrackedSession } from './cachekeep-registry.ts' import { signRequestBody } from './cch.ts' import { orderClaudeCodeBody } from './claude-code.ts' +import { dumpDirectRequest, dumpResponseArtifact } from './dump.ts' import { logger } from './logger.ts' export const CLAUDE_CACHE_KEEP_COMMAND_NAME = 'claude-cachekeep' @@ -328,6 +329,7 @@ export type CacheKeepTarget = { cacheExpiresAt: number dayKey: string oauthAccountId?: string + isSubagent: boolean } export type CacheKeepPrewarmResult = @@ -360,6 +362,17 @@ export class CacheKeepManager { onTrackedSessionsChanged?: ( sessions: readonly CacheKeepTrackedSession[], ) => Promise | void + prepareBody?: ( + bodyText: string, + target: CacheKeepTarget, + ) => string | Promise + onResponse?: (input: { + target: CacheKeepTarget + bodyText: string + status: number + data: unknown + receivedAt: number + }) => void | Promise }, ) {} @@ -482,6 +495,7 @@ export class CacheKeepManager { storage: AccountStorage | null cacheMode: string oauthAccountId?: string + isSubagent?: boolean }) { if (!input.sessionId) return { tracked: false, reason: 'missing session id' } @@ -517,6 +531,7 @@ export class CacheKeepManager { cacheExpiresAt: now + CACHE_KEEP_TTL_MS, dayKey: today, oauthAccountId: input.oauthAccountId, + isSubagent: input.isSubagent ?? false, }) this.pruneTargets(now, today) this.publishTrackedSessions() @@ -530,6 +545,7 @@ export class CacheKeepManager { headers: Headers bodyText: string oauthAccountId?: string + isSubagent?: boolean }): Promise { const headers: Record = {} input.headers.forEach((value, key) => { @@ -543,6 +559,7 @@ export class CacheKeepManager { cacheExpiresAt: this.options.now?.() ?? Date.now(), dayKey: '', oauthAccountId: input.oauthAccountId, + isSubagent: input.isSubagent ?? false, } return this.sendPrewarm(target) } @@ -584,11 +601,23 @@ export class CacheKeepManager { private async sendPrewarm( target: CacheKeepTarget, ): Promise { - const prewarm = await buildCacheKeepPrewarmBody(target.bodyText) + let bodyText = target.bodyText + if (this.options.prepareBody) { + try { + bodyText = await this.options.prepareBody(bodyText, target) + } catch (error) { + logger.warn('cachekeep', 'prepare body failed', { + session: target.id, + error: error instanceof Error ? error.message : String(error), + }) + } + } + const preparedTarget = { ...target, bodyText } + const prewarm = await buildCacheKeepPrewarmBody(bodyText) if (!prewarm.ok) return prewarm const fetchImpl = this.options.fetchImpl ?? fetch - const prewarmTarget = { ...target, bodyText: prewarm.bodyText } + const prewarmTarget = { ...preparedTarget, bodyText: prewarm.bodyText } const headers = this.options.prepareHeaders ? await this.options.prepareHeaders( new Headers(target.headers), @@ -605,21 +634,54 @@ export class CacheKeepManager { this.options.prewarmTimeoutMs ?? CACHE_KEEP_PREWARM_TIMEOUT_MS, ), }) + const receivedAt = this.options.now?.() ?? Date.now() + const raw = await response.text().catch(() => '') + let data: unknown = null + try { + data = raw ? JSON.parse(raw) : null + } catch {} + try { + await this.options.onResponse?.({ + target, + bodyText: prewarm.bodyText, + status: response.status, + data, + receivedAt, + }) + } catch {} + try { + const dumpHandle = await dumpDirectRequest({ + affinity: target.id, + route: 'cachekeep', + status: response.status, + bodyText: prewarm.bodyText, + url: target.url, + method: 'POST', + headers, + tag: 'cachekeep', + }) + await dumpResponseArtifact(dumpHandle, { + status: response.status, + message: data, + }) + } catch (error) { + logger.debug('cachekeep', 'dump failed', { + session: target.id, + error: error instanceof Error ? error.message : String(error), + }) + } if (!response.ok) { return { ok: false, - reason: await response - .text() - .catch(() => '') - .then((body) => body || `HTTP ${response.status}`), + reason: raw || `HTTP ${response.status}`, status: response.status, } } - const data = (await response.json().catch(() => null)) as Record< - string, - unknown - > | null - const usage = data?.usage as + const objectData = + data && typeof data === 'object' && !Array.isArray(data) + ? (data as Record) + : null + const usage = objectData?.usage as | { input_tokens?: number cache_creation_input_tokens?: number diff --git a/packages/core/src/cch.ts b/packages/core/src/cch.ts index 4cc8243f..2904d012 100644 --- a/packages/core/src/cch.ts +++ b/packages/core/src/cch.ts @@ -58,6 +58,12 @@ export async function computeCCH(bodyBytes: Uint8Array): Promise { return (hash & 0xfffffn).toString(16).padStart(5, '0') } +export async function computeXxhash64Hex(value: string): Promise { + await ensureXxhash() + const hash = xxhash64Raw?.(new TextEncoder().encode(value), 0n) ?? 0n + return hash.toString(16).padStart(16, '0').slice(0, 16) +} + export function resetBillingHeaderCCH(bodyString: string): string { return bodyString.replace(BILLING_HEADER_CCH_PATTERN, `$1${CCH_PLACEHOLDER}`) } diff --git a/packages/core/src/claude-code.ts b/packages/core/src/claude-code.ts index a9c34d8e..54321a6b 100644 --- a/packages/core/src/claude-code.ts +++ b/packages/core/src/claude-code.ts @@ -18,8 +18,13 @@ export type ClaudeCodeIdentity = { const IDENTITY_CACHE_LIMIT = 1_000 const identityCache = new Map() -function setBounded(map: Map, key: K, value: V) { - if (!map.has(key) && map.size >= IDENTITY_CACHE_LIMIT) { +export function setBounded( + map: Map, + key: K, + value: V, + limit = IDENTITY_CACHE_LIMIT, +) { + if (!map.has(key) && map.size >= limit) { const oldest = map.keys().next().value if (oldest !== undefined) map.delete(oldest) } diff --git a/packages/core/src/dump.ts b/packages/core/src/dump.ts index 00ea38f6..16d279a5 100644 --- a/packages/core/src/dump.ts +++ b/packages/core/src/dump.ts @@ -29,7 +29,8 @@ const DEFAULT_DUMP_MAX_BYTES = 512 * 1024 * 1024 const DUMP_SWEEP_INTERVAL_MS = 5 * 60 * 1000 const DUMP_SWEEP_NEWNESS_FLOOR_MS = 60 * 1000 const DUMP_PARTIAL_STALE_MS = 10 * 60 * 1000 -const DUMP_ARTIFACT_SUFFIX_PATTERN = /\.(body|meta|relay|request)\.json$/ +const DUMP_ARTIFACT_SUFFIX_PATTERN = + /\.(body|meta|relay|request|response)\.json$/ // Earlier builds emitted five-digit counters; current counters grow without truncation. const DUMP_ARTIFACT_ID_PATTERN = /^\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}-\d{3}Z-\d{5,}-(.+)$/ @@ -54,6 +55,13 @@ export type DumpCommandAction = | { type: 'disable' } | { type: 'usage' } +export type DumpTag = 'cachekeep' + +export type DumpHandle = { + responsePath: string + tag?: DumpTag +} + export function isDumpEnabled() { return dumpEnabled } @@ -311,6 +319,10 @@ function dumpRequestSegment(input: { return `direct${route}` } +function dumpTagSegment(tag: DumpTag | undefined) { + return tag ? `-prewarm-${tag}` : '' +} + function directDumpPreviousKey(input: { affinity?: string | null route?: string @@ -467,26 +479,29 @@ async function dumpRequest(input: { previousBodyText?: string payload?: unknown relayBytes?: number + tag?: DumpTag request?: { url?: string method?: string headers?: DumpHeaders } }) { - if (!dumpEnabled) return + if (!dumpEnabled) return null nextDumpId += 1 const affinity = input.affinity?.trim() || 'session-unknown' - const id = `${new Date().toISOString().replace(/[:.]/g, '-')}-${String(nextDumpId).padStart(6, '0')}-${dumpFileSessionSegment(affinity)}-${dumpRequestSegment(input)}` + const id = `${new Date().toISOString().replace(/[:.]/g, '-')}-${String(nextDumpId).padStart(6, '0')}-${dumpFileSessionSegment(affinity)}${dumpTagSegment(input.tag)}-${dumpRequestSegment(input)}` const dumpDir = getDumpDirectory() const prefix = join(dumpDir, id) const files: { body: string metadata: string + response: string relay?: string request?: string } = { body: `${prefix}.body.json`, metadata: `${prefix}.meta.json`, + response: `${prefix}.response.json`, } if (input.payload !== undefined) files.relay = `${prefix}.relay.json` if (input.request !== undefined) files.request = `${prefix}.request.json` @@ -503,6 +518,7 @@ async function dumpRequest(input: { route: input.route, status: input.status, error: input.error, + tag: input.tag, bodyBytes: input.bodyText.length, relayBytes: input.relayBytes, bodyHash: hashText(input.bodyText), @@ -551,6 +567,11 @@ async function dumpRequest(input: { relayLog( `dump failed: ${error instanceof Error ? error.message : String(error)}`, ) + return null + } + return { + responsePath: files.response, + ...(input.tag ? { tag: input.tag } : {}), } } @@ -563,11 +584,12 @@ export async function dumpDirectRequest(input: { url?: string method?: string headers?: DumpHeaders -}) { - if (!dumpEnabled) return + tag?: DumpTag +}): Promise { + if (!dumpEnabled) return null const previousKey = directDumpPreviousKey(input) const previousBodyText = directDumpPreviousBodies.get(previousKey) - await dumpRequest({ + const handle = await dumpRequest({ affinity: input.affinity, transport: 'direct', route: input.route, @@ -580,8 +602,10 @@ export async function dumpDirectRequest(input: { method: input.method, headers: input.headers, }, + tag: input.tag, }) rememberDirectDumpBody(previousKey, input.bodyText) + return handle } export async function dumpRelayRequest(input: { @@ -594,8 +618,9 @@ export async function dumpRelayRequest(input: { previousBodyText?: string payload: unknown relayBytes: number -}) { - await dumpRequest({ + tag?: DumpTag +}): Promise { + return dumpRequest({ affinity: input.affinity, transport: input.transport, protocol: input.protocol, @@ -605,5 +630,37 @@ export async function dumpRelayRequest(input: { previousBodyText: input.previousBodyText, payload: input.payload, relayBytes: input.relayBytes, + tag: input.tag, }) } + +export async function dumpResponseArtifact( + handle: DumpHandle | null, + input: { status: number; message: unknown }, +): Promise { + if (!handle) return + const message = + input.message != null && + typeof input.message === 'object' && + !Array.isArray(input.message) + ? (input.message as Record) + : {} + const artifact: Record = { status: input.status } + if (typeof message.id === 'string' && message.id.length > 0) + artifact.message_id = message.id + if (typeof message.model === 'string' && message.model.length > 0) + artifact.model = message.model + if (Object.hasOwn(message, 'usage')) artifact.usage = message.usage + if (Object.hasOwn(message, 'diagnostics')) + artifact.diagnostics = message.diagnostics + try { + await writeDumpFile( + handle.responsePath, + `${JSON.stringify(artifact, null, 2)}\n`, + ) + } catch (error) { + relayLog( + `dump response failed: ${error instanceof Error ? error.message : String(error)}`, + ) + } +} diff --git a/packages/core/src/relay.ts b/packages/core/src/relay.ts index 5fb0169d..085cd46d 100644 --- a/packages/core/src/relay.ts +++ b/packages/core/src/relay.ts @@ -1109,6 +1109,7 @@ export async function sendViaRelay(options: { * within an attempt are ignored. */ onResponseHeaders?: (headers: Headers) => void + onDumpCreated?: (handle: { responsePath: string; tag?: 'cachekeep' }) => void setTimeoutImpl?: typeof globalThis.setTimeout clearTimeoutImpl?: typeof globalThis.clearTimeout }): Promise { @@ -1122,6 +1123,7 @@ export async function sendViaRelay(options: { affinity: explicitAffinity, optimisticResponse, onResponseHeaders, + onDumpCreated, setTimeoutImpl = globalThis.setTimeout, clearTimeoutImpl = globalThis.clearTimeout, } = options @@ -1237,7 +1239,7 @@ export async function sendViaRelay(options: { `used relay transport=${result.transport} protocol=${result.protocol} mode=${result.payload.mode} status=${result.response.status} session=${shortAffinity(affinity)} bodyBytes=${bodyText.length} relayBytes=${actualPayloadBytes}`, ) const dumpStart = perfNowMs() - await dumpRelayRequest({ + const dumpHandle = await dumpRelayRequest({ affinity, transport: result.transport, protocol: result.protocol, @@ -1248,6 +1250,9 @@ export async function sendViaRelay(options: { payload: result.payload, relayBytes: actualPayloadBytes, }) + try { + if (dumpHandle) onDumpCreated?.(dumpHandle) + } catch {} relayPerfLog('dump', { session: shortAffinity(affinity), ms: formatMs(perfNowMs() - dumpStart), diff --git a/packages/core/src/tests/dump.test.ts b/packages/core/src/tests/dump.test.ts index 2be787c9..37e98807 100644 --- a/packages/core/src/tests/dump.test.ts +++ b/packages/core/src/tests/dump.test.ts @@ -1,5 +1,6 @@ import { afterEach, describe, expect, test } from 'bun:test' import { + chmod, lstat, mkdir, mkdtemp, @@ -13,7 +14,15 @@ import { } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' -import { sweepDumpDirectory, writeDumpFile } from '../dump' +import { + dumpDirectRequest, + dumpRelayRequest, + dumpResponseArtifact, + resetDumpState, + setDumpEnabled, + sweepDumpDirectory, + writeDumpFile, +} from '../dump' const dumpDirs: string[] = [] const dumpLinks: string[] = [] @@ -22,12 +31,13 @@ const originalDumpDir = process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR function dumpArtifactName( id: number, - kind: 'body' | 'meta' | 'relay' | 'request' = 'body', + kind: 'body' | 'meta' | 'relay' | 'request' | 'response' = 'body', ) { return `2026-07-17T12-00-00-000Z-${String(id).padStart(6, '0')}-session-direct.${kind}.json` } afterEach(async () => { + resetDumpState() if (originalDumpMaxBytes === undefined) { delete process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_MAX_BYTES } else { @@ -48,6 +58,132 @@ afterEach(async () => { ) }) +test('request dumps return response handles only when enabled', async () => { + const dumpDir = await mkdtemp( + join(tmpdir(), 'opencode-anthropic-auth-dumps-test-'), + ) + dumpDirs.push(dumpDir) + process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = dumpDir + + expect( + await dumpDirectRequest({ + affinity: 'ses-a', + bodyText: '{}', + route: 'oauth', + }), + ).toBeNull() + setDumpEnabled(true) + const direct = await dumpDirectRequest({ + affinity: 'ses-a', + bodyText: '{}', + route: 'oauth', + }) + const relay = await dumpRelayRequest({ + affinity: 'ses-b', + transport: 'http', + protocol: 1, + mode: 'full_sync', + bodyText: '{}', + payload: {}, + relayBytes: 2, + }) + expect(direct?.responsePath).toMatch(/\.response\.json$/) + expect(relay?.responsePath).toMatch(/\.response\.json$/) + expect(await lstat(direct!.responsePath).catch(() => null)).toBeNull() +}) + +test('failed request dumps return no handle or orphan response artifact', async () => { + if (process.getuid?.() === 0) return + const dumpDir = await mkdtemp( + join(tmpdir(), 'opencode-anthropic-auth-dumps-test-'), + ) + dumpDirs.push(dumpDir) + await chmod(dumpDir, 0o500) + try { + process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = dumpDir + setDumpEnabled(true) + const handle = await dumpDirectRequest({ + affinity: 'ses-failure', + bodyText: '{}', + }) + expect(handle).toBeNull() + expect( + (await readdir(dumpDir)).some((name) => name.endsWith('.response.json')), + ).toBe(false) + } finally { + await chmod(dumpDir, 0o700) + } +}) + +test('response artifacts sanitize message fields and preserve diagnostics presence', async () => { + const dumpDir = await mkdtemp( + join(tmpdir(), 'opencode-anthropic-auth-dumps-test-'), + ) + dumpDirs.push(dumpDir) + process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = dumpDir + setDumpEnabled(true) + const handle = await dumpDirectRequest({ affinity: 'ses-a', bodyText: '{}' }) + expect(handle).not.toBeNull() + await dumpResponseArtifact(handle, { + status: 200, + message: { + id: 'msg_provider', + model: 'claude-opus-4-7', + usage: { input_tokens: 1 }, + diagnostics: null, + content: [{ text: 'secret' }], + }, + }) + const artifact = JSON.parse(await readFile(handle!.responsePath, 'utf8')) + expect(artifact).toEqual({ + status: 200, + message_id: 'msg_provider', + model: 'claude-opus-4-7', + usage: { input_tokens: 1 }, + diagnostics: null, + }) + expect(JSON.stringify(artifact)).not.toContain('secret') +}) + +test('tagged prewarm dumps include the tag in filenames and metadata', async () => { + const dumpDir = await mkdtemp( + join(tmpdir(), 'opencode-anthropic-auth-dumps-test-'), + ) + dumpDirs.push(dumpDir) + process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = dumpDir + setDumpEnabled(true) + const handle = await dumpDirectRequest({ + affinity: 'ses-a', + bodyText: '{}', + tag: 'cachekeep', + }) + expect(handle?.tag).toBe('cachekeep') + expect(handle?.responsePath).toContain('-prewarm-cachekeep-') + const files = await readdir(dumpDir) + const metadata = JSON.parse( + await readFile( + join(dumpDir, files.find((name) => name.endsWith('.meta.json'))!), + 'utf8', + ), + ) + expect(metadata.tag).toBe('cachekeep') +}) + +test('dump sweep recognizes response artifacts', async () => { + const dumpDir = await mkdtemp( + join(tmpdir(), 'opencode-anthropic-auth-dumps-test-'), + ) + dumpDirs.push(dumpDir) + process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = dumpDir + const response = join(dumpDir, dumpArtifactName(1, 'response')) + await writeFile(response, '12345678') + await utimes(response, new Date(1_000), new Date(1_000)) + expect(await sweepDumpDirectory({ dumpDir, maxBytes: 1 })).toEqual({ + removed: 1, + freedBytes: 8, + }) +}) + test('dump sweep deletes oldest files until the directory is under its cap', async () => { const dumpDir = await mkdtemp( join(tmpdir(), 'opencode-anthropic-auth-dumps-test-'), diff --git a/packages/opencode/README.md b/packages/opencode/README.md index a2ab7709..866f1632 100644 --- a/packages/opencode/README.md +++ b/packages/opencode/README.md @@ -393,6 +393,74 @@ Request bodies, headers, and tokens remain in memory. A lease-backed file under Pre-warm requests preserve explicit cache anchors but remove response-only fields that Anthropic rejects with `max_tokens: 0`, such as streaming, enabled thinking, structured output format, and forced/any tool choice. The feature works only while OpenCode or Pi is running and the machine is awake, and cache writes are still billed when the cache entry is no longer warm. +### Cache diagnostics (beta) + +The `cache-diagnosis-2026-04-07` beta is measure-only. It asks Anthropic to report prompt-cache diagnostics; it does not change cache controls or routing. OpenCode captures the provider's top-level response ID as an opaque string and sends it as `diagnostics.previous_message_id` on the next request in the same session. The first request sends `null`. + +The `MC-CACHE-DIAG ` line is a versioned, one-line JSON record. Version `2` contains these fields: + +| Field | Type | Source | +| --- | --- | --- | +| `v` | `2` | Capture schema | +| `source` | string | Observation path; known values are `"turn"` and `"prewarm_cachekeep"`, but consumers must tolerate future values | +| `synthetic` | boolean | Whether this observation was generated by plugin machinery rather than a real turn | +| `account_id` | string | Persisted plugin-internal OAuth account identifier used by routing and the sidebar; stable across restarts and token refreshes, so consumers may key timelines on it. Opaque mixed key space (the main account is a sentinel string, fallbacks are UUIDs) — never validate its shape | +| `betas_hash` | 16-character lowercase hex | xxHash64 (seed `0`) of the sorted `anthropic-beta` list actually sent, truncated to its first 16 hexadecimal characters | +| `requested_model` | string, optional | Sent request-body model, present only when it differs from the served Anthropic response model | +| `session_id` | string | OpenCode session affinity | +| `ts_ms_received` | integer | Local response receipt time | +| `model` | string | Anthropic response | +| `is_subagent` | boolean | Original request context | +| `ttl_sent` | `"1h" \| "5m" \| null` | Last valid cache breakpoint in the sent body | +| `cache_read` | number | `usage.cache_read_input_tokens` | +| `cache_creation` | number | `usage.cache_creation_input_tokens` | +| `input_tokens` | number | `usage.input_tokens` | +| `ephemeral_5m_tokens` | number | `usage.cache_creation.ephemeral_5m_input_tokens` | +| `ephemeral_1h_tokens` | number | `usage.cache_creation.ephemeral_1h_input_tokens` | +| `message_id` | string | Anthropic response top-level `id` | +| `previous_message_id` | string \| null | Sent `diagnostics.previous_message_id` | +| `diag_state` | `"absent" \| "server_null" \| "pending" \| "populated"` | Anthropic response envelope | +| `miss_reason` | string, optional | `diagnostics.cache_miss_reason.type` | +| `cache_missed_input_tokens` | number, optional | `diagnostics.cache_miss_reason.cache_missed_input_tokens` | + +`cache_missed_input_tokens` is a byte-derived, pre-tokenization magnitude indicator, not a token count: it can differ from and occasionally exceed `input_tokens`, so never difference it against usage fields. + +Consumers classifying these records should check `cache_read` before `miss_reason`: `cache_read > 0` is a fact and `miss_reason` is an interpretation, and facts win. On multi-turn agent traffic a populated `*_changed` reason routinely coexists with a full prefix hit (each turn appends messages), so a populated-first classifier inverts every such record into a false miss. + +Known sources map to `synthetic` as follows: + +| `source` | `synthetic` | +| --- | --- | +| `turn` | `false` | +| `prewarm_cachekeep` | `true` | + +`synthetic` wins on conflict. A disagreement for a known source is an emitter defect: OpenCode writes one warning and still emits the record with the supplied `synthetic` value. Unknown future sources have no mapping and must not be rejected by consumers. + +The first observation of each `betas_hash` in a process also writes `MC-CACHE-DIAG-BETAS {"hash":"…","betas":[…]}` on the `cache-diagnostics` logger channel. Its sorted beta list makes an observed hash interpretable without reconstructing headers; repeated hashes do not emit another side-channel line. + +The trailing space in `MC-CACHE-DIAG ` is load-bearing: a record line has a space after `MC-CACHE-DIAG`, while the beta side-channel line has a hyphen. Do not trim the delimiter or write a trimming consumer. `grep -c "MC-CACHE-DIAG"` over-counts because it includes beta lines; count records with `grep -c "MC-CACHE-DIAG "` or an anchored equivalent. + +The four diagnostic states are: + +| State | Response shape | +| --- | --- | +| `absent` | No `diagnostics` property | +| `server_null` | `diagnostics: null` | +| `pending` | `diagnostics.cache_miss_reason: null` | +| `populated` | An object `diagnostics.cache_miss_reason` with a non-empty string `type`, including `"unavailable"` | + +Only valid Message envelopes produce an `MC-CACHE-DIAG ` record. Malformed or error responses are not mapped to `absent`; they retain ordinary response handling and dumps. `ttl_sent` is reduced from the last valid breakpoint in the actual body sent. A short-gap `previous_message_not_found` result is also recorded as a canary when the previous provider ID was captured less than five minutes earlier. + +`unavailable` is expected when any prompt-affecting parameter changes between requests. This includes the active Anthropic beta-header set; structured-output requests use a different beta set and can therefore produce `unavailable`. Diagnostics are scoped to the provider's cache fingerprint, organization, and workspace. Fingerprints expire, so a later request can miss after a long gap even when the body is unchanged. These records measure the result; they do not guarantee a cache hit. + +Diagnostics comparison requires a cacheable prefix. Requests below the model's cacheable minimum or without cache breakpoints return `diagnostics: null` even when the request genuinely changed; consumers must gate interpretation of `null` on observed cache activity. + +`diag_state` is always a string and never JSON null: `absent | server_null | pending | populated`. + +Version 1 records come from the unversioned-source era and cannot distinguish prewarms from turns; consumers must treat their source as unknown and cannot split machinery from traffic retroactively. Version 2 always states the source. + +When request dumps are enabled, each response gets a `.response.json` artifact containing status and parsed response metadata, but no response content. Cache keepalive prewarms are tagged `-prewarm-cachekeep` in dump filenames and metadata. If Prime prewarming is enabled in a build that supports it, those artifacts use `-prewarm-prime`. Treat request bodies and related dump files as sensitive local debugging data. + ## Claude fast mode Both OpenCode and Pi packages can persistently request Anthropic fast mode for supported Opus models: diff --git a/packages/opencode/src/cache-diagnostics.ts b/packages/opencode/src/cache-diagnostics.ts new file mode 100644 index 00000000..8509e52a --- /dev/null +++ b/packages/opencode/src/cache-diagnostics.ts @@ -0,0 +1,322 @@ +import { + setBounded, + stickyRetryAfterWithJitter, +} from '@cortexkit/anthropic-auth-core' + +export const CACHE_DIAGNOSTICS_BETA = 'cache-diagnosis-2026-04-07' +export const CACHE_DIAGNOSTICS_LOG_PREFIX = 'MC-CACHE-DIAG ' +export const CACHE_DIAGNOSTICS_SESSION_LIMIT = 1_000 +export const CACHE_DIAGNOSTICS_SHORT_GAP_MS = 5 * 60_000 + +export type CacheDiagnosticsState = + | 'absent' + | 'server_null' + | 'pending' + | 'populated' + +export type CacheDiagnosticsSource = string + +export const CACHE_DIAGNOSTICS_SOURCE_SYNTHETIC = { + turn: false, + prewarm_cachekeep: true, +} as const + +export const CACHE_DIAGNOSTICS_BETAS_LIMIT = 1_000 + +export type CacheDiagnosticsRecord = { + v: 2 + source: CacheDiagnosticsSource + account_id: string + synthetic: boolean + betas_hash: string + requested_model?: string + session_id: string + ts_ms_received: number + model: string + is_subagent: boolean + ttl_sent: '1h' | '5m' | null + cache_read: number + cache_creation: number + input_tokens: number + ephemeral_5m_tokens: number + ephemeral_1h_tokens: number + message_id: string + previous_message_id: string | null + diag_state: CacheDiagnosticsState + miss_reason?: string + cache_missed_input_tokens?: number +} + +export type CacheDiagnosticsRequestContext = { + sessionId: string + previousMessageId: string | null + previousMessageReceivedAt?: number + isSubagent: boolean + ttlSent: '1h' | '5m' | null +} + +type CapturedMessage = { messageId: string; receivedAt: number } + +export class CacheDiagnosticsTracker { + private readonly entries = new Map() + + previousFor(sessionId: string): CapturedMessage | null { + return this.entries.get(sessionId) ?? null + } + + capture(sessionId: string, messageId: string, receivedAt: number): void { + setBounded( + this.entries, + sessionId, + { messageId, receivedAt }, + CACHE_DIAGNOSTICS_SESSION_LIMIT, + ) + } +} + +export class CacheDiagnosticsBetaTracker { + private readonly hashes = new Map() + + capture(hash: string, betas: string[]): string | null { + if (this.hashes.has(hash)) return null + setBounded(this.hashes, hash, true, CACHE_DIAGNOSTICS_BETAS_LIMIT) + return `MC-CACHE-DIAG-BETAS ${JSON.stringify({ + hash, + betas: [...betas].sort(), + })}` + } +} + +export function copyCacheDiagnosticsContext( + contexts: WeakMap, + source: Response, + destination: Response, +): void { + const context = contexts.get(source) + if (context) contexts.set(destination, context) +} + +export async function withStickyRetryAfter( + response: Response, + sessionId: string, + retryAfterSeconds: number, + streamingRateLimit: boolean, + contexts: WeakMap, +) { + const headers = new Headers(response.headers) + headers.set( + 'retry-after', + String(stickyRetryAfterWithJitter(sessionId, retryAfterSeconds)), + ) + if (streamingRateLimit) { + await response.body?.cancel().catch(() => {}) + headers.set('content-type', 'application/json') + return new Response( + JSON.stringify({ + type: 'error', + error: { + type: 'rate_limit_error', + message: + 'Sticky OAuth account five-hour quota resets shortly; retaining session affinity.', + }, + }), + { status: 429, headers }, + ) + } + const retriedResponse = new Response(response.body, { + status: response.status, + statusText: response.statusText, + headers, + }) + copyCacheDiagnosticsContext(contexts, response, retriedResponse) + return retriedResponse +} + +export function applyCacheDiagnosticsOptIn( + body: Record, + previousMessageId: string | null, +): void { + body.diagnostics = { previous_message_id: previousMessageId } +} + +function isRecord(value: unknown): value is Record { + return value != null && typeof value === 'object' && !Array.isArray(value) +} + +function isBreakpoint(value: unknown): value is Record { + return isRecord(value) && value.type === 'ephemeral' +} + +function appendBreakpoint(value: unknown, output: Array<'1h' | '5m'>) { + if (!isBreakpoint(value)) return + output.push(value.ttl === '1h' ? '1h' : '5m') +} + +function appendArrayBreakpoints(value: unknown, output: Array<'1h' | '5m'>) { + if (!Array.isArray(value)) return + for (const entry of value) { + if (isRecord(entry)) appendBreakpoint(entry.cache_control, output) + } +} + +export function summarizeCacheTtl(body: unknown): '1h' | '5m' | null { + if (!isRecord(body)) return null + const breakpoints: Array<'1h' | '5m'> = [] + appendBreakpoint(body.cache_control, breakpoints) + appendArrayBreakpoints(body.system, breakpoints) + if (Array.isArray(body.messages)) { + for (const message of body.messages) { + if (isRecord(message)) + appendArrayBreakpoints(message.content, breakpoints) + } + } + return breakpoints.at(-1) ?? null +} + +function classifyDiagnostics(message: Record): { + state: CacheDiagnosticsState + missReason?: string + cacheMissedInputTokens?: number +} | null { + if (!Object.hasOwn(message, 'diagnostics')) return { state: 'absent' } + const diagnostics = message.diagnostics + if (diagnostics === null) return { state: 'server_null' } + if (!isRecord(diagnostics)) return null + if (!Object.hasOwn(diagnostics, 'cache_miss_reason')) return null + const reason = diagnostics.cache_miss_reason + if (reason === null) return { state: 'pending' } + if (!isRecord(reason) || typeof reason.type !== 'string' || !reason.type) + return null + const missed = reason.cache_missed_input_tokens + if (missed !== undefined && typeof missed !== 'number') return null + return { + state: 'populated', + missReason: reason.type, + cacheMissedInputTokens: missed, + } +} + +function numericUsage( + usage: Record, + key: string, +): number | null { + return typeof usage[key] === 'number' ? usage[key] : null +} + +export function buildCacheDiagnosticsRecord(input: { + request: CacheDiagnosticsRequestContext + source: CacheDiagnosticsSource + accountId: string + synthetic: boolean + betasHash: string + requestedModel?: string + onWarning?: (message: string) => void + message: unknown + receivedAt: number +}): { + record?: CacheDiagnosticsRecord + messageId?: string + canary?: { messageId: string; previousMessageId: string } +} { + if ( + typeof input.source !== 'string' || + input.source.length === 0 || + typeof input.accountId !== 'string' || + input.accountId.length === 0 || + typeof input.synthetic !== 'boolean' || + typeof input.betasHash !== 'string' || + input.betasHash.length === 0 + ) + return {} + const expectedSynthetic = + CACHE_DIAGNOSTICS_SOURCE_SYNTHETIC[ + input.source as keyof typeof CACHE_DIAGNOSTICS_SOURCE_SYNTHETIC + ] + if ( + expectedSynthetic !== undefined && + expectedSynthetic !== input.synthetic + ) { + input.onWarning?.( + `cache diagnostics source/synthetic mismatch: source=${input.source} expected=${expectedSynthetic} actual=${input.synthetic}`, + ) + } + const receivedAt = Math.trunc(input.receivedAt) + if (!Number.isFinite(input.receivedAt) || receivedAt !== input.receivedAt) + return {} + if (!isRecord(input.message)) return {} + const messageId = input.message.id + const model = input.message.model + const usage = input.message.usage + if (typeof messageId !== 'string' || messageId.length === 0) return {} + if (typeof model !== 'string' || model.length === 0 || !isRecord(usage)) + return {} + const inputTokens = numericUsage(usage, 'input_tokens') + const cacheRead = numericUsage(usage, 'cache_read_input_tokens') + const cacheCreation = numericUsage(usage, 'cache_creation_input_tokens') + const cacheCreationDetails = usage.cache_creation + if ( + inputTokens === null || + cacheRead === null || + cacheCreation === null || + !isRecord(cacheCreationDetails) + ) + return {} + const ephemeral5m = numericUsage( + cacheCreationDetails, + 'ephemeral_5m_input_tokens', + ) + const ephemeral1h = numericUsage( + cacheCreationDetails, + 'ephemeral_1h_input_tokens', + ) + if (ephemeral5m === null || ephemeral1h === null) return {} + const diagnostics = classifyDiagnostics(input.message) + if (!diagnostics) return {} + + const record: CacheDiagnosticsRecord = { + v: 2, + source: input.source, + account_id: input.accountId, + synthetic: input.synthetic, + betas_hash: input.betasHash, + session_id: input.request.sessionId, + ts_ms_received: receivedAt, + model, + is_subagent: input.request.isSubagent, + ttl_sent: input.request.ttlSent, + cache_read: cacheRead, + cache_creation: cacheCreation, + input_tokens: inputTokens, + ephemeral_5m_tokens: ephemeral5m, + ephemeral_1h_tokens: ephemeral1h, + message_id: messageId, + previous_message_id: input.request.previousMessageId, + diag_state: diagnostics.state, + } + if (diagnostics.missReason !== undefined) + record.miss_reason = diagnostics.missReason + if (diagnostics.cacheMissedInputTokens !== undefined) + record.cache_missed_input_tokens = diagnostics.cacheMissedInputTokens + if (input.requestedModel !== undefined && input.requestedModel !== model) + record.requested_model = input.requestedModel + + const canary = + diagnostics.missReason === 'previous_message_not_found' && + input.request.previousMessageId !== null && + input.request.previousMessageReceivedAt !== undefined && + input.receivedAt - input.request.previousMessageReceivedAt < + CACHE_DIAGNOSTICS_SHORT_GAP_MS && + input.receivedAt >= input.request.previousMessageReceivedAt + ? { + messageId, + previousMessageId: input.request.previousMessageId, + } + : undefined + return { record, messageId, ...(canary ? { canary } : {}) } +} + +export function formatCacheDiagnosticsLogLine( + record: CacheDiagnosticsRecord, +): string { + return `${CACHE_DIAGNOSTICS_LOG_PREFIX}${JSON.stringify(record)}` +} diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index abced905..5c5ec5a5 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -25,10 +25,13 @@ import { CLAUDE_LOGGING_COMMAND_NAME, CLAUDE_QUOTAS_COMMAND_NAME, CLAUDE_ROUTING_COMMAND_NAME, + computeXxhash64Hex, createEmptyStorage, createStickyNoRouteResponse, + type DumpHandle, decideStickyQuotaFailure, dumpDirectRequest, + dumpResponseArtifact, exchange, executeAccountCommand, executeCache1hCommand, @@ -131,11 +134,23 @@ import { setRoutingMode, shouldFallbackStatus, stickyQuotaSnapshotIsFresh, - stickyRetryAfterWithJitter, stickyRouteFamilyForModel, tokenFingerprint, } from '@cortexkit/anthropic-auth-core' import type { Plugin } from '@opencode-ai/plugin' +import { + applyCacheDiagnosticsOptIn, + buildCacheDiagnosticsRecord, + CACHE_DIAGNOSTICS_BETA, + CacheDiagnosticsBetaTracker, + type CacheDiagnosticsRequestContext, + type CacheDiagnosticsSource, + CacheDiagnosticsTracker, + copyCacheDiagnosticsContext, + formatCacheDiagnosticsLogLine, + summarizeCacheTtl, + withStickyRetryAfter, +} from './cache-diagnostics.ts' import { FableFallbackManager, type FableFallbackPlan, @@ -1162,6 +1177,150 @@ const anthropicAuthPlugin = async ( }, }) fallbackManager.startBackgroundRefresh() + const cacheDiagnosticsTracker = new CacheDiagnosticsTracker() + const cacheDiagnosticsBetaTracker = new CacheDiagnosticsBetaTracker() + type CacheDiagnosticsResponse = { + request?: CacheDiagnosticsRequestContext + trackSessionId?: string + source: CacheDiagnosticsSource + accountId: string + synthetic: boolean + betasHash: string + betas: string[] + requestedModel?: string + dump: DumpHandle | null + status: number + streaming: boolean + dumpWrite: Promise + } + const cacheDiagnosticsResponses = new WeakMap< + Response, + CacheDiagnosticsResponse + >() + const cacheKeepDiagnosticsRequests = new Map< + string, + CacheDiagnosticsRequestContext & { + accountId: string + synthetic: boolean + betasHash?: string + betas?: string[] + requestedModel?: string + } + >() + + async function getCacheDiagnosticsBetas(headers: Headers) { + const betas = (headers.get('anthropic-beta') ?? '') + .split(',') + .map((beta) => beta.trim()) + .filter(Boolean) + .sort() + return { + betas, + betasHash: await computeXxhash64Hex(betas.join(',')), + } + } + + function observeCacheDiagnosticsMessage(input: { + source: CacheDiagnosticsSource + accountId: string + synthetic: boolean + betasHash: string + betas: string[] + requestedModel?: string + request?: CacheDiagnosticsRequestContext + trackSessionId?: string + status: number + message: unknown + receivedAt: number + dump?: DumpHandle | null + dumpWrite?: Promise + }) { + try { + if (input.dumpWrite) { + void input.dumpWrite + .then(() => + dumpResponseArtifact(input.dump ?? null, { + status: input.status, + message: input.message, + }), + ) + .catch(() => {}) + } + if (!input.request) return + const observed = buildCacheDiagnosticsRecord({ + request: input.request, + source: input.source, + accountId: input.accountId, + synthetic: input.synthetic, + betasHash: input.betasHash, + requestedModel: input.requestedModel, + onWarning: (message) => logger.warn('cache-diagnostics', message), + message: input.message, + receivedAt: input.receivedAt, + }) + if (!observed.record || !observed.messageId) { + logger.debug('cache-diagnostics', 'skipped invalid response envelope', { + status: input.status, + }) + return + } + logger.info( + 'cache-diagnostics', + formatCacheDiagnosticsLogLine(observed.record), + ) + const betaLine = cacheDiagnosticsBetaTracker.capture( + input.betasHash, + input.betas, + ) + if (betaLine) logger.info('cache-diagnostics', betaLine) + if (observed.canary) { + logger.warn( + 'cache-diagnostics', + 'short-gap previous_message_not_found', + { + message_id: observed.canary.messageId, + previous_message_id: observed.canary.previousMessageId, + }, + ) + } + if (input.trackSessionId) { + cacheDiagnosticsTracker.capture( + input.trackSessionId, + observed.messageId, + input.receivedAt, + ) + } + } catch (error) { + logger.debug('cache-diagnostics', 'response observation failed', { + status: input.status, + error: error instanceof Error ? error.message : String(error), + }) + } + } + + function observeCacheDiagnosticsResponse( + response: Response, + message: unknown, + ) { + const context = cacheDiagnosticsResponses.get(response) + if (!context) return + observeCacheDiagnosticsMessage({ + ...context, + message, + receivedAt: Date.now(), + }) + } + + function attachCacheDiagnosticsResponse( + response: Response, + input: Omit, + ) { + const dumpWrite = Promise.resolve( + dumpResponseArtifact(input.dump, { status: input.status, message: null }), + ).catch(() => {}) + cacheDiagnosticsResponses.set(response, { ...input, dumpWrite }) + } + let latestRefreshMainAccessToken: (() => Promise) | null = null const cacheKeepRegistry = new CacheKeepSessionRegistry({ directory: @@ -1179,6 +1338,82 @@ const anthropicAuthPlugin = async ( await cacheKeepRegistry.publish(sessions) aggregateCacheKeepSessions = await cacheKeepRegistry.list(sessions) }, + prepareBody: (bodyText, target) => { + if ( + !new Headers(target.headers) + .get('anthropic-beta') + ?.split(',') + .map((beta) => beta.trim()) + .includes(CACHE_DIAGNOSTICS_BETA) + ) { + return bodyText + } + try { + const body = JSON.parse(bodyText) as Record + const previous = cacheDiagnosticsTracker.previousFor(target.id) + const previousMessageId = previous?.messageId ?? null + applyCacheDiagnosticsOptIn(body, previousMessageId) + cacheKeepDiagnosticsRequests.set(target.id, { + sessionId: target.id, + previousMessageId, + ...(previous + ? { previousMessageReceivedAt: previous.receivedAt } + : {}), + isSubagent: target.isSubagent, + ttlSent: summarizeCacheTtl(body), + accountId: target.oauthAccountId ?? 'main', + synthetic: true, + requestedModel: + typeof body.model === 'string' ? body.model : undefined, + }) + return JSON.stringify(body) + } catch { + return bodyText + } + }, + onResponse: ({ target, bodyText, status, data, receivedAt }) => { + const prepared = cacheKeepDiagnosticsRequests.get(target.id) + if (!prepared?.betasHash || !prepared.betas) { + cacheKeepDiagnosticsRequests.delete(target.id) + return + } + try { + const sentBody = JSON.parse(bodyText) + const diagnostics = + sentBody && typeof sentBody === 'object' && !Array.isArray(sentBody) + ? (sentBody as { diagnostics?: unknown }).diagnostics + : undefined + const previousMessageId = + diagnostics && + typeof diagnostics === 'object' && + !Array.isArray(diagnostics) && + (diagnostics as { previous_message_id?: unknown }) + .previous_message_id === prepared.previousMessageId + ? prepared.previousMessageId + : undefined + if (previousMessageId === undefined) return + observeCacheDiagnosticsMessage({ + source: 'prewarm_cachekeep', + accountId: prepared.accountId, + synthetic: prepared.synthetic, + betasHash: prepared.betasHash, + betas: prepared.betas, + requestedModel: prepared.requestedModel, + request: { + ...prepared, + previousMessageId, + ttlSent: summarizeCacheTtl(sentBody), + }, + trackSessionId: target.id, + status, + message: data, + receivedAt, + }) + } catch { + } finally { + cacheKeepDiagnosticsRequests.delete(target.id) + } + }, prepareHeaders: async (headers, target) => { let accessToken: string | undefined const accountId = target.oauthAccountId @@ -1242,6 +1477,12 @@ const anthropicAuthPlugin = async ( ]), ) if (parsedBody.speed === 'fast') addFastModeBetaHeader(headers) + const prepared = cacheKeepDiagnosticsRequests.get(target.id) + if (prepared) { + const { betas, betasHash } = await getCacheDiagnosticsBetas(headers) + prepared.betas = betas + prepared.betasHash = betasHash + } } catch { setOAuthHeaders(headers, accessToken) } @@ -3252,12 +3493,18 @@ const anthropicAuthPlugin = async ( bytes, rateLimited: true, }) + const inspectedResponse = new Response(stream, { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }) + copyCacheDiagnosticsContext( + cacheDiagnosticsResponses, + response, + inspectedResponse, + ) return { - response: new Response(stream, { - status: response.status, - statusText: response.statusText, - headers: response.headers, - }), + response: inspectedResponse, rateLimited: true, } } @@ -3284,12 +3531,18 @@ const anthropicAuthPlugin = async ( bytes, rateLimited: false, }) + const inspectedResponse = new Response(stream, { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }) + copyCacheDiagnosticsContext( + cacheDiagnosticsResponses, + response, + inspectedResponse, + ) return { - response: new Response(stream, { - status: response.status, - statusText: response.statusText, - headers: response.headers, - }), + response: inspectedResponse, rateLimited: false, } } @@ -3332,6 +3585,8 @@ const anthropicAuthPlugin = async ( requestHeaders.delete('x-session-affinity') requestHeaders.delete('x-opencode-session') let body = init?.body + let streaming = false + let dump: DumpHandle | null = null const originalBytes = typeof body === 'string' ? body.length : undefined @@ -3358,6 +3613,9 @@ const anthropicAuthPlugin = async ( mergeAnthropicBetas(requestHeaders.get('anthropic-beta'), []), ) if (fastModeRequested) addFastModeBetaHeader(requestHeaders) + try { + streaming = JSON.parse(body).stream === true + } catch {} trace?.mark('rewrite_body', { route, ms: roundMs(nowMs() - rewriteStart), @@ -3372,6 +3630,8 @@ const anthropicAuthPlugin = async ( configureApiRouteHeaders(requestHeaders, account) } + const cacheDiagnosticsBetas = + await getCacheDiagnosticsBetas(requestHeaders) const rewritten = rewriteUrl(input, { baseURL: account.baseURL }) const sendStart = nowMs() let response: Response @@ -3398,7 +3658,7 @@ const anthropicAuthPlugin = async ( throw error } if (typeof body === 'string') { - await dumpDirectRequest({ + dump = await dumpDirectRequest({ affinity: directAffinity, route, status: response.status, @@ -3409,6 +3669,16 @@ const anthropicAuthPlugin = async ( headers: requestHeaders, }) } + attachCacheDiagnosticsResponse(response, { + source: 'turn', + accountId: account.id, + synthetic: false, + ...cacheDiagnosticsBetas, + requestedModel: parseRequestModel(body), + dump, + status: response.status, + streaming, + }) trace?.mark('send_headers_received', { route, ms: roundMs(nowMs() - sendStart), @@ -3445,6 +3715,17 @@ const anthropicAuthPlugin = async ( requestHeaders.delete('x-session-affinity') requestHeaders.delete('x-opencode-session') let body = init?.body + const previousDiagnosticsMessage = relayAffinity + ? cacheDiagnosticsTracker.previousFor(relayAffinity) + : null + const cacheDiagnosticsPreviousMessageId = + previousDiagnosticsMessage?.messageId ?? null + let cacheDiagnosticsRequest: + | CacheDiagnosticsRequestContext + | undefined + let streaming = false + let directDump: DumpHandle | null = null + let relayDump: DumpHandle | null = null let modelForIdentity: string | undefined if (body && typeof body === 'string') { const modelParseStart = nowMs() @@ -3498,6 +3779,7 @@ const anthropicAuthPlugin = async ( identity, hybridStandbyAnchor: standbyCacheAnchor, serverSideFallbackEnabled: fallbackMode === 'server', + cacheDiagnosticsPreviousMessageId, perf: (stage, data) => { trace?.mark(`rewrite_body_${stage}`, { route, ...data }) if ( @@ -3530,10 +3812,42 @@ const anthropicAuthPlugin = async ( } const headerBodyParseStart = nowMs() try { + const finalBody = JSON.parse(body) as Record setOAuthHeaders(requestHeaders, accessToken, { - body: JSON.parse(body), + body: finalBody, identity, }) + const diagnostics = finalBody.diagnostics + const sentPreviousMessageId = + diagnostics && + typeof diagnostics === 'object' && + !Array.isArray(diagnostics) && + (diagnostics as { previous_message_id?: unknown }) + .previous_message_id === cacheDiagnosticsPreviousMessageId + ? cacheDiagnosticsPreviousMessageId + : undefined + if ( + sentPreviousMessageId !== undefined && + requestHeaders + .get('anthropic-beta') + ?.split(',') + .map((beta) => beta.trim()) + .includes(CACHE_DIAGNOSTICS_BETA) + ) { + cacheDiagnosticsRequest = { + sessionId: relayAffinity ?? 'session-unknown', + previousMessageId: sentPreviousMessageId, + ...(previousDiagnosticsMessage + ? { + previousMessageReceivedAt: + previousDiagnosticsMessage.receivedAt, + } + : {}), + isSubagent: subagentRequest, + ttlSent: summarizeCacheTtl(finalBody), + } + } + streaming = finalBody.stream === true trace?.mark('set_oauth_headers_body_parse', { route, ms: roundMs(nowMs() - headerBodyParseStart), @@ -3562,6 +3876,8 @@ const anthropicAuthPlugin = async ( }) } + const cacheDiagnosticsBetas = + await getCacheDiagnosticsBetas(requestHeaders) const rewritten = rewriteUrl(input) if (fableRequest && typeof body === 'string') { fableRequest.warmTarget = { @@ -3587,6 +3903,7 @@ const anthropicAuthPlugin = async ( storage, cacheMode: 'hybrid', oauthAccountId, + isSubagent: subagentRequest, }) trace?.mark('cachekeep_track', { session: relayAffinity, @@ -3609,7 +3926,7 @@ const anthropicAuthPlugin = async ( ...(isInsecure() && { tls: { rejectUnauthorized: false } }), }) if (typeof body === 'string') { - await dumpDirectRequest({ + directDump = await dumpDirectRequest({ affinity: relayAffinity, route, status: response.status, @@ -3657,6 +3974,9 @@ const anthropicAuthPlugin = async ( optimisticResponse: relayConfig?.transport === 'websocket', onResponseHeaders: (headers) => harvestQuotaHeaders(headers, served), + onDumpCreated: (handle) => { + relayDump = handle + }, }) trace?.mark('send_headers_received', { route, @@ -3667,6 +3987,18 @@ const anthropicAuthPlugin = async ( }) if (usedDirectFetch) harvestQuotaHeaders(response.headers, served) + attachCacheDiagnosticsResponse(response, { + source: 'turn', + accountId: oauthAccountId, + synthetic: false, + ...cacheDiagnosticsBetas, + requestedModel: parseRequestModel(body), + request: cacheDiagnosticsRequest, + trackSessionId: relayAffinity ?? undefined, + dump: usedDirectFetch ? directDump : relayDump, + status: response.status, + streaming, + }) return response } @@ -3928,39 +4260,6 @@ const anthropicAuthPlugin = async ( } } - async function withStickyRetryAfter( - response: Response, - sessionId: string, - retryAfterSeconds: number, - streamingRateLimit = false, - ) { - const headers = new Headers(response.headers) - headers.set( - 'retry-after', - String(stickyRetryAfterWithJitter(sessionId, retryAfterSeconds)), - ) - if (streamingRateLimit) { - await response.body?.cancel().catch(() => {}) - headers.set('content-type', 'application/json') - return new Response( - JSON.stringify({ - type: 'error', - error: { - type: 'rate_limit_error', - message: - 'Sticky OAuth account five-hour quota resets shortly; retaining session affinity.', - }, - }), - { status: 429, headers }, - ) - } - return new Response(response.body, { - status: response.status, - statusText: response.statusText, - headers, - }) - } - async function tryUsableFallbackAccounts( input: string | URL | Request, init: RequestInit | undefined, @@ -4201,9 +4500,23 @@ const anthropicAuthPlugin = async ( ? initialBody.length : undefined, }) - const wrapResponse = (response: Response) => - createStrippedStream(response, { + const wrapResponse = (response: Response) => { + const diagnosticsContext = + cacheDiagnosticsResponses.get(response) + return createStrippedStream(response, { perf: (stage, data) => trace.mark(stage, data), + ...(diagnosticsContext + ? diagnosticsContext.streaming + ? { + onMessageStart: (message) => + observeCacheDiagnosticsResponse(response, message), + } + : { + onMessageResponse: (message) => + observeCacheDiagnosticsResponse(response, message), + responseMode: 'json' as const, + } + : {}), contentFilterModel: fablePlan?.requestedModel, ...(!fablePlan?.downgraded && fablePlan ? { @@ -4317,6 +4630,7 @@ const anthropicAuthPlugin = async ( } : {}), }) + } const authStart = nowMs() const auth = await getAuth() trace.mark('get_auth', { @@ -4553,6 +4867,8 @@ const anthropicAuthPlugin = async ( ), sessionId, proactiveQuotaDecision.retryAfterSeconds, + false, + cacheDiagnosticsResponses, ), false, ) @@ -4647,6 +4963,7 @@ const anthropicAuthPlugin = async ( sessionId, decision.retryAfterSeconds, inspected.streamingRateLimit, + cacheDiagnosticsResponses, ), ) } diff --git a/packages/opencode/src/server-fallback.ts b/packages/opencode/src/server-fallback.ts index 6e52130a..9d3ce0e4 100644 --- a/packages/opencode/src/server-fallback.ts +++ b/packages/opencode/src/server-fallback.ts @@ -216,10 +216,13 @@ function sameRequestedModel(requestedModel: string, servedModel: string) { export function createServerSideFallbackStreamRewriter(options: { requestedModel: string + maxPendingBytes?: number onOutcome?: (outcome: ServerSideFallbackOutcome) => void onRefusalAfterToolUse?: () => boolean | undefined }) { let pending = '' + let pendingBytes = 0 + let passthrough = false let servedModel: string | undefined let handoff: ServerSideFallbackMarker | undefined let fallbackIterationModel: string | undefined @@ -323,21 +326,41 @@ export function createServerSideFallbackStreamRewriter(options: { boundary.index + boundary.length, ) pending = pending.slice(boundary.index + boundary.length) + pendingBytes = new TextEncoder().encode(pending).byteLength output += rewriteEvent(rawEvent, delimiter) } return output } return { + pendingLength() { + return pendingBytes + }, push(text: string) { + if (passthrough) return text + const textBytes = new TextEncoder().encode(text).byteLength pending += text - return drain() + pendingBytes += textBytes + const output = drain() + if ( + options.maxPendingBytes !== undefined && + pendingBytes > options.maxPendingBytes + ) { + const tail = pending + pending = '' + pendingBytes = 0 + passthrough = true + return output + tail + } + return output }, flush() { + if (passthrough) return '' const output = drain() if (!pending) return output const tail = rewriteEvent(pending, '') pending = '' + pendingBytes = 0 return output + tail }, } diff --git a/packages/opencode/src/tests/cache-diagnostics.test.ts b/packages/opencode/src/tests/cache-diagnostics.test.ts new file mode 100644 index 00000000..673251a8 --- /dev/null +++ b/packages/opencode/src/tests/cache-diagnostics.test.ts @@ -0,0 +1,386 @@ +import { describe, expect, test } from 'bun:test' +import { + applyCacheDiagnosticsOptIn, + buildCacheDiagnosticsRecord, + CACHE_DIAGNOSTICS_LOG_PREFIX, + CACHE_DIAGNOSTICS_SOURCE_SYNTHETIC, + CacheDiagnosticsBetaTracker, + CacheDiagnosticsTracker, + formatCacheDiagnosticsLogLine, + summarizeCacheTtl, + withStickyRetryAfter, +} from '../cache-diagnostics' + +const usage = { + input_tokens: 10, + cache_read_input_tokens: 20, + cache_creation_input_tokens: 30, + cache_creation: { + ephemeral_5m_input_tokens: 40, + ephemeral_1h_input_tokens: 50, + }, +} + +const message = (diagnostics?: unknown, model = 'claude-opus-4-7') => ({ + id: 'provider-response-id-with-no-msg-prefix', + model, + usage, + ...(diagnostics === undefined ? {} : { diagnostics }), +}) + +const request = { + sessionId: 'ses-a', + previousMessageId: null, + isSubagent: false, + ttlSent: '1h' as const, +} + +const input = (overrides: Record = {}) => ({ + request, + source: 'turn', + accountId: 'oauth-account-a', + synthetic: false, + betasHash: '0123456789abcdef', + message: message({ cache_miss_reason: { type: 'unavailable' } }), + receivedAt: 100, + ...overrides, +}) + +describe('CacheDiagnosticsTracker', () => { + test('returns null before capture and preserves opaque provider ids', () => { + const tracker = new CacheDiagnosticsTracker() + expect(tracker.previousFor('ses-a')).toBeNull() + tracker.capture('ses-a', 'provider-response-id-with-no-msg-prefix', 100) + expect(tracker.previousFor('ses-a')).toEqual({ + messageId: 'provider-response-id-with-no-msg-prefix', + receivedAt: 100, + }) + }) + + test('evicts only the oldest unique session at the bounded limit', () => { + const tracker = new CacheDiagnosticsTracker() + for (let index = 0; index < 1_000; index += 1) + tracker.capture(`ses-${index}`, `provider-${index}`, index) + tracker.capture('ses-0', 'provider-0-updated', 2_000) + tracker.capture('ses-1000', 'provider-1000', 2_001) + expect(tracker.previousFor('ses-0')).toBeNull() + expect(tracker.previousFor('ses-1')?.messageId).toBe('provider-1') + }) + + test('copies response context across a sticky retry rewrap', async () => { + const contexts = new WeakMap() + const source = new Response('{}') + contexts.set(source, { sessionId: 'ses-retry' }) + + const destination = await withStickyRetryAfter( + source, + 'ses-retry', + 60, + false, + contexts, + ) + + expect(contexts.get(destination)).toEqual({ sessionId: 'ses-retry' }) + }) +}) + +describe('cache diagnostics v2 contract', () => { + test('writes required v2 metadata and omits a matching requested model', () => { + const result = buildCacheDiagnosticsRecord( + input({ requestedModel: 'claude-opus-4-7' }), + ) + + expect(result.record).toMatchObject({ + v: 2, + source: 'turn', + account_id: 'oauth-account-a', + synthetic: false, + betas_hash: '0123456789abcdef', + cache_read: 20, + cache_creation: 30, + input_tokens: 10, + }) + expect(result.record).not.toHaveProperty('requested_model') + }) + + test('records requested_model only when the served response model differs', () => { + const result = buildCacheDiagnosticsRecord( + input({ requestedModel: 'claude-sonnet-4-7' }), + ) + + expect(result.record?.requested_model).toBe('claude-sonnet-4-7') + }) + + test.each([ + ['turn', false, 'account-turn'], + ['prewarm_cachekeep', true, 'account-prewarm'], + ] as const)('records account_id for %s observations', (source, synthetic, accountId) => { + const result = buildCacheDiagnosticsRecord( + input({ source, synthetic, accountId }), + ) + + expect(result.record).toMatchObject({ + source, + synthetic, + account_id: accountId, + }) + }) + + test('warns on a known source synthetic mismatch while preserving synthetic', () => { + const warnings: string[] = [] + const result = buildCacheDiagnosticsRecord( + input({ + synthetic: true, + onWarning: (message: string) => warnings.push(message), + }), + ) + + expect(result.record?.synthetic).toBe(true) + expect(warnings).toEqual([ + 'cache diagnostics source/synthetic mismatch: source=turn expected=false actual=true', + ]) + }) + + test('does not warn for an unknown source', () => { + const warnings: string[] = [] + const result = buildCacheDiagnosticsRecord( + input({ + source: 'future_machine', + synthetic: true, + onWarning: (message: string) => warnings.push(message), + }), + ) + + expect(result.record?.source).toBe('future_machine') + expect(warnings).toEqual([]) + }) + + test('exports the known source synthetic mapping', () => { + expect(CACHE_DIAGNOSTICS_SOURCE_SYNTHETIC).toEqual({ + turn: false, + prewarm_cachekeep: true, + }) + }) + + test('emits beta side-channel once per hash and retains differing sets', () => { + const tracker = new CacheDiagnosticsBetaTracker() + + expect(tracker.capture('0123456789abcdef', ['beta-b', 'beta-a'])).toBe( + 'MC-CACHE-DIAG-BETAS {"hash":"0123456789abcdef","betas":["beta-a","beta-b"]}', + ) + expect(tracker.capture('0123456789abcdef', ['beta-a', 'beta-b'])).toBeNull() + expect(tracker.capture('fedcba9876543210', ['beta-c'])).toBe( + 'MC-CACHE-DIAG-BETAS {"hash":"fedcba9876543210","betas":["beta-c"]}', + ) + }) + + test('formats a v2 record with the load-bearing record prefix', () => { + const result = buildCacheDiagnosticsRecord(input()) + expect(result.record).toBeDefined() + const line = formatCacheDiagnosticsLogLine(result.record!) + expect(line.startsWith(CACHE_DIAGNOSTICS_LOG_PREFIX)).toBe(true) + expect( + JSON.parse(line.slice(CACHE_DIAGNOSTICS_LOG_PREFIX.length)), + ).toMatchObject({ + v: 2, + session_id: 'ses-a', + message_id: 'provider-response-id-with-no-msg-prefix', + }) + }) + + test.each([ + [{}, 'absent'], + [{ diagnostics: null }, 'server_null'], + [{ diagnostics: { cache_miss_reason: null } }, 'pending'], + [ + { diagnostics: { cache_miss_reason: { type: 'unavailable' } } }, + 'populated', + ], + ] as const)('classifies diagnostics %j as %s', (extra, state) => { + expect( + buildCacheDiagnosticsRecord( + input({ message: { ...message(undefined), ...extra } }), + ).record?.diag_state, + ).toBe(state) + }) + + test('rejects invalid required v2 metadata', () => { + expect(buildCacheDiagnosticsRecord(input({ accountId: '' }))).toEqual({}) + expect(buildCacheDiagnosticsRecord(input({ betasHash: '' }))).toEqual({}) + expect(buildCacheDiagnosticsRecord(input({ source: '' }))).toEqual({}) + }) + + test('copies usage values by property and keeps unavailable ordinary', () => { + const result = buildCacheDiagnosticsRecord(input()) + + expect(result.record).toMatchObject({ + cache_read: 20, + cache_creation: 30, + input_tokens: 10, + ephemeral_5m_tokens: 40, + ephemeral_1h_tokens: 50, + miss_reason: 'unavailable', + }) + }) + + test('parses the captured populated API response', () => { + const result = buildCacheDiagnosticsRecord( + input({ + message: { + id: 'msg_011SampleAnthropicId0000', + model: 'claude-opus-5', + usage: { + input_tokens: 14, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 24837, + cache_creation: { + ephemeral_5m_input_tokens: 0, + ephemeral_1h_input_tokens: 24837, + }, + }, + diagnostics: { + cache_miss_reason: { + type: 'system_changed', + cache_missed_input_tokens: 23963, + }, + }, + }, + }), + ) + + expect(result.record).toMatchObject({ + diag_state: 'populated', + miss_reason: 'system_changed', + cache_missed_input_tokens: 23963, + }) + }) + + test('rejects malformed diagnostics without fabricating absent state', () => { + expect( + buildCacheDiagnosticsRecord( + input({ message: message({ cache_miss_reason: 42 }) }), + ), + ).toEqual({}) + }) + + test.each([ + { cache_miss_reason: 'system_changed' }, + { cache_miss_reason: {} }, + { cache_miss_reason: { type: 42 } }, + ])('rejects non-wire populated reason %j', (diagnostics) => { + expect( + buildCacheDiagnosticsRecord(input({ message: message(diagnostics) })), + ).toEqual({}) + }) + + test.each([ + { diagnostics: 'pending' }, + { diagnostics: [] }, + { diagnostics: 42 }, + ])('rejects malformed diagnostics shape %j', (extra) => { + expect( + buildCacheDiagnosticsRecord( + input({ message: { ...message(), ...extra } }), + ), + ).toEqual({}) + }) + + test('rejects non-finite and fractional receipt timestamps', () => { + expect(buildCacheDiagnosticsRecord(input({ receivedAt: 100.5 }))).toEqual( + {}, + ) + expect( + buildCacheDiagnosticsRecord( + input({ receivedAt: Number.POSITIVE_INFINITY }), + ), + ).toEqual({}) + }) + + test('applies opt-in without replacing existing body fields', () => { + const body: Record = { model: 'x' } + applyCacheDiagnosticsOptIn(body, null) + expect(body).toEqual({ + model: 'x', + diagnostics: { previous_message_id: null }, + }) + }) + + test.each([ + [{ cache_control: { type: 'ephemeral', ttl: '1h' } }, '1h'], + [{ cache_control: { type: 'ephemeral' } }, '5m'], + [{}, null], + ] as const)('summarizes TTL as %s', (body, expected) => { + expect(summarizeCacheTtl(body)).toBe(expected) + }) + + test.each([ + [ + { + system: [{ cache_control: { type: 'ephemeral' } }], + messages: [ + { content: [{ cache_control: { type: 'ephemeral', ttl: '1h' } }] }, + ], + }, + '1h', + ], + [ + { + system: [{ cache_control: { type: 'ephemeral', ttl: '1h' } }], + messages: [{ content: [{ cache_control: { type: 'ephemeral' } }] }], + }, + '5m', + ], + ] as const)('uses the last cache breakpoint as TTL %s', (body, expected) => { + expect(summarizeCacheTtl(body)).toBe(expected) + }) + + test('emits a short-gap canary only for previous_message_not_found', () => { + const result = buildCacheDiagnosticsRecord( + input({ + request: { + ...request, + previousMessageId: 'provider-previous', + previousMessageReceivedAt: 1_000, + }, + message: message({ + cache_miss_reason: { type: 'previous_message_not_found' }, + }), + receivedAt: 1_000 + 5 * 60_000 - 1, + }), + ) + + expect(result.canary).toEqual({ + messageId: 'provider-response-id-with-no-msg-prefix', + previousMessageId: 'provider-previous', + }) + }) + + test.each([ + { reason: 'unavailable', previousMessageId: 'provider-previous', age: 1 }, + { + reason: 'previous_message_not_found', + previousMessageId: 'provider-previous', + age: 5 * 60_000, + }, + { + reason: 'previous_message_not_found', + previousMessageId: null, + age: 1, + }, + ])('does not emit a canary for %j', ({ reason, previousMessageId, age }) => { + const previousMessageReceivedAt = 1_000 + const result = buildCacheDiagnosticsRecord( + input({ + request: { + ...request, + previousMessageId, + previousMessageReceivedAt, + }, + message: message({ cache_miss_reason: { type: reason } }), + receivedAt: previousMessageReceivedAt + age, + }), + ) + + expect(result.canary).toBeUndefined() + }) +}) diff --git a/packages/opencode/src/tests/cachekeep.test.ts b/packages/opencode/src/tests/cachekeep.test.ts index a350dad8..636ce782 100644 --- a/packages/opencode/src/tests/cachekeep.test.ts +++ b/packages/opencode/src/tests/cachekeep.test.ts @@ -1,4 +1,7 @@ import { describe, expect, mock, test } from 'bun:test' +import { mkdtemp, readdir, readFile, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' import { type AccountStorage, buildCacheKeepPrewarmBody, @@ -7,6 +10,8 @@ import { CacheKeepManager, executeCacheKeepCommand, parseCacheKeepCommandAction, + resetDumpState, + setDumpEnabled, } from '@cortexkit/anthropic-auth-core' const hybridStorage = (): AccountStorage => ({ @@ -128,6 +133,138 @@ describe('cachekeep prewarm body', () => { }) describe('CacheKeepManager', () => { + test('writes cachekeep-tagged prewarm request and response artifacts', async () => { + const dumpDir = await mkdtemp(join(tmpdir(), 'cachekeep-dump-test-')) + const originalDumpDir = process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR + process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = dumpDir + setDumpEnabled(true) + try { + const manager = new CacheKeepManager({ + loadStorage: () => Promise.resolve(hybridStorage()), + fetchImpl: mock( + async () => new Response('malformed', { status: 429 }), + ) as unknown as typeof fetch, + }) + await manager.prewarmNow({ + sessionId: 'ses_dump', + url: 'https://api.anthropic.com/v1/messages', + headers: new Headers(), + bodyText: JSON.stringify({ + system: [ + { + type: 'text', + text: 'stable', + cache_control: { type: 'ephemeral' }, + }, + ], + messages: [{ role: 'user', content: 'hello' }], + }), + }) + const files = await readdir(dumpDir) + expect(files.some((file) => file.includes('-prewarm-cachekeep-'))).toBe( + true, + ) + const metadataPath = files.find((file) => file.endsWith('.meta.json')) + expect(metadataPath).toBeDefined() + const metadata = JSON.parse( + await readFile(join(dumpDir, metadataPath!), 'utf8'), + ) + expect(metadata.tag).toBe('cachekeep') + const responsePath = files.find((file) => file.endsWith('.response.json')) + expect(responsePath).toBeDefined() + expect( + JSON.parse(await readFile(join(dumpDir, responsePath!), 'utf8')), + ).toEqual({ status: 429 }) + } finally { + resetDumpState() + if (originalDumpDir === undefined) + delete process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR + else process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = originalDumpDir + await rm(dumpDir, { recursive: true, force: true }) + } + }) + + test('prepares the tracked body and observes the exact sent body', async () => { + const sent: string[] = [] + const observed: unknown[] = [] + const body = JSON.stringify({ + model: 'claude-opus-4-7', + max_tokens: 100, + stream: true, + system: [ + { type: 'text', text: 'stable', cache_control: { type: 'ephemeral' } }, + ], + messages: [{ role: 'user', content: 'hello' }], + }) + const manager = new CacheKeepManager({ + loadStorage: () => Promise.resolve(hybridStorage()), + fetchImpl: mock(async (_input, init) => { + sent.push(String(init?.body)) + return new Response( + JSON.stringify({ + usage: { + input_tokens: 2, + cache_creation_input_tokens: 1, + cache_read_input_tokens: 3, + cache_creation: { + ephemeral_5m_input_tokens: 4, + ephemeral_1h_input_tokens: 5, + }, + }, + }), + { status: 200 }, + ) + }) as unknown as typeof fetch, + prepareBody: (value) => value.replace('hello', 'prepared'), + onResponse: (input) => { + observed.push(input) + }, + }) + const result = await manager.prewarmNow({ + sessionId: 'ses_prepare', + url: 'https://api.anthropic.com/v1/messages', + headers: new Headers(), + bodyText: body, + }) + expect(result.ok).toBe(true) + expect(sent[0]).toContain('prepared') + expect(observed).toHaveLength(1) + expect(sent[0]).toBeDefined() + expect((observed[0] as { bodyText: string }).bodyText).toBe(sent[0]!) + expect( + (observed[0] as { data: { usage: { cache_creation: unknown } } }).data + .usage.cache_creation, + ).toEqual({ ephemeral_5m_input_tokens: 4, ephemeral_1h_input_tokens: 5 }) + }) + + test('observer errors do not break malformed responses or scheduling', async () => { + const manager = new CacheKeepManager({ + loadStorage: () => Promise.resolve(hybridStorage()), + fetchImpl: mock( + async () => new Response('not-json', { status: 429 }), + ) as unknown as typeof fetch, + onResponse: () => { + throw new Error('observer failure') + }, + }) + const result = await manager.prewarmNow({ + sessionId: 'ses_observer', + url: 'https://api.anthropic.com/v1/messages', + headers: new Headers(), + bodyText: JSON.stringify({ + system: [ + { + type: 'text', + text: 'stable', + cache_control: { type: 'ephemeral' }, + }, + ], + messages: [{ role: 'user', content: 'hello' }], + }), + }) + expect(result).toMatchObject({ ok: false, status: 429 }) + }) + test('tracks hybrid sessions and prewarms five minutes before expiry', async () => { let now = new Date('2026-05-18T10:00:00').getTime() const calls: Array<{ url: string; body: string }> = [] diff --git a/packages/opencode/src/tests/index.test.ts b/packages/opencode/src/tests/index.test.ts index 89e1c61e..3bd6603d 100644 --- a/packages/opencode/src/tests/index.test.ts +++ b/packages/opencode/src/tests/index.test.ts @@ -9403,6 +9403,647 @@ describe('auth.loader', () => { }) }) +describe('cache diagnostics', () => { + const originalFetch = globalThis.fetch + const originalDateNow = Date.now + + const message = ( + id: string, + diagnostics: unknown = { cache_miss_reason: null }, + ) => ({ + id, + model: 'claude-opus-4-8', + usage: { + input_tokens: 101, + cache_read_input_tokens: 75, + cache_creation_input_tokens: 26, + cache_creation: { + ephemeral_5m_input_tokens: 20, + ephemeral_1h_input_tokens: 6, + }, + }, + diagnostics, + }) + + const sseResponse = (data: Record, status = 200) => + new Response( + `event: message_start\ndata: ${JSON.stringify({ type: 'message_start', message: data })}\n\n` + + 'event: message_stop\ndata: {"type":"message_stop"}\n\n', + { status }, + ) + + const oauthLoader = () => + Promise.resolve({ + type: 'oauth' as const, + access: 'main-access', + refresh: 'main-refresh', + expires: Date.now() + 100_000, + }) + + beforeEach(async () => { + globalThis.fetch = originalFetch + Date.now = originalDateNow + pluginTimerOverrides = { + setInterval: mock( + () => ({ unref() {} }) as unknown as ReturnType, + ) as unknown as typeof setInterval, + clearInterval: mock(() => {}) as unknown as typeof clearInterval, + } + resetCache1hState() + resetDumpState() + setLogLevel('info') + process.env.OPENCODE_ANTHROPIC_AUTH_DISABLE_PROFILE_HYDRATION = '1' + await useTempAccountFile( + createFallbackStorage({ accounts: [], quota: { enabled: false } }), + ) + }) + + afterEach(async () => { + __setLogTestSink(null) + globalThis.fetch = originalFetch + Date.now = originalDateNow + pluginTimerOverrides = {} + resetDumpState() + delete process.env.OPENCODE_ANTHROPIC_AUTH_DISABLE_PROFILE_HYDRATION + await drainSidebarWrites() + restoreProcessTestFiles() + if (tempConfigDir) { + await rm(tempConfigDir, { recursive: true, force: true }) + tempConfigDir = undefined + } + }) + + test('cache diagnostics binds the provider predecessor at request time', async () => { + const sentBodies: Record[] = [] + const records: LogTestRecord[] = [] + const delayedReleases = new Map void>() + const delayedResponse = (providerId: string) => { + const body = new ReadableStream({ + start(controller) { + return new Promise((resolve) => { + delayedReleases.set(providerId, () => { + const payload = message(providerId) + controller.enqueue( + new TextEncoder().encode( + `event: message_start\ndata: ${JSON.stringify({ type: 'message_start', message: payload })}\n\n` + + 'event: message_stop\ndata: {"type":"message_stop"}\n\n', + ), + ) + controller.close() + resolve() + }) + }) + }, + }) + return new Response(body, { status: 200 }) + } + globalThis.fetch = mock((_input: any, init: RequestInit) => { + sentBodies.push(JSON.parse(String(init.body))) + const responsePlan = [ + ['ses-diag-A', 'provider-A'], + ['ses-diag-B', 'provider-B'], + ['ses-diag-A', 'provider-A-after'], + ['ses-diag-B', 'provider-B-after'], + ] as const + const response = responsePlan[sentBodies.length - 1] + if (!response) throw new Error('unexpected request') + return Promise.resolve( + sentBodies.length <= 2 + ? sseResponse(message(response[1])) + : delayedResponse(response[1]), + ) + }) as unknown as typeof fetch + __setLogTestSink((record) => records.push(record)) + + const plugin = await getPlugin( + createMockClient([ + { info: { id: 'msg_opencode_decoy', role: 'assistant' } }, + ]), + ) + const result = await plugin.auth.loader(oauthLoader, { models: {} }) + const request = (sessionId: string) => ({ + method: 'POST', + headers: { 'x-session-affinity': sessionId }, + body: JSON.stringify({ + model: 'claude-opus-4-8', + stream: true, + messages: [{ role: 'user', content: 'hello' }], + }), + }) + + await (await result.fetch(MESSAGES_URL, request('ses-diag-A'))).text() + await (await result.fetch(MESSAGES_URL, request('ses-diag-B'))).text() + const delayedA = await result.fetch(MESSAGES_URL, request('ses-diag-A')) + const delayedB = await result.fetch(MESSAGES_URL, request('ses-diag-B')) + delayedReleases.get('provider-A-after')?.() + await delayedA.text() + delayedReleases.get('provider-B-after')?.() + await delayedB.text() + + expect(sentBodies.map((body) => body.diagnostics)).toEqual([ + { previous_message_id: null }, + { previous_message_id: null }, + { previous_message_id: 'provider-A' }, + { previous_message_id: 'provider-B' }, + ]) + const lines = records + .filter( + (record) => + record.channel === 'cache-diagnostics' && + record.message.startsWith('MC-CACHE-DIAG '), + ) + .map((record) => JSON.parse(record.message.replace('MC-CACHE-DIAG ', ''))) + expect(lines).toHaveLength(4) + expect( + lines.find((line) => line.message_id === 'provider-A-after'), + ).toMatchObject({ + previous_message_id: 'provider-A', + ttl_sent: null, + cache_read: 75, + cache_creation: 26, + input_tokens: 101, + ephemeral_5m_tokens: 20, + ephemeral_1h_tokens: 6, + session_id: 'ses-diag-A', + is_subagent: false, + v: 2, + source: 'turn', + synthetic: false, + account_id: 'main', + betas_hash: expect.stringMatching(/^[0-9a-f]{16}$/), + }) + expect( + lines.find((line) => line.message_id === 'provider-B-after'), + ).toMatchObject({ + previous_message_id: 'provider-B', + session_id: 'ses-diag-B', + }) + }) + + test('cache diagnostics isolates missing affinity and strips the parent header', async () => { + const sentBodies: Record[] = [] + const sentHeaders: Headers[] = [] + const records: LogTestRecord[] = [] + let requestNumber = 0 + globalThis.fetch = mock((_input: any, init: RequestInit) => { + sentBodies.push(JSON.parse(String(init.body))) + sentHeaders.push(new Headers(init.headers)) + requestNumber++ + return Promise.resolve(sseResponse(message(`provider-${requestNumber}`))) + }) as unknown as typeof fetch + __setLogTestSink((record) => records.push(record)) + + const plugin = await getPlugin() + const result = await plugin.auth.loader(oauthLoader, { models: {} }) + const body = JSON.stringify({ + model: 'claude-opus-4-8', + stream: true, + messages: [{ role: 'user', content: 'hello' }], + }) + await ( + await result.fetch(MESSAGES_URL, { + method: 'POST', + headers: { 'x-parent-session-id': 'parent-1' }, + body, + }) + ).text() + await (await result.fetch(MESSAGES_URL, { method: 'POST', body })).text() + + expect(sentBodies.map((entry) => entry.diagnostics)).toEqual([ + { previous_message_id: null }, + { previous_message_id: null }, + ]) + expect(sentHeaders[0]?.has('x-parent-session-id')).toBe(false) + const lines = records + .filter( + (record) => + record.channel === 'cache-diagnostics' && + record.message.startsWith('MC-CACHE-DIAG '), + ) + .map((record) => JSON.parse(record.message.replace('MC-CACHE-DIAG ', ''))) + expect(lines).toHaveLength(2) + expect(lines[0]).toMatchObject({ + session_id: 'session-unknown', + is_subagent: true, + }) + expect(lines[1]).toMatchObject({ + session_id: 'session-unknown', + is_subagent: false, + }) + }) + + test('cache diagnostics emits the opt-in beta for normal and structured OAuth requests', async () => { + const sentBodies: Record[] = [] + const betaHeaders: string[] = [] + const records: LogTestRecord[] = [] + globalThis.fetch = mock((_input: any, init: RequestInit) => { + sentBodies.push(JSON.parse(String(init.body))) + betaHeaders.push(new Headers(init.headers).get('anthropic-beta') ?? '') + return Promise.resolve( + sseResponse(message(`provider-${sentBodies.length}`)), + ) + }) as unknown as typeof fetch + __setLogTestSink((record) => records.push(record)) + + const plugin = await getPlugin() + const result = await plugin.auth.loader(oauthLoader, { models: {} }) + for (const output_config of [ + undefined, + { format: { type: 'json_schema' } }, + ]) { + await ( + await result.fetch(MESSAGES_URL, { + method: 'POST', + body: JSON.stringify({ + model: 'claude-opus-4-8', + stream: true, + messages: [{ role: 'user', content: 'hello' }], + ...(output_config ? { output_config } : {}), + }), + }) + ).text() + } + + expect(sentBodies.map((body) => body.diagnostics)).toEqual([ + { previous_message_id: null }, + { previous_message_id: null }, + ]) + expect( + betaHeaders.every((beta) => beta.includes('cache-diagnosis-2026-04-07')), + ).toBe(true) + const betaLines = records + .filter( + (record) => + record.channel === 'cache-diagnostics' && + record.message.startsWith('MC-CACHE-DIAG-BETAS '), + ) + .map((record) => + JSON.parse(record.message.replace('MC-CACHE-DIAG-BETAS ', '')), + ) + expect(betaLines).toHaveLength(2) + expect(new Set(betaLines.map((line) => line.hash)).size).toBe(2) + }) + + test('cache diagnostics observes non-streaming envelopes and carries their provider id forward', async () => { + const sentBodies: Record[] = [] + const records: LogTestRecord[] = [] + globalThis.fetch = mock((_input: any, init: RequestInit) => { + sentBodies.push(JSON.parse(String(init.body))) + const id = sentBodies.length === 1 ? 'provider-json-A' : 'provider-json-B' + return Promise.resolve( + new Response(JSON.stringify(message(id)), { status: 200 }), + ) + }) as unknown as typeof fetch + __setLogTestSink((record) => records.push(record)) + + const plugin = await getPlugin() + const result = await plugin.auth.loader(oauthLoader, { models: {} }) + const request = { + method: 'POST', + headers: { 'x-session-affinity': 'ses-json' }, + body: JSON.stringify({ + model: 'claude-opus-4-8', + stream: false, + messages: [{ role: 'user', content: 'hello' }], + }), + } + await (await result.fetch(MESSAGES_URL, request)).text() + await (await result.fetch(MESSAGES_URL, request)).text() + + expect(sentBodies[1]?.diagnostics).toEqual({ + previous_message_id: 'provider-json-A', + }) + expect( + records.filter( + (record) => + record.channel === 'cache-diagnostics' && + record.message.startsWith('MC-CACHE-DIAG '), + ), + ).toHaveLength(2) + }) + + test('cache diagnostics records cachekeep prewarms and carries their provider id forward', async () => { + let now = 1_000 + const intervals: Array<{ callback: () => unknown; ms: number }> = [] + const sentBodies: Record[] = [] + const records: LogTestRecord[] = [] + let prewarmStartedFlag = false + let resolvePrewarmStarted: (() => void) | undefined + const prewarmStarted = new Promise((resolve) => { + resolvePrewarmStarted = () => { + prewarmStartedFlag = true + resolve() + } + }) + Date.now = mock(() => now) as unknown as typeof Date.now + pluginTimerOverrides = { + setInterval: mock((callback: () => unknown, ms: number) => { + intervals.push({ callback, ms }) + return { unref() {} } as unknown as ReturnType + }) as unknown as typeof setInterval, + clearInterval: mock(() => {}) as unknown as typeof clearInterval, + } + await useTempAccountFile( + createFallbackStorage({ + accounts: [], + quota: { enabled: false }, + claudeCache: { enabled: true, mode: 'hybrid' }, + cacheKeep: { enabled: true, always: true, subagents: true }, + }), + ) + let normalRequests = 0 + globalThis.fetch = mock((_input: any, init: RequestInit) => { + const body = JSON.parse(String(init.body)) as Record + sentBodies.push(body) + if (body.max_tokens === 0) { + resolvePrewarmStarted?.() + return Promise.resolve( + new Response(JSON.stringify(message('provider-warm-B')), { + status: 200, + }), + ) + } + normalRequests++ + return Promise.resolve( + sseResponse( + message(normalRequests === 1 ? 'provider-real-A' : 'provider-real-C'), + ), + ) + }) as unknown as typeof fetch + __setLogTestSink((record) => records.push(record)) + + const plugin = await getPlugin() + const result = await plugin.auth.loader(oauthLoader, { models: {} }) + const request = { + method: 'POST', + headers: { 'x-session-affinity': 'ses-cachekeep' }, + body: JSON.stringify({ + model: 'claude-opus-4-8', + stream: true, + messages: [{ role: 'user', content: 'hello' }], + }), + } + await (await result.fetch(MESSAGES_URL, request)).text() + now += 55 * 60_000 + const cacheKeepTick = intervals.at(-1) + if (!cacheKeepTick) throw new Error('missing cachekeep interval') + cacheKeepTick.callback() + await Bun.sleep(20) + await prewarmStarted + expect(prewarmStartedFlag).toBe(true) + for ( + let attempt = 0; + attempt < 50 && + !records.some( + (record) => + record.channel === 'cache-diagnostics' && + record.message.includes('provider-warm-B'), + ); + attempt++ + ) { + await Bun.sleep(10) + } + await (await result.fetch(MESSAGES_URL, request)).text() + + const prewarmBody = sentBodies.find((body) => body.max_tokens === 0) + expect(prewarmBody?.diagnostics).toEqual({ + previous_message_id: 'provider-real-A', + }) + expect(sentBodies.at(-1)?.diagnostics).toEqual({ + previous_message_id: 'provider-warm-B', + }) + const prewarmRecord = records.find( + (record) => + record.channel === 'cache-diagnostics' && + record.message.includes('provider-warm-B'), + ) + expect(prewarmRecord).toBeDefined() + expect( + JSON.parse(prewarmRecord!.message.replace('MC-CACHE-DIAG ', '')), + ).toMatchObject({ + is_subagent: false, + ttl_sent: '1h', + previous_message_id: 'provider-real-A', + source: 'prewarm_cachekeep', + synthetic: true, + account_id: 'main', + betas_hash: expect.stringMatching(/^[0-9a-f]{16}$/), + }) + }) + + test('cache diagnostics logs a short-gap previous-message canary but not unavailable', async () => { + const records: LogTestRecord[] = [] + let now = 1_000 + Date.now = mock(() => now) as unknown as typeof Date.now + const responses = [ + message('provider-canary-A'), + message('provider-canary-B', { + cache_miss_reason: { type: 'previous_message_not_found' }, + }), + message('provider-canary-C', { + cache_miss_reason: { type: 'unavailable' }, + }), + message('provider-canary-D', { + cache_miss_reason: { type: 'previous_message_not_found' }, + }), + ] + globalThis.fetch = mock(() => { + const response = responses.shift() + if (!response) throw new Error('unexpected request') + return Promise.resolve(sseResponse(response)) + }) as unknown as typeof fetch + __setLogTestSink((record) => records.push(record)) + + const plugin = await getPlugin() + const result = await plugin.auth.loader(oauthLoader, { models: {} }) + const request = { + method: 'POST', + headers: { 'x-session-affinity': 'ses-canary' }, + body: JSON.stringify({ + model: 'claude-opus-4-8', + stream: true, + messages: [{ role: 'user', content: 'hello' }], + }), + } + await (await result.fetch(MESSAGES_URL, request)).text() + now += 299_999 + await (await result.fetch(MESSAGES_URL, request)).text() + now += 300_000 + await (await result.fetch(MESSAGES_URL, request)).text() + now += 300_000 + await (await result.fetch(MESSAGES_URL, request)).text() + + const warnings = records.filter( + (record) => + record.level === 'warn' && record.channel === 'cache-diagnostics', + ) + expect(warnings).toHaveLength(1) + expect(warnings[0]?.payload).toMatchObject({ + message_id: 'provider-canary-B', + previous_message_id: 'provider-canary-A', + }) + }) + + test('cache diagnostics stays disabled on API-key fallback while preserving its response artifact', async () => { + const originalDumpDir = process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR + const dumpDir = await mkdtemp(join(tmpdir(), 'cache-diagnostics-api-dump-')) + process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = dumpDir + try { + await useTempAccountFile( + createFallbackStorage({ + dump: { enabled: true }, + quota: { + enabled: true, + checkIntervalMinutes: 5, + minimumRemaining: { five_hour: 10, seven_day: 20 }, + failClosedOnUnknownQuota: true, + mainQuota: { + five_hour: { usedPercent: 100, remainingPercent: 0 }, + seven_day: { usedPercent: 50, remainingPercent: 50 }, + }, + mainQuotaCheckedAt: Date.now(), + mainQuotaToken: tokenFingerprint('main-access'), + } as AccountStorage['quota'], + accounts: [ + { + id: 'kie-opus', + type: 'api', + apiKey: 'kie-key', + baseURL: 'https://api.kie.ai/claude', + authHeader: 'authorization-bearer', + }, + ], + }), + ) + const records: LogTestRecord[] = [] + let sentBody: Record | undefined + let sentBeta = '' + globalThis.fetch = mock((_input: any, init: RequestInit) => { + sentBody = JSON.parse(String(init.body)) + sentBeta = new Headers(init.headers).get('anthropic-beta') ?? '' + return Promise.resolve(sseResponse(message('provider-api'))) + }) as unknown as typeof fetch + __setLogTestSink((record) => records.push(record)) + + const plugin = await getPlugin() + const result = await plugin.auth.loader(oauthLoader, { models: {} }) + await ( + await result.fetch(MESSAGES_URL, { + method: 'POST', + headers: { 'x-session-affinity': 'ses-api' }, + body: JSON.stringify({ + model: 'claude-opus-4-8', + stream: true, + messages: [{ role: 'user', content: 'hello' }], + }), + }) + ).text() + + expect(sentBody?.diagnostics).toBeUndefined() + expect(sentBeta).not.toContain('cache-diagnosis-2026-04-07') + expect( + records.filter((record) => record.channel === 'cache-diagnostics'), + ).toHaveLength(0) + for (let attempt = 0; attempt < 50; attempt++) { + if ( + (await readdir(dumpDir)).some((file) => + file.endsWith('.response.json'), + ) + ) + break + await Bun.sleep(10) + } + const responseFile = (await readdir(dumpDir)).find((file) => + file.endsWith('.response.json'), + ) + expect(responseFile).toBeString() + expect( + JSON.parse(await readFile(join(dumpDir, responseFile!), 'utf8')), + ).toMatchObject({ + status: 200, + message_id: 'provider-api', + }) + } finally { + if (originalDumpDir === undefined) { + delete process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR + } else { + process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = originalDumpDir + } + await rm(dumpDir, { recursive: true, force: true }) + } + }) + + test('cache diagnostics writes sanitized response artifacts for valid and malformed direct responses', async () => { + const originalDumpDir = process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR + const dumpDir = await mkdtemp( + join(tmpdir(), 'cache-diagnostics-dump-test-'), + ) + process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = dumpDir + try { + await useTempAccountFile( + createFallbackStorage({ + accounts: [], + quota: { enabled: false }, + dump: { enabled: true }, + }), + ) + const responses = [ + sseResponse({ + ...message('provider-dump'), + content: [{ text: 'secret' }], + }), + new Response('not an envelope', { status: 503 }), + ] + globalThis.fetch = mock(() => { + const response = responses.shift() + if (!response) throw new Error('unexpected request') + return Promise.resolve(response) + }) as unknown as typeof fetch + + const plugin = await getPlugin() + const result = await plugin.auth.loader(oauthLoader, { models: {} }) + const request = { + method: 'POST', + headers: { 'x-session-affinity': 'ses-dump' }, + body: JSON.stringify({ + model: 'claude-opus-4-8', + stream: true, + messages: [{ role: 'user', content: 'hello' }], + }), + } + await (await result.fetch(MESSAGES_URL, request)).text() + await (await result.fetch(MESSAGES_URL, request)).text() + + let responseFiles: string[] = [] + for (let attempt = 0; attempt < 50; attempt++) { + responseFiles = (await readdir(dumpDir)).filter((file) => + file.endsWith('.response.json'), + ) + if (responseFiles.length === 2) break + await Bun.sleep(10) + } + expect(responseFiles).toHaveLength(2) + const artifacts = await Promise.all( + responseFiles.map(async (file) => + JSON.parse(await readFile(join(dumpDir, file), 'utf8')), + ), + ) + expect(artifacts).toContainEqual( + expect.objectContaining({ status: 200, message_id: 'provider-dump' }), + ) + expect(artifacts).toContainEqual({ status: 503 }) + expect(JSON.stringify(artifacts)).not.toContain('secret') + } finally { + if (originalDumpDir === undefined) { + delete process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR + } else { + process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = originalDumpDir + } + await rm(dumpDir, { recursive: true, force: true }) + } + }) +}) + describe('killswitch fetch gate', () => { const originalFetch = globalThis.fetch diff --git a/packages/opencode/src/tests/plugin-exports.test.ts b/packages/opencode/src/tests/plugin-exports.test.ts index b7b2bd6b..5569889c 100644 --- a/packages/opencode/src/tests/plugin-exports.test.ts +++ b/packages/opencode/src/tests/plugin-exports.test.ts @@ -1,9 +1,13 @@ import { describe, expect, test } from 'bun:test' describe('plugin module exports', () => { - test('does not expose a state-mutating test hook to the OpenCode loader', async () => { + test('plugin module exports remain limited to loader-compatible factories and helpers', async () => { const pluginModule = await import('../index') - expect('__setBootProfileHydrationForTest' in pluginModule).toBe(false) + expect(Object.keys(pluginModule).sort()).toEqual([ + 'AnthropicAuthPlugin', + 'formatKillswitchBlockMessage', + 'resolveScopedDrivenBlock', + ]) }) }) diff --git a/packages/opencode/src/tests/relay.test.ts b/packages/opencode/src/tests/relay.test.ts index 554e9d97..0ed21908 100644 --- a/packages/opencode/src/tests/relay.test.ts +++ b/packages/opencode/src/tests/relay.test.ts @@ -283,6 +283,46 @@ describe('relay client', () => { } }) + test('reports the created HTTP relay dump handle without replacing the response', async () => { + const originalFetch = globalThis.fetch + const originalDumpDir = process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR + process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = await mkdtemp( + join(tmpdir(), 'anthropic-auth-relay-handle-'), + ) + setDumpEnabled(true) + globalThis.fetch = mock( + async () => new Response('relay', { status: 201 }), + ) as unknown as typeof fetch + const handles: unknown[] = [] + try { + const response = await sendViaRelay({ + config, + input: 'https://api.anthropic.com/v1/messages?beta=true', + init: { method: 'POST' }, + headers: headers('session-relay-handle'), + body: '{}', + fallback: async () => new Response('direct'), + onDumpCreated: (handle) => { + handles.push(handle) + throw new Error('observer failure') + }, + }) + expect(response.status).toBe(201) + expect(handles).toHaveLength(1) + expect((handles[0] as { responsePath: string }).responsePath).toEndWith( + '.response.json', + ) + } finally { + const dumpDir = getDumpDirectory() + resetDumpState() + globalThis.fetch = originalFetch + if (originalDumpDir === undefined) + delete process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR + else process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = originalDumpDir + await rm(dumpDir, { recursive: true, force: true }) + } + }) + test('retries with full sync when relay reports state mismatch', async () => { const calls: unknown[] = [] const originalFetch = globalThis.fetch @@ -1686,6 +1726,7 @@ describe('relay client', () => { const dumpDir = await mkdtemp(join(tmpdir(), 'anthropic-auth-dump-test-')) process.env.OPENCODE_ANTHROPIC_AUTH_DUMP_DIR = dumpDir const sentPayloads: unknown[] = [] + const handles: unknown[] = [] setDumpEnabled(true) class DumpingWebSocket extends EventTarget { @@ -1754,6 +1795,7 @@ describe('relay client', () => { headers: headers('session-relay-ws-dump-exact'), body: JSON.stringify({ messages: ['one'] }), fallback: async () => new Response('direct'), + onDumpCreated: (handle) => handles.push(handle), }) const files = await readdir(getDumpDirectory()) @@ -1774,6 +1816,10 @@ describe('relay client', () => { expect(relay).toMatchObject({ protocol: 2, mode: 'full_sync' }) expect(relay.id).toBeString() expect(meta.relayBytes).toBe(JSON.stringify(sentPayloads[0]).length) + expect(handles).toHaveLength(1) + expect((handles[0] as { responsePath: string }).responsePath).toEndWith( + '.response.json', + ) } finally { resetDumpState() globalThis.WebSocket = originalWebSocket diff --git a/packages/opencode/src/tests/transform.test.ts b/packages/opencode/src/tests/transform.test.ts index 2368f3d3..c8ee9ac7 100644 --- a/packages/opencode/src/tests/transform.test.ts +++ b/packages/opencode/src/tests/transform.test.ts @@ -8,6 +8,7 @@ import { } from '@cortexkit/anthropic-auth-core' import dedent from 'dedent' import { + createServerSideFallbackStreamRewriter, SERVER_FALLBACK_MARKER_TEXT, SERVER_FALLBACK_SIGNATURE_PREFIX, SERVER_SIDE_FALLBACK_BETA, @@ -21,6 +22,7 @@ import { isInsecure, mergeBetaHeaders, mergeHeaders, + NON_STREAMING_DIAGNOSTICS_MAX_BYTES, prefixToolNames, prepareFableCacheWarmSource, prependClaudeCodeIdentity, @@ -483,6 +485,190 @@ describe('isInsecure', () => { }) describe('createStrippedStream', () => { + test('observes a split message_start envelope exactly once', async () => { + const message = { + id: 'msg_provider_1', + usage: { input_tokens: 3 }, + diagnostics: { cache_miss_reason: null }, + } + const payload = sse('message_start', { type: 'message_start', message }) + const seen: unknown[] = [] + const stream = new ReadableStream({ + start(controller) { + const encoder = new TextEncoder() + controller.enqueue(encoder.encode(payload.slice(0, 17))) + controller.enqueue(encoder.encode(payload.slice(17))) + controller.close() + }, + }) + await createStrippedStream(new Response(stream), { + onMessageStart: (value) => seen.push(value), + }).text() + expect(seen).toEqual([message]) + }) + + test('captures message_start before disabling diagnostics on an oversized stream tail', async () => { + const start = sse('message_start', { + type: 'message_start', + message: { id: 'msg_stream_start', usage: { input_tokens: 1 } }, + }) + const body = `${start}${'x'.repeat(NON_STREAMING_DIAGNOSTICS_MAX_BYTES * 2)}` + const seen: unknown[] = [] + const perf: Array> = [] + const response = createStrippedStream( + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(body)) + controller.close() + }, + }), + ), + { + onMessageStart: (message) => seen.push(message), + perf: (_stage, stats) => perf.push(stats ?? {}), + }, + ) + + expect(await response.text()).toBe(body) + expect(seen).toEqual([ + { id: 'msg_stream_start', usage: { input_tokens: 1 } }, + ]) + expect( + Math.max(...perf.map((stats) => Number(stats.ssePendingChars ?? 0))), + ).toBeLessThanOrEqual(NON_STREAMING_DIAGNOSTICS_MAX_BYTES) + expect(perf.at(-1)?.ssePendingOverflowCount).toBe(1) + }) + + test('observes a split non-streaming message response without changing bytes', async () => { + const message = { + id: 'msg_provider_2', + usage: { input_tokens: 3 }, + diagnostics: { cache_miss_reason: { type: 'unavailable' } }, + } + const body = JSON.stringify(message) + const seen: unknown[] = [] + const stream = new ReadableStream({ + start(controller) { + const encoder = new TextEncoder() + controller.enqueue(encoder.encode(body.slice(0, 9))) + controller.enqueue(encoder.encode(body.slice(9))) + controller.close() + }, + }) + const response = createStrippedStream(new Response(stream), { + responseMode: 'json', + onMessageResponse: (value) => seen.push(value), + }) + expect(await response.text()).toBe(body) + expect(seen).toEqual([message]) + }) + + test('passes over-cap non-streaming bytes unchanged without observing diagnostics', async () => { + const body = JSON.stringify({ + id: 'msg_provider_over_cap', + usage: { input_tokens: 3 }, + diagnostics: { cache_miss_reason: { type: 'unavailable' } }, + padding: 'x'.repeat(NON_STREAMING_DIAGNOSTICS_MAX_BYTES * 2), + }) + const seen: unknown[] = [] + const perf: Array> = [] + const response = createStrippedStream(new Response(body), { + responseMode: 'json', + serverSideFallbackModel: 'claude-fable-5', + onComplete: () => {}, + onMessageResponse: (value) => seen.push(value), + perf: (_stage, stats) => perf.push(stats ?? {}), + }) + expect(await response.text()).toBe(body) + expect(seen).toEqual([]) + expect( + Math.max(...perf.map((stats) => Number(stats.ssePendingChars ?? 0))), + ).toBe(0) + expect( + Math.max(...perf.map((stats) => Number(stats.sseErrorPending ?? 0))), + ).toBe(0) + expect( + Math.max(...perf.map((stats) => Number(stats.sseFinishPending ?? 0))), + ).toBeLessThanOrEqual(NON_STREAMING_DIAGNOSTICS_MAX_BYTES) + expect( + Math.max( + ...perf.map((stats) => Number(stats.serverFallbackPending ?? 0)), + ), + ).toBeLessThanOrEqual(NON_STREAMING_DIAGNOSTICS_MAX_BYTES) + }) + + test('bounds server fallback pending state', () => { + const rewriter = createServerSideFallbackStreamRewriter({ + requestedModel: 'claude-fable-5', + maxPendingBytes: NON_STREAMING_DIAGNOSTICS_MAX_BYTES, + }) + rewriter.push('x'.repeat(NON_STREAMING_DIAGNOSTICS_MAX_BYTES * 2)) + expect(rewriter.pendingLength()).toBeLessThanOrEqual( + NON_STREAMING_DIAGNOSTICS_MAX_BYTES, + ) + }) + + test('drains complete fallback frames before passing through an oversized tail', async () => { + const outcomes: ServerSideFallbackOutcome[] = [] + const frames = [ + sse('message_start', { + type: 'message_start', + message: { id: 'msg_overflow', model: 'claude-opus-5' }, + }), + sse('content_block_start', { + type: 'content_block_start', + index: 0, + content_block: { + type: 'fallback', + from: { model: 'claude-fable-5' }, + to: { model: 'claude-opus-5' }, + }, + }), + sse('message_delta', { + type: 'message_delta', + delta: { stop_reason: 'end_turn' }, + usage: { + iterations: [{ type: 'fallback_message', model: 'claude-opus-5' }], + }, + }), + ].join('') + const tail = 'unparseable-tail'.repeat(600_000) + const body = `${frames}${tail}` + const response = createStrippedStream( + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(body)) + controller.close() + }, + }), + ), + { + serverSideFallbackModel: 'claude-fable-5', + onServerSideFallbackOutcome: (outcome) => outcomes.push(outcome), + }, + ) + + const text = await response.text() + expect(text).toContain(SERVER_FALLBACK_SIGNATURE_PREFIX) + expect(text).toContain(tail) + expect(outcomes).toHaveLength(1) + }) + + test('swallows observation callback errors', async () => { + const payload = sse('message_start', { + type: 'message_start', + message: { id: 'msg_provider_3' }, + }) + const response = createStrippedStream(new Response(payload), { + onMessageStart: () => { + throw new Error('observer failure') + }, + }) + expect(await response.text()).toBe(payload) + }) + test('strips tool prefixes from streamed response body', async () => { const chunks = [ 'data: {"type":"content_block_start","content_block":{"type":"tool_use","name":"mcp_bash"}}\n\n', @@ -1264,6 +1450,58 @@ describe('prependClaudeCodeIdentity', () => { }) describe('rewriteRequestBody', () => { + test('injects cache diagnostics with a null previous message id', async () => { + const result = JSON.parse( + await rewriteRequestBody( + JSON.stringify({ messages: [{ role: 'user', content: 'hi' }] }), + { cacheDiagnosticsPreviousMessageId: null }, + ), + ) + expect(result.diagnostics).toEqual({ previous_message_id: null }) + }) + + test('copies an opaque cache diagnostics message id by identity', async () => { + const id = 'msg_opaque_provider_value' + const result = JSON.parse( + await rewriteRequestBody( + JSON.stringify({ messages: [{ role: 'user', content: 'hi' }] }), + { cacheDiagnosticsPreviousMessageId: id }, + ), + ) + expect(result.diagnostics).toEqual({ previous_message_id: id }) + }) + + test('injects diagnostics for structured output bodies', async () => { + const result = JSON.parse( + await rewriteRequestBody( + JSON.stringify({ + messages: [{ role: 'user', content: 'hi' }], + output_config: { format: { type: 'json_schema' } }, + }), + { cacheDiagnosticsPreviousMessageId: null }, + ), + ) + expect(result.diagnostics).toEqual({ previous_message_id: null }) + }) + + test('does not inject diagnostics when the opt-in is omitted', async () => { + const result = JSON.parse( + await rewriteRequestBody( + JSON.stringify({ messages: [{ role: 'user', content: 'hi' }] }), + ), + ) + expect(result.diagnostics).toBeUndefined() + }) + + test('returns invalid JSON unchanged when diagnostics injection is requested', async () => { + const body = '{not json' + expect( + await rewriteRequestBody(body, { + cacheDiagnosticsPreviousMessageId: null, + }), + ).toBe(body) + }) + test('prefixes tool names and rewrites system prompt', async () => { const body = JSON.stringify({ tools: [{ name: 'bash', type: 'function' }], diff --git a/packages/opencode/src/transform.ts b/packages/opencode/src/transform.ts index 06f366e9..d4d7d3c1 100644 --- a/packages/opencode/src/transform.ts +++ b/packages/opencode/src/transform.ts @@ -22,10 +22,15 @@ import { orderClaudeCodeBody, PARAGRAPH_REMOVAL_ANCHORS, REQUIRED_BETAS, + selectClaudeCodeBetas, signRequestBody, TEXT_REPLACEMENTS, TOOL_PREFIX, } from '@cortexkit/anthropic-auth-core' +import { + applyCacheDiagnosticsOptIn, + CACHE_DIAGNOSTICS_BETA, +} from './cache-diagnostics' import { makeByteBoundedMemo } from './sanitize-memo' import { applyServerSideFallbackToBody, @@ -34,6 +39,8 @@ import { type ServerSideFallbackOutcome, } from './server-fallback' +export const NON_STREAMING_DIAGNOSTICS_MAX_BYTES = 8 * 1024 * 1024 + /** * Prefix a tool name with TOOL_PREFIX and uppercase the first character. * Claude Code uses PascalCase tool names (e.g. mcp_Bash, mcp_Read); @@ -1177,6 +1184,7 @@ export async function rewriteRequestBody( perf?: RewritePerfCallback hybridStandbyAnchor?: HybridMessageCacheAnchor serverSideFallbackEnabled?: boolean + cacheDiagnosticsPreviousMessageId?: string | null } = {}, ): Promise { try { @@ -1286,6 +1294,16 @@ export async function rewriteRequestBody( delete parsed.speed } + if ( + options.cacheDiagnosticsPreviousMessageId !== undefined && + selectClaudeCodeBetas(parsed).split(',').includes(CACHE_DIAGNOSTICS_BETA) + ) { + applyCacheDiagnosticsOptIn( + parsed, + options.cacheDiagnosticsPreviousMessageId, + ) + } + const metadataStart = rewriteNowMs() if (options.identity) applyClaudeCodeMetadata(parsed, options.identity) options.perf?.('metadata', { @@ -1328,10 +1346,13 @@ type SseEventSummary = { inputJsonDeltaBytes?: number signatureDeltaBytes?: number redactedThinkingBytes?: number + message?: Record } type SseDiagnosticState = { pending: string + pendingOverflowCount: number + disabled: boolean events: number parseErrors: number eventCounts: Record @@ -1351,6 +1372,8 @@ const sseDiagnosticEncoder = new TextEncoder() function createSseDiagnosticState(): SseDiagnosticState { return { pending: '', + pendingOverflowCount: 0, + disabled: false, events: 0, parseErrors: 0, eventCounts: {}, @@ -1472,6 +1495,7 @@ function summarizeSseEvent(rawEvent: string): SseEventSummary | null { if (usage) { summary.stopReason ??= stringField(message, 'stop_reason') } + if (summary.type === 'message_start' && message) summary.message = message return summary } @@ -1484,8 +1508,13 @@ function findSseBoundary(value: string) { return { index: crlf, length: 4 } } -function updateSseDiagnostics(state: SseDiagnosticState, text: string) { - if (!text) return +function updateSseDiagnostics( + state: SseDiagnosticState, + text: string, + maxPendingBytes: number, + onEvent?: (summary: SseEventSummary) => void, +) { + if (!text || state.disabled) return state.pending += text while (true) { @@ -1508,10 +1537,17 @@ function updateSseDiagnostics(state: SseDiagnosticState, text: string) { state.signatureDeltaBytes += summary.signatureDeltaBytes ?? 0 state.redactedThinkingBytes += summary.redactedThinkingBytes ?? 0 state.last = summary + onEvent?.(summary) if (summary.dataBytes > 0 && !summary.type && !summary.event) { state.parseErrors++ } } + + if (sseDiagnosticEncoder.encode(state.pending).byteLength > maxPendingBytes) { + state.pending = '' + state.disabled = true + state.pendingOverflowCount++ + } } type SseErrorState = { @@ -1521,6 +1557,7 @@ type SseErrorState = { type SseFinishState = { pending: string completed: boolean + disabled: boolean } type SseFinishUpdate = @@ -1528,14 +1565,22 @@ type SseFinishUpdate = | { type: 'complete'; finishReason: string } function createSseFinishState(): SseFinishState { - return { pending: '', completed: false } + return { pending: '', completed: false, disabled: false } } function updateSseFinishState( state: SseFinishState, text: string, + maxPendingBytes: number, ): SseFinishUpdate | null { - if (!text || state.completed) return null + if (!text || state.completed || state.disabled) return null + if ( + new TextEncoder().encode(state.pending + text).byteLength > maxPendingBytes + ) { + state.pending = '' + state.disabled = true + return null + } state.pending += text while (true) { @@ -1675,6 +1720,7 @@ function sseDiagnosticStats(state: SseDiagnosticState) { return { sseEvents: state.events, ssePendingChars: state.pending.length, + ssePendingOverflowCount: state.pendingOverflowCount, sseParseErrors: state.parseErrors, sseEventCounts: { ...state.eventCounts }, sseTypeCounts: { ...state.typeCounts }, @@ -1703,6 +1749,9 @@ export function createStrippedStream( contentFilterModel?: unknown serverSideFallbackModel?: string onServerSideFallbackOutcome?: (outcome: ServerSideFallbackOutcome) => void + onMessageStart?: (message: Record) => void + onMessageResponse?: (message: Record) => void + responseMode?: 'json' } = {}, ): Response { if (!response.body) return response @@ -1710,6 +1759,7 @@ export function createStrippedStream( const reader = response.body.getReader() const decoder = new TextDecoder() const encoder = new TextEncoder() + const jsonMode = options.responseMode === 'json' let pending = '' let chunkCount = 0 let pullCount = 0 @@ -1720,7 +1770,19 @@ export function createStrippedStream( let readerReleased = false let lastProgressAt = rewriteNowMs() const streamStart = rewriteNowMs() - const sseDiagnostics = options.perf ? createSseDiagnosticState() : undefined + const sseDiagnostics = + options.perf || options.onMessageStart + ? createSseDiagnosticState() + : undefined + let responseText = '' + let responseTextBytes = 0 + let responseTextOverflowed = false + const observe = (callback: (() => void) | undefined) => { + if (!callback) return + try { + callback() + } catch {} + } const sseErrors = createSseErrorState() const sseFinish = options.onContentFilter || options.onComplete @@ -1740,6 +1802,7 @@ export function createStrippedStream( const serverSideFallback = options.serverSideFallbackModel ? createServerSideFallbackStreamRewriter({ requestedModel: options.serverSideFallbackModel, + maxPendingBytes: NON_STREAMING_DIAGNOSTICS_MAX_BYTES, onOutcome: options.onServerSideFallbackOutcome, onRefusalAfterToolUse: options.onContentFilter ? () => invokeContentFilter(true) @@ -1749,7 +1812,11 @@ export function createStrippedStream( const updateFinish = (text: string) => { if (!sseFinish) return null - const update = updateSseFinishState(sseFinish, text) + const update = updateSseFinishState( + sseFinish, + text, + NON_STREAMING_DIAGNOSTICS_MAX_BYTES, + ) if (update?.type === 'content-filter' && options.onContentFilter) { return invokeContentFilter() ? retryableFableContentFilterError(options.contentFilterModel) @@ -1779,6 +1846,9 @@ export function createStrippedStream( inputBytes, outputBytes, pendingChars: pending.length, + sseErrorPending: sseErrors.pending.length, + sseFinishPending: sseFinish?.pending.length ?? 0, + serverFallbackPending: serverSideFallback?.pendingLength() ?? 0, rewriteMs: rewriteRoundMs(rewriteMs), totalMs: rewriteRoundMs(rewriteNowMs() - streamStart), ...(sseDiagnostics ? sseDiagnosticStats(sseDiagnostics) : {}), @@ -1808,16 +1878,49 @@ export function createStrippedStream( const readMs = rewriteRoundMs(rewriteNowMs() - readStart) if (done) { const finalDecoded = decoder.decode() - if (sseDiagnostics) - updateSseDiagnostics(sseDiagnostics, finalDecoded) + if (sseDiagnostics && !jsonMode) + updateSseDiagnostics( + sseDiagnostics, + finalDecoded, + NON_STREAMING_DIAGNOSTICS_MAX_BYTES, + (summary) => { + const message = summary.message + if (summary.type === 'message_start' && message) + observe(() => options.onMessageStart?.(message)) + }, + ) + if (jsonMode) { + const finalBytes = encoder.encode(finalDecoded).byteLength + if ( + responseTextBytes + finalBytes <= + NON_STREAMING_DIAGNOSTICS_MAX_BYTES + ) { + responseText += finalDecoded + responseTextBytes += finalBytes + } else { + responseTextOverflowed = true + } + } + if (jsonMode && !responseTextOverflowed) { + try { + const message = JSON.parse(responseText) + if ( + message && + typeof message === 'object' && + !Array.isArray(message) + ) + observe(() => options.onMessageResponse?.(message)) + } catch {} + } const rewriteStart = rewriteNowMs() const serverRewritten = serverSideFallback ? serverSideFallback.push(finalDecoded) + serverSideFallback.flush() : finalDecoded - const retryableStreamError = - updateSseErrorState(sseErrors, finalDecoded) ?? - updateFinish(serverRewritten) + const retryableStreamError = jsonMode + ? updateFinish(serverRewritten) + : (updateSseErrorState(sseErrors, finalDecoded) ?? + updateFinish(serverRewritten)) if (retryableStreamError) { logProgress('stream_tool_prefix_retryable_error', { error: retryableStreamError.message, @@ -1845,14 +1948,37 @@ export function createStrippedStream( chunkCount++ inputBytes += value.byteLength const decoded = decoder.decode(value, { stream: true }) - if (sseDiagnostics) updateSseDiagnostics(sseDiagnostics, decoded) + if (jsonMode) { + const decodedBytes = value.byteLength + if ( + responseTextBytes + decodedBytes <= + NON_STREAMING_DIAGNOSTICS_MAX_BYTES + ) { + responseText += decoded + responseTextBytes += decodedBytes + } else { + responseTextOverflowed = true + } + } + if (sseDiagnostics && !jsonMode) + updateSseDiagnostics( + sseDiagnostics, + decoded, + NON_STREAMING_DIAGNOSTICS_MAX_BYTES, + (summary) => { + const message = summary.message + if (summary.type === 'message_start' && message) + observe(() => options.onMessageStart?.(message)) + }, + ) const rewriteStart = rewriteNowMs() const serverRewritten = serverSideFallback ? serverSideFallback.push(decoded) : decoded - const retryableStreamError = - updateSseErrorState(sseErrors, decoded) ?? - updateFinish(serverRewritten) + const retryableStreamError = jsonMode + ? updateFinish(serverRewritten) + : (updateSseErrorState(sseErrors, decoded) ?? + updateFinish(serverRewritten)) if (retryableStreamError) { logProgress('stream_tool_prefix_retryable_error', { error: retryableStreamError.message,