From e37daaf2bfc272225513d958684595709026fab0 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 | 1 + packages/core/src/cachekeep.ts | 84 ++- packages/core/src/cch.ts | 6 + packages/core/src/claude-code.ts | 9 +- packages/core/src/dump.ts | 80 ++- packages/core/src/relay.ts | 7 +- packages/core/src/tests/dump.test.ts | 166 ++++- packages/opencode/README.md | 70 ++ packages/opencode/src/cache-diagnostics.ts | 325 +++++++++ packages/opencode/src/index.ts | 415 +++++++++-- packages/opencode/src/server-fallback.ts | 44 +- .../src/tests/cache-diagnostics.test.ts | 402 +++++++++++ packages/opencode/src/tests/cachekeep.test.ts | 137 ++++ packages/opencode/src/tests/index.test.ts | 667 ++++++++++++++++++ packages/opencode/src/tests/relay.test.ts | 46 ++ .../src/tests/server-fallback.test.ts | 55 ++ packages/opencode/src/tests/transform.test.ts | 380 ++++++++++ packages/opencode/src/transform.ts | 212 +++++- 18 files changed, 3009 insertions(+), 97 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 a9324263..1a054768 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ This repo is a CortexKit-maintained Anthropic auth monorepo for OpenCode and Pi. - Retry failed CacheKeep prewarms while the last confirmed cache can still be alive, and serialize overlapping manager ticks to avoid duplicate requests. - Preserve fresh scoped-only fallback quota snapshots, permanent refresh-error classification across lock contention, and explicit re-login guidance for unusable fallback accounts. - Evict complete request artifact groups when enforcing the dump-directory size cap. +- Capture Anthropic cache diagnostics in versioned `MC-CACHE-DIAG ` debug 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 diff --git a/packages/core/src/cachekeep.ts b/packages/core/src/cachekeep.ts index bf061fc2..bdb15ff7 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' @@ -332,6 +333,7 @@ export type CacheKeepTarget = { consecutiveFailures: number dayKey: string oauthAccountId?: string + isSubagent: boolean } function cacheKeepRetryDelayMs(targetId: string, failureCount: number) { @@ -381,6 +383,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 }, ) {} @@ -497,6 +510,7 @@ export class CacheKeepManager { storage: AccountStorage | null cacheMode: string oauthAccountId?: string + isSubagent?: boolean }) { if (!input.sessionId) return { tracked: false, reason: 'missing session id' } @@ -534,6 +548,7 @@ export class CacheKeepManager { consecutiveFailures: 0, dayKey: today, oauthAccountId: input.oauthAccountId, + isSubagent: input.isSubagent ?? false, }) this.pruneTargets(now, today) this.publishTrackedSessions() @@ -547,6 +562,7 @@ export class CacheKeepManager { headers: Headers bodyText: string oauthAccountId?: string + isSubagent?: boolean }): Promise { const headers: Record = {} input.headers.forEach((value, key) => { @@ -562,6 +578,7 @@ export class CacheKeepManager { consecutiveFailures: 0, dayKey: '', oauthAccountId: input.oauthAccountId, + isSubagent: input.isSubagent ?? false, } return this.sendPrewarm(target) } @@ -622,11 +639,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), @@ -652,21 +681,54 @@ export class CacheKeepManager { transient: true, } } + 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 466651c6..30b718c5 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 } @@ -329,13 +337,15 @@ function shortAffinity(affinity: string) { return affinity.length <= 16 ? affinity : `${affinity.slice(0, 12)}…` } -function dumpFileSessionSegment(affinity: string) { +function dumpFileSessionSegment(affinity: string, maxLength = 80) { const normalized = affinity .trim() .replace(/[^a-zA-Z0-9._-]+/g, '-') .replace(/^-+|-+$/g, '') if (!normalized) return 'session-unknown' - return normalized.length <= 80 ? normalized : normalized.slice(0, 80) + return normalized.length <= maxLength + ? normalized + : normalized.slice(0, maxLength) } function dumpRequestSegment(input: { @@ -351,6 +361,10 @@ function dumpRequestSegment(input: { return `direct${route}` } +function dumpTagSegment(tag: DumpTag | undefined) { + return tag ? `-prewarm-${tag}` : '' +} + function directDumpPreviousKey(input: { affinity?: string | null route?: string @@ -507,26 +521,30 @@ 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 tagSegment = dumpTagSegment(input.tag) + const id = `${new Date().toISOString().replace(/[:.]/g, '-')}-${String(nextDumpId).padStart(6, '0')}-${dumpFileSessionSegment(affinity, 80 - tagSegment.length)}${tagSegment}-${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` @@ -543,6 +561,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), @@ -591,6 +610,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 } : {}), } } @@ -603,11 +627,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, @@ -620,8 +645,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: { @@ -634,8 +661,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, @@ -645,5 +673,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 16c33c31..75614da1 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,158 @@ 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('sweeps tagged dumps with a maximum-length affinity segment', 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) + + await dumpDirectRequest({ + affinity: 'a'.repeat(80), + bodyText: '{"messages":[]}', + tag: 'cachekeep', + }) + + expect(await readdir(dumpDir)).not.toEqual([]) + const result = await sweepDumpDirectory({ + dumpDir, + maxBytes: 1, + minAgeMs: 0, + now: Date.now() + 1, + }) + + expect(result.removed).toBeGreaterThan(0) + expect(await readdir(dumpDir)).toEqual([]) +}) + +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 cb862492..4d085fed 100644 --- a/packages/opencode/README.md +++ b/packages/opencode/README.md @@ -412,6 +412,76 @@ Prime marker identities live in `anthropic-auth-state.json`. Plugin-owned refres Pi exposes `/claude-prime` as a status-only command. Its `on` and `off` arguments are ignored; enable or disable priming from OpenCode. +### 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. Records and their beta side-channel are emitted only at debug level; use `/claude-logging debug` to capture them. Because emission is gated on the runtime log level, an empty record stream is ambiguous between a quiet system and a level below debug — liveness checks must cross-reference an independent activity witness (billing, usage rows, or dumps) instead of reading silence as absence of cache traffic. 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. + +All required usage counters must be finite, non-negative numbers. A malformed required counter rejects the whole record rather than emitting a partial receipt. + +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":[…]}` at debug level 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..54a87a59 --- /dev/null +++ b/packages/opencode/src/cache-diagnostics.ts @@ -0,0 +1,325 @@ +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 { + const value = usage[key] + return typeof value === 'number' && Number.isFinite(value) && value >= 0 + ? value + : 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 4a53347f..8ac222fe 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -28,11 +28,14 @@ import { CLAUDE_PRIME_COMMAND_NAME, CLAUDE_QUOTAS_COMMAND_NAME, CLAUDE_ROUTING_COMMAND_NAME, + computeXxhash64Hex, continueMainPrimeAuthLineageAfterRefresh, createEmptyStorage, createStickyNoRouteResponse, + type DumpHandle, decideStickyQuotaFailure, dumpDirectRequest, + dumpResponseArtifact, exchange, executeAccountCommand, executeCache1hCommand, @@ -147,11 +150,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, @@ -1188,6 +1203,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.debug( + 'cache-diagnostics', + formatCacheDiagnosticsLogLine(observed.record), + ) + const betaLine = cacheDiagnosticsBetaTracker.capture( + input.betasHash, + input.betas, + ) + if (betaLine) logger.debug('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: @@ -1205,6 +1364,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 @@ -1268,6 +1503,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) } @@ -3680,12 +3921,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, } } @@ -3712,12 +3959,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, } } @@ -3760,6 +4013,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 @@ -3786,6 +4041,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), @@ -3800,6 +4058,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 @@ -3826,7 +4086,7 @@ const anthropicAuthPlugin = async ( throw error } if (typeof body === 'string') { - await dumpDirectRequest({ + dump = await dumpDirectRequest({ affinity: directAffinity, route, status: response.status, @@ -3837,6 +4097,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), @@ -3873,6 +4143,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() @@ -3926,6 +4207,7 @@ const anthropicAuthPlugin = async ( identity, hybridStandbyAnchor: standbyCacheAnchor, serverSideFallbackEnabled: fallbackMode === 'server', + cacheDiagnosticsPreviousMessageId, perf: (stage, data) => { trace?.mark(`rewrite_body_${stage}`, { route, ...data }) if ( @@ -3958,10 +4240,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), @@ -3990,6 +4304,8 @@ const anthropicAuthPlugin = async ( }) } + const cacheDiagnosticsBetas = + await getCacheDiagnosticsBetas(requestHeaders) const rewritten = rewriteUrl(input) if (fableRequest && typeof body === 'string') { fableRequest.warmTarget = { @@ -4015,6 +4331,7 @@ const anthropicAuthPlugin = async ( storage, cacheMode: 'hybrid', oauthAccountId, + isSubagent: subagentRequest, }) trace?.mark('cachekeep_track', { session: relayAffinity, @@ -4037,7 +4354,7 @@ const anthropicAuthPlugin = async ( ...(isInsecure() && { tls: { rejectUnauthorized: false } }), }) if (typeof body === 'string') { - await dumpDirectRequest({ + directDump = await dumpDirectRequest({ affinity: relayAffinity, route, status: response.status, @@ -4085,6 +4402,9 @@ const anthropicAuthPlugin = async ( optimisticResponse: relayConfig?.transport === 'websocket', onResponseHeaders: (headers) => harvestQuotaHeaders(headers, served), + onDumpCreated: (handle) => { + relayDump = handle + }, }) trace?.mark('send_headers_received', { route, @@ -4095,6 +4415,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 } @@ -4356,39 +4688,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, @@ -4629,9 +4928,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 ? { @@ -4745,6 +5058,7 @@ const anthropicAuthPlugin = async ( } : {}), }) + } const authStart = nowMs() const auth = await getAuth() trace.mark('get_auth', { @@ -4984,6 +5298,8 @@ const anthropicAuthPlugin = async ( ), sessionId, proactiveQuotaDecision.retryAfterSeconds, + false, + cacheDiagnosticsResponses, ), false, ) @@ -5078,6 +5394,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..d45971f1 100644 --- a/packages/opencode/src/server-fallback.ts +++ b/packages/opencode/src/server-fallback.ts @@ -216,10 +216,14 @@ 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 resync = false + let resyncCarry = '' let servedModel: string | undefined let handoff: ServerSideFallbackMarker | undefined let fallbackIterationModel: string | undefined @@ -323,21 +327,59 @@ 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) { + let output = '' + if (resync) { + const window = resyncCarry + text + const boundary = findSseBoundary(window) + if (!boundary) { + resyncCarry = window.slice(-3) + return text + } + const end = boundary.index + boundary.length + output = text.slice(0, Math.max(0, end - resyncCarry.length)) + text = window.slice(end) + resync = false + resyncCarry = '' + } + if (!text) return output + const textBytes = new TextEncoder().encode(text).byteLength pending += text - return drain() + pendingBytes += textBytes + output += drain() + if ( + options.maxPendingBytes !== undefined && + pendingBytes > options.maxPendingBytes + ) { + const tail = pending + pending = '' + pendingBytes = 0 + resync = true + resyncCarry = tail.slice(-3) + return output + tail + } + return output }, flush() { + if (resync) { + resyncCarry = '' + 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..f7601760 --- /dev/null +++ b/packages/opencode/src/tests/cache-diagnostics.test.ts @@ -0,0 +1,402 @@ +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.each([ + ['negative cache_read', { cache_read_input_tokens: -1 }], + ['non-finite input_tokens', { input_tokens: Number.POSITIVE_INFINITY }], + ])('rejects a record with %s', (_name, usageOverrides) => { + expect( + buildCacheDiagnosticsRecord( + input({ + message: { + ...message({ cache_miss_reason: { type: 'unavailable' } }), + usage: { ...usage, ...usageOverrides }, + }, + }), + ), + ).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 415f6d3b..783289c4 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 1b34dcc8..e870ebba 100644 --- a/packages/opencode/src/tests/index.test.ts +++ b/packages/opencode/src/tests/index.test.ts @@ -9465,6 +9465,673 @@ 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: {} }) + setLogLevel('debug') + 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', + }) + setLogLevel('info') + }) + + 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: {} }) + setLogLevel('debug') + 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, + }) + setLogLevel('info') + }) + + 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 mockClient = createMockClient() + const plugin = await getPlugin(mockClient) + const result = await plugin.auth.loader(oauthLoader, { models: {} }) + await expect( + plugin['command.execute.before']({ + command: 'claude-logging', + arguments: 'debug', + sessionID: 'session-cache-diagnostics', + }), + ).rejects.toThrow('__OPENCODE_ANTHROPIC_AUTH_COMMAND_HANDLED__') + 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) + expect( + records + .filter( + (record) => + record.channel === 'cache-diagnostics' && + record.message.startsWith('MC-CACHE-DIAG'), + ) + .map((record) => record.level), + ).toEqual(['debug', 'debug', 'debug', 'debug']) + setLogLevel('info') + }) + + 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: {} }) + setLogLevel('debug') + 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) + setLogLevel('info') + }) + + 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: {} }) + setLogLevel('debug') + 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}$/), + }) + setLogLevel('info') + }) + + 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/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/server-fallback.test.ts b/packages/opencode/src/tests/server-fallback.test.ts index 9c97ee92..910223f1 100644 --- a/packages/opencode/src/tests/server-fallback.test.ts +++ b/packages/opencode/src/tests/server-fallback.test.ts @@ -450,4 +450,59 @@ describe('createServerSideFallbackStreamRewriter', () => { }), ]) }) + + test.each([ + ['LF', '\n\n', 1], + ['CRLF', '\r\n\r\n', 2], + ])('resyncs after an oversized frame ending in a split %s boundary', (_name, delimiter, splitAt) => { + let handled = 0 + const skipped = `data: ${'x'.repeat(256)}` + const later = [ + sse('content_block_start', { + type: 'content_block_start', + index: 0, + content_block: { + type: 'tool_use', + id: 'toolu_after_overflow', + name: 'mcp_Bash', + input: {}, + }, + }), + sse('content_block_stop', { type: 'content_block_stop', index: 0 }), + sse('content_block_start', { + type: 'content_block_start', + index: 1, + content_block: { + type: 'fallback', + from: { model: 'claude-fable-5' }, + to: { model: 'claude-opus-4-8' }, + }, + }), + sse('message_delta', { + type: 'message_delta', + delta: { stop_reason: 'refusal' }, + }), + ].join('') + const rewriter = createServerSideFallbackStreamRewriter({ + requestedModel: 'claude-fable-5', + maxPendingBytes: 64, + onRefusalAfterToolUse: () => { + handled++ + return true + }, + }) + + const output = + rewriter.push(`${skipped}${delimiter.slice(0, splitAt)}`) + + rewriter.push(`${delimiter.slice(splitAt)}${later}`) + + rewriter.flush() + + expect(output.slice(0, skipped.length + delimiter.length)).toBe( + `${skipped}${delimiter}`, + ) + expect(output).toContain(SERVER_FALLBACK_SIGNATURE_PREFIX) + expect(output).toContain('"stop_reason":"tool_use"') + expect(output).not.toContain('"stop_reason":"refusal"') + expect(handled).toBe(1) + }) }) diff --git a/packages/opencode/src/tests/transform.test.ts b/packages/opencode/src/tests/transform.test.ts index 2368f3d3..930aa196 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,288 @@ 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('drains an over-cap refusal terminal frame before disabling the tail', async () => { + const contentFilters: boolean[] = [] + const perf: Array> = [] + const terminal = sse('message_delta', { + type: 'message_delta', + delta: { stop_reason: 'refusal' }, + }) + const body = `${terminal}${'x'.repeat(NON_STREAMING_DIAGNOSTICS_MAX_BYTES * 2)}` + const response = createStrippedStream( + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(body)) + controller.close() + }, + }), + ), + { + onContentFilter: () => { + contentFilters.push(true) + return false + }, + perf: (_stage, stats) => perf.push(stats ?? {}), + }, + ) + + expect(await response.text()).toBe(body) + expect(contentFilters).toEqual([true]) + expect(perf.at(-1)).toMatchObject({ + sseFinishPending: 0, + sseFinishDisabled: true, + }) + }) + + test('drains an over-cap ordinary terminal frame before disabling the tail', async () => { + const completed: string[] = [] + const perf: Array> = [] + const terminal = sse('message_delta', { + type: 'message_delta', + delta: { stop_reason: 'end_turn' }, + }) + const body = `${terminal}${'x'.repeat(NON_STREAMING_DIAGNOSTICS_MAX_BYTES * 2)}` + const response = createStrippedStream( + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(body)) + controller.close() + }, + }), + ), + { + onComplete: (finishReason) => completed.push(finishReason), + perf: (_stage, stats) => perf.push(stats ?? {}), + }, + ) + + expect(await response.text()).toBe(body) + expect(completed).toEqual(['end_turn']) + expect(perf.at(-1)).toMatchObject({ + sseFinishPending: 0, + sseFinishDisabled: true, + }) + }) + + test.each([ + ['LF', '\n\n', 1], + ['CRLF', '\r\n\r\n', 2], + ])('resyncs finish detection after an oversized frame ending in a split %s boundary', async (_name, delimiter, splitAt) => { + const completed: string[] = [] + const skipped = `data: ${'x'.repeat(NON_STREAMING_DIAGNOSTICS_MAX_BYTES * 2)}` + const terminal = sse('message_delta', { + type: 'message_delta', + delta: { stop_reason: 'end_turn' }, + }) + const body = `${skipped}${delimiter}${terminal}` + const response = createStrippedStream( + new Response( + new ReadableStream({ + start(controller) { + const encoder = new TextEncoder() + controller.enqueue( + encoder.encode(`${skipped}${delimiter.slice(0, splitAt)}`), + ) + controller.enqueue( + encoder.encode(`${delimiter.slice(splitAt)}${terminal}`), + ) + controller.close() + }, + }), + ), + { onComplete: (finishReason) => completed.push(finishReason) }, + ) + + expect(await response.text()).toBe(body) + expect(completed).toEqual(['end_turn']) + }) + + 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', @@ -666,6 +950,50 @@ describe('createStrippedStream', () => { ).toBe(true) }) + test.each([ + ['LF', '\n\n', 1], + ['CRLF', '\r\n\r\n', 2], + ])('resyncs retryable stream errors after an oversized frame ending in a split %s boundary', async (_name, delimiter, splitAt) => { + const encoder = new TextEncoder() + const perf: Array> = [] + const skipped = `data: ${'x'.repeat(NON_STREAMING_DIAGNOSTICS_MAX_BYTES * 2)}` + const error = sse('error', { + type: 'error', + error: { type: 'overloaded_error', message: 'temporarily overloaded' }, + }) + const response = createStrippedStream( + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue( + encoder.encode(`${skipped}${delimiter.slice(0, splitAt)}`), + ) + controller.enqueue( + encoder.encode(`${delimiter.slice(splitAt)}${error}`), + ) + controller.close() + }, + }), + ), + { perf: (_stage, stats) => perf.push(stats ?? {}) }, + ) + + let caught: unknown + try { + await response.text() + } catch (error) { + caught = error + } + + expect((caught as { code?: string }).code).toBe('ECONNRESET') + expect((caught as { providerErrorType?: string }).providerErrorType).toBe( + 'overloaded_error', + ) + expect( + Math.max(...perf.map((stats) => Number(stats.sseErrorPending ?? 0))), + ).toBeLessThanOrEqual(NON_STREAMING_DIAGNOSTICS_MAX_BYTES) + }) + test('reports server-side fallback outcomes without turning them into retryable errors', async () => { const encoder = new TextEncoder() const outcomes: ServerSideFallbackOutcome[] = [] @@ -1264,6 +1592,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..dc8776ba 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,19 +1537,30 @@ 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 = { pending: string + disabled: boolean + resyncCarry: string } type SseFinishState = { pending: string completed: boolean + disabled: boolean + resyncCarry: string } type SseFinishUpdate = @@ -1528,28 +1568,50 @@ type SseFinishUpdate = | { type: 'complete'; finishReason: string } function createSseFinishState(): SseFinishState { - return { pending: '', completed: false } + return { pending: '', completed: false, disabled: false, resyncCarry: '' } } function updateSseFinishState( state: SseFinishState, text: string, + maxPendingBytes: number, ): SseFinishUpdate | null { if (!text || state.completed) return null + if (state.disabled) { + const window = state.resyncCarry + text + const boundary = findSseBoundary(window) + if (!boundary) { + state.resyncCarry = window.slice(-3) + return null + } + state.disabled = false + state.resyncCarry = '' + text = window.slice(boundary.index + boundary.length) + if (!text) return null + } state.pending += text + let update: SseFinishUpdate | null = null while (true) { const boundary = findSseBoundary(state.pending) - if (!boundary) return null + if (!boundary) break const rawEvent = state.pending.slice(0, boundary.index) state.pending = state.pending.slice(boundary.index + boundary.length) const summary = summarizeSseEvent(rawEvent) if (summary?.type !== 'message_delta' || !summary.stopReason) continue state.completed = true - return summary.stopReason === 'refusal' - ? { type: 'content-filter' } - : { type: 'complete', finishReason: summary.stopReason } + update ??= + summary.stopReason === 'refusal' + ? { type: 'content-filter' } + : { type: 'complete', finishReason: summary.stopReason } } + if (new TextEncoder().encode(state.pending).byteLength > maxPendingBytes) { + const tail = state.pending + state.pending = '' + state.disabled = true + state.resyncCarry = tail.slice(-3) + } + return update } type RetryableAnthropicStreamError = Error & { @@ -1559,7 +1621,7 @@ type RetryableAnthropicStreamError = Error & { } function createSseErrorState(): SseErrorState { - return { pending: '' } + return { pending: '', disabled: false, resyncCarry: '' } } function isRetryableAnthropicStreamError( @@ -1654,9 +1716,23 @@ function retryableAnthropicStreamErrorFromRawEvent( function updateSseErrorState( state: SseErrorState, text: string, + maxPendingBytes: number, ): RetryableAnthropicStreamError | null { if (!text) return null + if (state.disabled) { + const window = state.resyncCarry + text + const boundary = findSseBoundary(window) + if (!boundary) { + state.resyncCarry = window.slice(-3) + return null + } + state.disabled = false + state.resyncCarry = '' + text = window.slice(boundary.index + boundary.length) + if (!text) return null + } state.pending += text + let retryable: RetryableAnthropicStreamError | null = null while (true) { const boundary = findSseBoundary(state.pending) @@ -1665,16 +1741,23 @@ function updateSseErrorState( const rawEvent = state.pending.slice(0, boundary.index) state.pending = state.pending.slice(boundary.index + boundary.length) const error = retryableAnthropicStreamErrorFromRawEvent(rawEvent) - if (error) return error + retryable ??= error } - return null + if (new TextEncoder().encode(state.pending).byteLength > maxPendingBytes) { + const tail = state.pending + state.pending = '' + state.disabled = true + state.resyncCarry = tail.slice(-3) + } + return retryable } 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 +1786,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 +1796,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 +1807,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 +1839,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 +1849,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 +1883,10 @@ export function createStrippedStream( inputBytes, outputBytes, pendingChars: pending.length, + sseErrorPending: sseErrors.pending.length, + sseFinishPending: sseFinish?.pending.length ?? 0, + sseFinishDisabled: sseFinish?.disabled ?? false, + serverFallbackPending: serverSideFallback?.pendingLength() ?? 0, rewriteMs: rewriteRoundMs(rewriteMs), totalMs: rewriteRoundMs(rewriteNowMs() - streamStart), ...(sseDiagnostics ? sseDiagnosticStats(sseDiagnostics) : {}), @@ -1808,16 +1916,52 @@ 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, + NON_STREAMING_DIAGNOSTICS_MAX_BYTES, + ) ?? updateFinish(serverRewritten)) if (retryableStreamError) { logProgress('stream_tool_prefix_retryable_error', { error: retryableStreamError.message, @@ -1845,14 +1989,40 @@ 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, + NON_STREAMING_DIAGNOSTICS_MAX_BYTES, + ) ?? updateFinish(serverRewritten)) if (retryableStreamError) { logProgress('stream_tool_prefix_retryable_error', { error: retryableStreamError.message,