From 1132c3e31c9038d31465a972b59484935c529aaa Mon Sep 17 00:00:00 2001 From: can1357 Date: Thu, 13 Aug 2026 03:53:47 +0200 Subject: [PATCH] feat: implemented anthropic keep-alive and migration support for updates - Added Anthropic prompt-cache refresh scheduling and state management to keep prompts warm across idle sessions. - Updated pricing models and database stats tracking to calculate and store cost-weighted cache savings. - Integrated cache savings metrics and efficiency displays into the stats CLI, dashboard routes, and UI components. - Added support for package renaming, manifest pointer tracking, and installation migration during CLI updates. --- nix/dev-shell.nix | 4 +- nix/package.nix | 26 +- packages/ai/CHANGELOG.md | 1 + packages/ai/src/providers/anthropic-client.ts | 16 +- packages/ai/src/providers/anthropic.ts | 366 +++++------------- packages/ai/src/stream.ts | 291 +++++++++++++- packages/ai/src/types.ts | 18 +- .../ai/src/utils/empty-completion-retry.ts | 4 +- packages/ai/test/anthropic-alignment.test.ts | 150 ++----- .../ai/test/anthropic-cache-refresh.test.ts | 287 ++++++++++++++ .../ai/test/anthropic-stream-envelope.test.ts | 13 +- packages/catalog/CHANGELOG.md | 2 + packages/catalog/src/models.ts | 25 +- packages/coding-agent/CHANGELOG.md | 2 + packages/coding-agent/src/cli/stats-cli.ts | 3 +- packages/coding-agent/src/cli/update-cli.ts | 279 +++++++++++-- packages/coding-agent/src/main.ts | 17 +- packages/coding-agent/src/sdk.ts | 1 + .../src/web/search/providers/anthropic.ts | 1 - ...pdate-rename-migration.integration.test.ts | 155 ++++++++ packages/coding-agent/test/update-cli.test.ts | 16 +- packages/natives/scripts/build-bindings.ts | 14 +- packages/stats/CHANGELOG.md | 4 + packages/stats/README.md | 3 +- .../stats/src/client/routes/ModelsRoute.tsx | 16 +- .../stats/src/client/routes/ProjectsRoute.tsx | 16 +- packages/stats/src/client/styles.css | 2 +- .../stats/src/client/ui/MetricCluster.tsx | 12 +- packages/stats/src/db.ts | 112 +++++- packages/stats/src/index.ts | 3 +- packages/stats/src/shared-types.ts | 7 +- packages/stats/test/db-cost.test.ts | 80 +++- .../stats/test/overview-token-labels.test.tsx | 6 + scripts/bazel-natives.ts | 34 +- scripts/host-detect.ts | 20 + 35 files changed, 1488 insertions(+), 518 deletions(-) create mode 100644 packages/ai/test/anthropic-cache-refresh.test.ts create mode 100644 packages/coding-agent/test/cli/update-rename-migration.integration.test.ts diff --git a/nix/dev-shell.nix b/nix/dev-shell.nix index fec011f05..2a5cfe3ef 100644 --- a/nix/dev-shell.nix +++ b/nix/dev-shell.nix @@ -20,7 +20,6 @@ pkgs.mkShell ( bun bun2nix rustToolchain - bazelisk cargo-nextest rustPlatform.bindgenHook nixfmt @@ -61,6 +60,9 @@ pkgs.mkShell ( ++ lib.optionals pkgs.stdenv.hostPlatform.isLinux linuxLibraries; CMAKE_POLICY_VERSION_MINIMUM = "3.5"; + # Bazel's downloaded host tools assume an FHS loader; Cargo is the + # repository's supported local-iteration path inside the Nix shell. + OMP_NATIVE_BUILD_BACKEND = "cargo"; PCRE2_SYS_STATIC = "1"; RUST_SRC_PATH = "${rustToolchain}/lib/rustlib/src/rust/library"; } diff --git a/nix/package.nix b/nix/package.nix index 79c8dfd4e..d306f8c4a 100644 --- a/nix/package.nix +++ b/nix/package.nix @@ -67,17 +67,6 @@ let runHook postInstall ''; }; - runtimeLibraries = [ - libopus - openssl - pcre2 - ] - ++ lib.optionals stdenv.hostPlatform.isLinux [ - libpulseaudio - pipewire - stdenv.cc.cc.lib - zlib - ]; in stdenv.mkDerivation ( { @@ -170,11 +159,18 @@ stdenv.mkDerivation ( install -Dm755 packages/coding-agent/dist/omp "$out/libexec/omp/omp" + ${lib.optionalString stdenv.hostPlatform.isLinux '' + # The addon is gzip-compressed inside the compiled binary, so the Nix + # store RUNPATH autoPatchelf wrote into it is invisible to the output + # reference scanner. Record it in plain text to pin those libraries + # (pipewire, libopus, libgcc) into the runtime closure. + mkdir -p "$out/nix-support" + patchelf --print-rpath "packages/natives/native/${platform.addon}" \ + > "$out/nix-support/embedded-addon-runpath" + ''} + makeWrapper "$out/libexec/omp/omp" "$out/bin/omp" \ - --set PI_SKIP_VERSION_CHECK 1 \ - ${ - if stdenv.hostPlatform.isLinux then "--prefix LD_LIBRARY_PATH" else "--prefix DYLD_LIBRARY_PATH" - } : "${lib.makeLibraryPath runtimeLibraries}" + --set PI_SKIP_VERSION_CHECK 1 runHook postInstall ''; diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 41506c50b..695ead872 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -5,6 +5,7 @@ ### Changed - OpenCode Go usage now comes from the official `GET /zen/go/v1/usage` endpoint (rolling 5h / weekly / monthly percent windows with server-computed resets) instead of synthesizing dollar estimates from OMP-observed request costs, so `/usage` reflects spend made outside OMP and the hardcoded $12/$30/$60 caps are gone. The usage probe now validates credentials (401 invalid key, 403 lapsed Go subscription), and a new ranking strategy routes multi-key pools by rolling/weekly headroom while keeping the monthly window display-only (an exhausted monthly can still serve requests via the console "Use balance" fallback) ([#8337](https://github.com/can1357/oh-my-pi/pull/8337) by [@will-bogusz](https://github.com/will-bogusz)). +- Changed direct Anthropic prompt caching to two rolling 5-minute breakpoints on the latest real messages. Opt-in session owners can now keep the prefix warm through three bounded idle refreshes, using non-streaming `max_tokens: 0` when thinking is inactive and aborting immediately after generation begins when thinking is active. ### Fixed diff --git a/packages/ai/src/providers/anthropic-client.ts b/packages/ai/src/providers/anthropic-client.ts index ea2fe6d38..3b8e0a751 100644 --- a/packages/ai/src/providers/anthropic-client.ts +++ b/packages/ai/src/providers/anthropic-client.ts @@ -27,7 +27,7 @@ import { AnthropicApiError, AnthropicConnectionError, AnthropicConnectionTimeout export { AnthropicApiError, AnthropicConnectionError, AnthropicConnectionTimeoutError }; import type { FetchImpl } from "../types"; -import type { MessageCreateParamsStreaming } from "./anthropic-wire"; +import type { MessageCreateParams } from "./anthropic-wire"; /** Default pre-response timeout, matching the SDK's 10-minute default. */ const DEFAULT_TIMEOUT_MS = 600_000; @@ -173,7 +173,7 @@ export class AnthropicMessages { this.#path = path; } - create(params: MessageCreateParamsStreaming, options?: AnthropicRequestOptions): AnthropicApiRequest { + create(params: MessageCreateParams, options?: AnthropicRequestOptions): AnthropicApiRequest { return this.#client.request(this.#path, params, options); } } @@ -184,8 +184,8 @@ export class AnthropicMessages { * alternative Messages-API client via `AnthropicOptions.client`. */ export interface AnthropicMessagesClientLike { - messages: { create(params: MessageCreateParamsStreaming, options?: AnthropicRequestOptions): unknown }; - beta?: { messages: { create(params: MessageCreateParamsStreaming, options?: AnthropicRequestOptions): unknown } }; + messages: { create(params: MessageCreateParams, options?: AnthropicRequestOptions): unknown }; + beta?: { messages: { create(params: MessageCreateParams, options?: AnthropicRequestOptions): unknown } }; } export class AnthropicMessagesClient implements AnthropicMessagesClientLike { @@ -199,7 +199,7 @@ export class AnthropicMessagesClient implements AnthropicMessagesClientLike { this.beta = { messages: new AnthropicMessages(this, "/v1/messages?beta=true") }; } - request(path: string, params: MessageCreateParamsStreaming, options?: AnthropicRequestOptions): AnthropicApiRequest { + request(path: string, params: MessageCreateParams, options?: AnthropicRequestOptions): AnthropicApiRequest { return new AnthropicApiRequest(() => this.#send(path, params, options)); } @@ -218,11 +218,7 @@ export class AnthropicMessagesClient implements AnthropicMessagesClientLike { return headers; } - async #send( - path: string, - params: MessageCreateParamsStreaming, - options?: AnthropicRequestOptions, - ): Promise { + async #send(path: string, params: MessageCreateParams, options?: AnthropicRequestOptions): Promise { const opts = this.#options; const fetchFn: FetchImpl = opts.fetch ?? fetch; const callerSignal = options?.signal; diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index f83d7e39d..2d3a49170 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/packages/ai/src/providers/anthropic.ts @@ -80,6 +80,7 @@ import { type ContentBlockParam, type FallbackParam, isAnthropicWebSearchHistoryBlock, + type MessageCreateParams, type MessageCreateParamsStreaming, type MessageParam, type RawMessageStreamEvent, @@ -482,17 +483,11 @@ function dropAnthropicStrictTools(params: MessageCreateParamsStreaming): void { function getCacheControl( model: Model<"anthropic-messages">, cacheRetention: CacheRetention | undefined, - isOAuthToken: boolean, ): { retention: CacheRetention; cacheControl?: AnthropicCacheControl } { - // OAuth mirrors Claude Code and always defaults to 1h retention. API-key - // requests also default to 1h where the endpoint supports it (canonical - // Anthropic API, `compat.supportsLongCacheRetention`): agent sessions - // routinely idle past 5 minutes waiting on background jobs, and a 5m - // breakpoint cold-misses the entire prefix on resume. PI_CACHE_RETENTION - // still overrides the API-key default in either direction. - const retention = isOAuthToken - ? (cacheRetention ?? "long") - : resolveCacheRetention(cacheRetention, model.compat.supportsLongCacheRetention ? "long" : "short"); + // Five-minute writes are the cheapest cache population strategy. Longer + // retention remains an explicit PI_CACHE_RETENTION/request override; idle + // sessions keep the short entry warm with bounded read-only refreshes. + const retention = resolveCacheRetention(cacheRetention, "short"); if (retention === "none") { return { retention }; } @@ -1653,6 +1648,31 @@ export function applyAnthropicUsageExtras(usage: Usage, source: AnthropicUsageLi } } +function parseAnthropicWireUsage(value: unknown): AnthropicWireUsage | undefined { + if (!isRecord(value)) return undefined; + const cacheCreation = isRecord(value.cache_creation) + ? { + ...(typeof value.cache_creation.ephemeral_5m_input_tokens === "number" + ? { ephemeral_5m_input_tokens: value.cache_creation.ephemeral_5m_input_tokens } + : {}), + ...(typeof value.cache_creation.ephemeral_1h_input_tokens === "number" + ? { ephemeral_1h_input_tokens: value.cache_creation.ephemeral_1h_input_tokens } + : {}), + } + : undefined; + return { + ...(typeof value.input_tokens === "number" ? { input_tokens: value.input_tokens } : {}), + ...(typeof value.output_tokens === "number" ? { output_tokens: value.output_tokens } : {}), + ...(typeof value.cache_read_input_tokens === "number" + ? { cache_read_input_tokens: value.cache_read_input_tokens } + : {}), + ...(typeof value.cache_creation_input_tokens === "number" + ? { cache_creation_input_tokens: value.cache_creation_input_tokens } + : {}), + ...(cacheCreation === undefined ? {} : { cache_creation: cacheCreation }), + }; +} + function parseAnthropicFallbackWireBlock(value: unknown): AnthropicFallbackContent | undefined { if (!isRecord(value) || value.type !== "fallback") return undefined; const from = isRecord(value.from) && typeof value.from.model === "string" ? value.from.model : undefined; @@ -1857,6 +1877,7 @@ const streamAnthropicOnce = ( }); } + const zeroOutputCacheRefresh = options?.anthropicCacheRefreshRequest === true; let client: AnthropicMessagesClientLike; let isOAuthToken: boolean; @@ -1927,7 +1948,7 @@ const streamAnthropicOnce = ( // requests must not deviate from CC's header fingerprint. if ( !(options?.isOAuth ?? isAnthropicOAuthToken(apiKey)) && - getCacheControl(model, options?.cacheRetention, false).cacheControl?.ttl === "1h" && + getCacheControl(model, options?.cacheRetention).cacheControl?.ttl === "1h" && !extraBetas.includes(extendedCacheTtlBeta) ) { extraBetas.push(extendedCacheTtlBeta); @@ -1958,7 +1979,7 @@ const streamAnthropicOnce = ( model, apiKey, extraBetas, - stream: true, + stream: !zeroOutputCacheRefresh, interleavedThinking: options?.interleavedThinking ?? true, headers: options?.headers, dynamicHeaders: copilotDynamicHeaders?.headers, @@ -2005,6 +2026,60 @@ const streamAnthropicOnce = ( return nextParams; }; let params = await prepareParams(); + const idleTimeoutMs = options?.streamIdleTimeoutMs ?? getStreamIdleTimeoutMs(model.compat.streamIdleTimeoutMs); + const firstEventTimeoutMs = options?.streamFirstEventTimeoutMs ?? getStreamFirstEventTimeoutMs(idleTimeoutMs); + const requestTimeoutMs = + firstEventTimeoutMs !== undefined && firstEventTimeoutMs > 0 ? firstEventTimeoutMs : undefined; + + if (zeroOutputCacheRefresh) { + const refreshParams: MessageCreateParams = { ...params, max_tokens: 0, stream: false }; + rawRequestDump = { + provider: model.provider, + api: output.api, + model: model.id, + method: "POST", + url: `${baseUrl}/v1/messages${isOAuthToken ? "?beta=true" : ""}`, + body: refreshParams, + }; + const { requestSignal } = activeAbortTracker; + const requestOptions = { + ...createSdkStreamRequestOptions(requestSignal, requestTimeoutMs), + maxRetries: 0, + }; + const request: unknown = + isOAuthToken && client.beta + ? client.beta.messages.create(refreshParams, requestOptions) + : client.messages.create(refreshParams, requestOptions); + if (!hasAnthropicRawResponseRequest(request)) { + throw new AIError.AnthropicStreamEnvelopeError( + "Anthropic cache refresh request did not expose a raw response", + ); + } + const response = await request.asResponse(); + await notifyProviderResponse(options, response, model, response.headers.get("request-id")); + const body: unknown = await response.json(); + if (!isRecord(body)) { + throw new AIError.AnthropicStreamEnvelopeError("Anthropic cache refresh returned a malformed response"); + } + const wireUsage = parseAnthropicWireUsage(body.usage); + if (!wireUsage) { + throw new AIError.AnthropicStreamEnvelopeError("Anthropic cache refresh response omitted usage"); + } + if (typeof body.id === "string") output.responseId = body.id; + output.usage.input = wireUsage.input_tokens ?? 0; + output.usage.output = wireUsage.output_tokens ?? 0; + output.usage.cacheRead = wireUsage.cache_read_input_tokens ?? 0; + output.usage.cacheWrite = wireUsage.cache_creation_input_tokens ?? 0; + applyAnthropicUsageExtras(output.usage, wireUsage); + output.usage.totalTokens = + output.usage.input + output.usage.output + output.usage.cacheRead + output.usage.cacheWrite; + calculateCost(model, output.usage); + output.duration = performance.now() - startTime; + stream.push({ type: "start", partial: output }); + stream.push({ type: "done", reason: "stop", message: output }); + stream.end(); + return; + } // Opt-in flag: the response parser only honors `fallback` content // blocks and `usage.iterations` when the current request opted into @@ -2019,10 +2094,6 @@ const streamAnthropicOnce = ( | (AnthropicServerToolContent & { [kStreamingPartialJson]?: string }) | (ToolCall & { [kStreamingPartialJson]: string; [kStreamingLastParseLen]?: number }) ) & { [kStreamingBlockIndex]: number }; - const idleTimeoutMs = options?.streamIdleTimeoutMs ?? getStreamIdleTimeoutMs(model.compat.streamIdleTimeoutMs); - const firstEventTimeoutMs = options?.streamFirstEventTimeoutMs ?? getStreamFirstEventTimeoutMs(idleTimeoutMs); - const requestTimeoutMs = - firstEventTimeoutMs !== undefined && firstEventTimeoutMs > 0 ? firstEventTimeoutMs : undefined; const blocks = output.content as Block[]; const finalizeStreamBlock = (block: Block, contentIndex: number): void => { if (block.type === "text") { @@ -2807,65 +2878,19 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = (model, con export type AnthropicSystemBlock = { type: "text"; text: string; - cache_control?: AnthropicCacheControl; }; type SystemBlockOptions = { includeClaudeCodeInstruction?: boolean; extraInstructions?: string[]; /** Text of the first user message — used as fingerprint seed for the billing header. */ firstUserMessageText?: string; - cacheControl?: AnthropicCacheControl; }; -/** - * Place system-block cache breakpoints that survive volatile project context. - * - * omp normally appends its project footer (cwd, date, workspace tree) after the - * stable system prefix. When cwd is outside a single direct child repository, - * an active-repo context block follows that footer. Caching up to the last three - * eligible blocks therefore covers both layouts: - * - * - stable prefix, project footer - * - stable prefix, project footer, active-repo context - * - * A footer change can then fall back to the stable-prefix entry instead of - * re-writing the entire system cache (issue #7324). - * - * @returns breakpoints placed, capped by `maxBreakpoints`. - */ -function cacheSystemPrefixBreakpoints( - blocks: AnthropicSystemBlock[], - cacheControl: AnthropicCacheControl | undefined, - maxBreakpoints: number, - firstCacheableIndex: number, -): number { - if (!cacheControl || maxBreakpoints <= 0) return 0; - let placed = 0; - for (let index = blocks.length - 1; index >= firstCacheableIndex && placed < maxBreakpoints; index--) { - if (blocks[index].cache_control != null) continue; - blocks[index] = { ...blocks[index], cache_control: cloneAnthropicCacheControl(cacheControl) }; - placed++; - } - return placed; -} - -/** - * First system-block index that may carry a cache breakpoint. Skips the OAuth - * cloak blocks that must stay uncached: the CC billing header (block 0, a - * per-request fingerprint) and the Claude Code identity instruction (block 1). - */ -function firstCacheableSystemIndex(blocks: readonly AnthropicSystemBlock[]): number { - let index = 0; - if (blocks[index]?.text?.startsWith(CLAUDE_BILLING_HEADER_PREFIX)) index++; - if (blocks[index]?.text === claudeCodeSystemInstruction) index++; - return index; -} - export function buildAnthropicSystemBlocks( systemPrompt: readonly string[] | undefined, options: SystemBlockOptions = {}, ): AnthropicSystemBlock[] | undefined { - const { includeClaudeCodeInstruction = false, extraInstructions = [], firstUserMessageText, cacheControl } = options; + const { includeClaudeCodeInstruction = false, extraInstructions = [], firstUserMessageText } = options; const sanitizedPrompts = normalizeSystemPrompts(systemPrompt); const trimmedInstructions = extraInstructions.map(instruction => instruction.trim()).filter(Boolean); const hasBillingHeader = sanitizedPrompts.some(prompt => prompt.startsWith(CLAUDE_BILLING_HEADER_PREFIX)); @@ -2882,7 +2907,6 @@ export function buildAnthropicSystemBlocks( for (const prompt of sanitizedPrompts) { blocks.push({ type: "text", text: prompt }); } - cacheSystemPrefixBreakpoints(blocks, cacheControl, 3, firstCacheableSystemIndex(blocks)); return blocks; } @@ -2894,10 +2918,6 @@ export function buildAnthropicSystemBlocks( for (const prompt of sanitizedPrompts) { blocks.push({ type: "text", text: prompt }); } - const lastIndex = blocks.length - 1; - if (cacheControl && lastIndex >= 0 && blocks[lastIndex].cache_control == null) { - blocks[lastIndex] = { ...blocks[lastIndex], cache_control: cloneAnthropicCacheControl(cacheControl) }; - } return blocks.length > 0 ? blocks : undefined; } @@ -3149,29 +3169,17 @@ function ensureMaxTokensForThinking(params: MessageCreateParamsStreaming, maxAll thinking.budget_tokens = clampedBudget; } -type CacheControlBlock = { - cache_control?: AnthropicCacheControl | null; -}; - -function applyCacheControlToLastTextBlock( - blocks: Array, - cacheControl: AnthropicCacheControl, -): boolean { - if (blocks.length === 0) return false; - for (let i = blocks.length - 1; i >= 0; i--) { - if (blocks[i].type === "text") { - if (blocks[i].cache_control != null) return false; - blocks[i] = { ...blocks[i], cache_control: cloneAnthropicCacheControl(cacheControl) }; - return true; +function applyCacheControlToLastBlock(blocks: ContentBlockParam[], cacheControl: AnthropicCacheControl): boolean { + for (let index = blocks.length - 1; index >= 0; index--) { + const block = blocks[index]; + // Anthropic rejects cache_control on generated reasoning and fallback + // boundary blocks. Preserve the requested trailing boundary on every + // ordinary content block, including tool use and tool results. + if (block.type === "thinking" || block.type === "redacted_thinking" || block.type === "fallback") { + continue; } - } - // No text block — fall back to the last block that accepts cache_control; - // thinking/redacted_thinking blocks reject the field with a 400. - for (let i = blocks.length - 1; i >= 0; i--) { - const type = blocks[i].type; - if (type === "thinking" || type === "redacted_thinking") continue; - if (blocks[i].cache_control != null) return false; - blocks[i] = { ...blocks[i], cache_control: cloneAnthropicCacheControl(cacheControl) }; + if ("cache_control" in block && block.cache_control != null) return false; + blocks[index] = { ...block, cache_control: cloneAnthropicCacheControl(cacheControl) }; return true; } return false; @@ -3180,28 +3188,10 @@ function applyCacheControlToLastTextBlock( function applyPromptCaching(params: MessageCreateParamsStreaming, cacheControl?: AnthropicCacheControl): void { if (!cacheControl) return; - const MAX_CACHE_BREAKPOINTS = 4; - let cacheBreakpointsUsed = countCacheControlBreakpoints(params); - if (cacheBreakpointsUsed >= MAX_CACHE_BREAKPOINTS) return; - let isCCLayout = false; - - if (params.system && Array.isArray(params.system) && params.system.length > 0) { - isCCLayout = params.system[0]?.text?.startsWith(CLAUDE_BILLING_HEADER_PREFIX) === true; - const maxSystemBreakpoints = Math.min(3, MAX_CACHE_BREAKPOINTS - cacheBreakpointsUsed); - cacheBreakpointsUsed += cacheSystemPrefixBreakpoints( - params.system as AnthropicSystemBlock[], - cacheControl, - maxSystemBreakpoints, - isCCLayout ? firstCacheableSystemIndex(params.system as AnthropicSystemBlock[]) : 0, - ); - } - - if (cacheBreakpointsUsed >= MAX_CACHE_BREAKPOINTS) return; - // `convertAnthropicMessages` appends this neutral pad after a trailing // assistant because Anthropic rejects assistant-prefill endings. It is absent - // from the next normal turn, so caching it wastes a scarce breakpoint; anchor - // the cache window on the preceding real assistant instead. + // from the next normal turn, so anchor the rolling window on the preceding + // real assistant instead. const trailingIndex = params.messages.length - 1; const trailingMessage = params.messages[trailingIndex]; const hasTrailingAssistantPad = @@ -3209,160 +3199,20 @@ function applyPromptCaching(params: MessageCreateParamsStreaming, cacheControl?: trailingMessage.content === "Continue." && params.messages[trailingIndex - 1]?.role === "assistant"; const messageEnd = hasTrailingAssistantPad ? trailingIndex - 1 : trailingIndex; - const messageWindowSize = isCCLayout ? 1 : 2; - const start = Math.max(0, messageEnd - messageWindowSize + 1); - for (let i = messageEnd; i >= start; i--) { - if (cacheBreakpointsUsed >= MAX_CACHE_BREAKPOINTS) break; - const message = params.messages[i]; + const start = Math.max(0, messageEnd - 1); + for (let index = messageEnd; index >= start; index--) { + const message = params.messages[index]; if (!message) continue; if (typeof message.content === "string") { message.content = [ { type: "text", text: message.content, cache_control: cloneAnthropicCacheControl(cacheControl) }, ]; - cacheBreakpointsUsed++; - } else if (Array.isArray(message.content) && message.content.length > 0) { - if ( - applyCacheControlToLastTextBlock( - message.content as Array, - cacheControl, - ) - ) { - cacheBreakpointsUsed++; - } + } else if (Array.isArray(message.content)) { + applyCacheControlToLastBlock(message.content, cacheControl); } } } -function normalizeCacheControlBlockTtl(block: CacheControlBlock, seenFiveMinute: { value: boolean }): void { - const cacheControl = block.cache_control; - if (!cacheControl) return; - if (cacheControl.ttl !== "1h") { - seenFiveMinute.value = true; - return; - } - if (seenFiveMinute.value) { - const normalized = cloneAnthropicCacheControl(cacheControl); - delete normalized.ttl; - block.cache_control = normalized; - } -} - -function normalizeCacheControlTtlOrdering(params: MessageCreateParamsStreaming): void { - const seenFiveMinute = { value: false }; - if (params.tools) { - for (const tool of params.tools as Array) { - normalizeCacheControlBlockTtl(tool, seenFiveMinute); - } - } - if (params.system && Array.isArray(params.system)) { - for (const block of params.system as Array) { - normalizeCacheControlBlockTtl(block, seenFiveMinute); - } - } - for (const message of params.messages) { - if (!Array.isArray(message.content)) continue; - for (const block of message.content as Array) { - normalizeCacheControlBlockTtl(block, seenFiveMinute); - } - } -} - -function findLastCacheControlIndex(blocks: T[]): number { - for (let index = blocks.length - 1; index >= 0; index--) { - if (blocks[index]?.cache_control != null) return index; - } - return -1; -} - -function stripCacheControlExceptIndex( - blocks: T[], - preserveIndex: number, - excessCounter: { value: number }, -): void { - for (let index = 0; index < blocks.length && excessCounter.value > 0; index++) { - if (index === preserveIndex) continue; - if (!blocks[index]?.cache_control) continue; - delete blocks[index].cache_control; - excessCounter.value--; - } -} - -function stripAllCacheControl(blocks: T[], excessCounter: { value: number }): void { - for (const block of blocks) { - if (excessCounter.value <= 0) return; - if (!block.cache_control) continue; - delete block.cache_control; - excessCounter.value--; - } -} - -function stripMessageCacheControl( - messages: MessageCreateParamsStreaming["messages"], - excessCounter: { value: number }, -): void { - for (const message of messages) { - if (excessCounter.value <= 0) return; - if (!Array.isArray(message.content)) continue; - for (const block of message.content as Array) { - if (excessCounter.value <= 0) return; - if (!block.cache_control) continue; - delete block.cache_control; - excessCounter.value--; - } - } -} - -function countCacheControlBreakpoints(params: MessageCreateParamsStreaming): number { - let total = 0; - if (params.tools) { - for (const tool of params.tools as Array) { - if (tool.cache_control) total++; - } - } - if (params.system && Array.isArray(params.system)) { - for (const block of params.system as Array) { - if (block.cache_control) total++; - } - } - for (const message of params.messages) { - if (!Array.isArray(message.content)) continue; - for (const block of message.content as Array) { - if (block.cache_control) total++; - } - } - return total; -} - -function enforceCacheControlLimit(params: MessageCreateParamsStreaming, maxBreakpoints: number): void { - const total = countCacheControlBreakpoints(params); - if (total <= maxBreakpoints) return; - const excessCounter = { value: total - maxBreakpoints }; - const systemBlocks = - params.system && Array.isArray(params.system) - ? (params.system as Array) - : []; - const toolBlocks = (params.tools ?? []) as Array; - const lastSystemIndex = findLastCacheControlIndex(systemBlocks); - const lastToolIndex = findLastCacheControlIndex(toolBlocks); - if (systemBlocks.length > 0) { - stripCacheControlExceptIndex(systemBlocks, lastSystemIndex, excessCounter); - } - if (excessCounter.value <= 0) return; - if (toolBlocks.length > 0) { - stripCacheControlExceptIndex(toolBlocks, lastToolIndex, excessCounter); - } - if (excessCounter.value <= 0) return; - stripMessageCacheControl(params.messages, excessCounter); - if (excessCounter.value <= 0) return; - if (systemBlocks.length > 0) { - stripAllCacheControl(systemBlocks, excessCounter); - } - if (excessCounter.value <= 0) return; - if (toolBlocks.length > 0) { - stripAllCacheControl(toolBlocks, excessCounter); - } -} - function usesAdaptiveThinkingTagOnly(model: Model<"anthropic-messages">): boolean { const thinking = model.thinking; if (thinking?.mode !== "anthropic-adaptive") return false; @@ -3445,7 +3295,7 @@ function buildParams( forceDemoteUnsignedThinking && model.compat.replayUnsignedThinking ? { ...model, compat: { ...model.compat, replayUnsignedThinking: false } } : model; - const { cacheControl } = getCacheControl(model, options?.cacheRetention, isOAuthToken); + const { cacheControl } = getCacheControl(model, options?.cacheRetention); // Pre-compute system blocks so they occupy the right slot in the serialized body. const shouldInjectClaudeCodeInstruction = isOAuthToken && !model.id.startsWith("claude-3-5-haiku"); @@ -3582,7 +3432,7 @@ function buildParams( ...(systemBlocks && { system: systemBlocks }), ...(tools !== undefined && { tools }), ...(metadata && { metadata }), - max_tokens: Math.min(maxOutputTokens, options?.maxTokens || modelMaxTokens), + max_tokens: Math.min(maxOutputTokens, options?.maxTokens ?? modelMaxTokens), ...(thinking && { thinking }), ...(contextManagement && { context_management: contextManagement }), ...(outputConfig && { output_config: outputConfig }), @@ -3647,8 +3497,6 @@ function buildParams( disableThinkingIfToolChoiceForced(params, model); ensureMaxTokensForThinking(params, maxOutputTokens); applyPromptCaching(params, cacheControl); - enforceCacheControlLimit(params, 4); - normalizeCacheControlTtlOrdering(params); return params; } diff --git a/packages/ai/src/stream.ts b/packages/ai/src/stream.ts index 8d84baef7..cd022b0db 100644 --- a/packages/ai/src/stream.ts +++ b/packages/ai/src/stream.ts @@ -24,6 +24,7 @@ import { isInvalidatedOAuthTokenError } from "./error/auth-classify"; import { isConcurrencyCapExclusion, isUsageLimitOutcome } from "./error/rate-limit"; import type { BedrockOptions } from "./providers/amazon-bedrock"; import type { AnthropicOptions } from "./providers/anthropic"; +import type { MessageCreateParamsStreaming } from "./providers/anthropic-wire"; import { coworkFetch } from "./providers/cowork-fetch"; import type { CursorOptions } from "./providers/cursor"; import type { DevinOptions } from "./providers/devin"; @@ -69,6 +70,7 @@ import type { FetchImpl, Model, OptionsForApi, + ProviderSessionState, SimpleStreamOptions, StreamOptions, ThinkingBudgets, @@ -1118,10 +1120,296 @@ function emitBufferedEvents(stream: AssistantMessageEventStream, events: Assista } } +const ANTHROPIC_CACHE_TTL_MS = 5 * 60_000; +const ANTHROPIC_CACHE_REFRESH_LEAD_MS = 15_000; +const ANTHROPIC_CACHE_REFRESH_LIMIT = 3; +const ANTHROPIC_CACHE_REFRESH_STATE_KEY = "anthropic-cache-refresh"; + +interface AnthropicCacheRefreshPlan { + refresh(controller: AbortController): Promise; +} + +class AnthropicCacheRefreshState implements ProviderSessionState { + #controller: AbortController | undefined; + #generation = 0; + #plan: AnthropicCacheRefreshPlan | undefined; + #refreshesRemaining = 0; + #timer: NodeJS.Timeout | undefined; + + cancel(): void { + this.#generation++; + if (this.#timer !== undefined) { + clearTimeout(this.#timer); + this.#timer = undefined; + } + this.#controller?.abort(); + this.#controller = undefined; + this.#plan = undefined; + this.#refreshesRemaining = 0; + } + + arm(plan: AnthropicCacheRefreshPlan, cacheTouchedAtMs: number): void { + this.cancel(); + this.#plan = plan; + this.#refreshesRemaining = ANTHROPIC_CACHE_REFRESH_LIMIT; + this.#schedule(cacheTouchedAtMs, this.#generation); + } + + close(): void { + this.cancel(); + } + + #schedule(cacheTouchedAtMs: number, generation: number): void { + const refreshAtMs = cacheTouchedAtMs + ANTHROPIC_CACHE_TTL_MS - ANTHROPIC_CACHE_REFRESH_LEAD_MS; + this.#timer = setTimeout( + () => { + this.#timer = undefined; + void this.#refresh(generation); + }, + Math.max(0, refreshAtMs - Date.now()), + ); + this.#timer.unref?.(); + } + + async #refresh(generation: number): Promise { + const plan = this.#plan; + if (generation !== this.#generation || !plan || this.#refreshesRemaining <= 0) return; + + const controller = new AbortController(); + this.#controller = controller; + let cacheTouchedAtMs: number | undefined; + try { + cacheTouchedAtMs = await plan.refresh(controller); + } catch (error) { + if (generation === this.#generation && !controller.signal.aborted) { + logger.debug("Anthropic prompt-cache refresh failed", { error: String(error) }); + } + } + if (generation !== this.#generation) return; + + this.#controller = undefined; + if (cacheTouchedAtMs === undefined) { + this.#plan = undefined; + this.#refreshesRemaining = 0; + return; + } + + this.#refreshesRemaining--; + if (this.#refreshesRemaining <= 0) { + this.#plan = undefined; + return; + } + this.#schedule(cacheTouchedAtMs, generation); + } +} + +function supportsAnthropicCacheRefresh(model: Model): boolean { + return ( + model.api === "anthropic-messages" && + model.provider === "anthropic" && + model.transport !== "pi-native" && + isLeakedThinkingHealExempt(model) + ); +} + +function isAnthropicRefreshPayload(payload: unknown): payload is MessageCreateParamsStreaming { + return ( + typeof payload === "object" && + payload !== null && + "messages" in payload && + Array.isArray(payload.messages) && + "max_tokens" in payload && + typeof payload.max_tokens === "number" + ); +} + +function isShortAnthropicCacheControl(cacheControl: unknown): boolean { + return ( + typeof cacheControl === "object" && + cacheControl !== null && + "type" in cacheControl && + cacheControl.type === "ephemeral" && + (!("ttl" in cacheControl) || cacheControl.ttl !== "1h") + ); +} + +function hasShortAnthropicMessageBreakpoint(payload: MessageCreateParamsStreaming): boolean { + for (const message of payload.messages) { + if (!Array.isArray(message.content)) continue; + for (const block of message.content) { + if ("cache_control" in block && isShortAnthropicCacheControl(block.cache_control)) return true; + } + } + return false; +} + +function isAnthropicGenerationEvent(event: AssistantMessageEvent): boolean { + switch (event.type) { + case "text_start": + case "thinking_start": + case "toolcall_start": + case "image_end": + return true; + case "text_delta": + case "thinking_delta": + case "toolcall_delta": + return event.delta.length > 0; + default: + return false; + } +} + +function isAnthropicThinkingActive(model: Model, payload: MessageCreateParamsStreaming): boolean { + if (payload.thinking) return payload.thinking.type !== "disabled"; + return model.thinking?.mode === "anthropic-adaptive" && payload.output_config?.effort != null; +} + +function createAnthropicCacheRefreshPlan( + model: Model, + context: Context, + options: SimpleStreamOptions | undefined, + payload: MessageCreateParamsStreaming, +): AnthropicCacheRefreshPlan { + const thinkingEnabled = isAnthropicThinkingActive(model, payload); + return { + async refresh(controller) { + let cacheRead = 0; + let cacheWrite = 0; + let cacheTouchedAtMs: number | undefined; + let canceledAfterGenerationStarted = false; + const response = streamSimpleRequest(model, context, { + ...options, + acceptEmptyResponse: true, + anthropicCacheRefreshRequest: !thinkingEnabled, + cacheRetention: "short", + maxTokens: thinkingEnabled ? options?.maxTokens : 0, + onPayload: () => ({ + ...payload, + max_tokens: thinkingEnabled ? payload.max_tokens : 0, + }), + onResponse: () => { + cacheTouchedAtMs = Date.now(); + }, + onSseEvent: undefined, + signal: controller.signal, + }); + + for await (const event of response) { + if ("partial" in event) { + cacheRead = event.partial.usage.cacheRead; + cacheWrite = event.partial.usage.cacheWrite; + } + if (event.type === "error") return undefined; + if (event.type === "done") { + cacheRead = event.message.usage.cacheRead; + cacheWrite = event.message.usage.cacheWrite; + return cacheTouchedAtMs !== undefined && cacheRead > 0 && cacheWrite === 0 + ? cacheTouchedAtMs + : undefined; + } + if (thinkingEnabled && isAnthropicGenerationEvent(event)) { + canceledAfterGenerationStarted = true; + controller.abort(); + break; + } + } + + if (canceledAfterGenerationStarted) { + try { + await response.result(); + } catch (error) { + if (!controller.signal.aborted) throw error; + } + } + return cacheTouchedAtMs !== undefined && cacheRead > 0 && cacheWrite === 0 ? cacheTouchedAtMs : undefined; + }, + }; +} + +function streamSimpleWithAnthropicCacheRefresh( + model: Model, + context: Context, + options: SimpleStreamOptions | undefined, +): AssistantMessageEventStream { + const providerSessionState = options?.providerSessionState; + if (!options?.anthropicCacheRefresh || !providerSessionState) { + return streamSimpleRequest(model, context, options); + } + + const existingState = providerSessionState.get(ANTHROPIC_CACHE_REFRESH_STATE_KEY); + if (existingState instanceof AnthropicCacheRefreshState) { + existingState.cancel(); + } else if (existingState) { + return streamSimpleRequest(model, context, options); + } + if (!supportsAnthropicCacheRefresh(model) || resolveCacheRetention(options.cacheRetention) !== "short") { + return streamSimpleRequest(model, context, options); + } + + const refreshState = existingState ?? new AnthropicCacheRefreshState(); + if (!existingState) providerSessionState.set(ANTHROPIC_CACHE_REFRESH_STATE_KEY, refreshState); + + let cacheTouchedAtMs: number | undefined; + let capturedPayload: MessageCreateParamsStreaming | undefined; + const inner = streamSimpleRequest(model, context, { + ...options, + onPayload: async (payload, payloadModel) => { + const replacement = await options?.onPayload?.(payload, payloadModel); + const finalPayload = replacement ?? payload; + if (isAnthropicRefreshPayload(finalPayload)) capturedPayload = finalPayload; + return replacement; + }, + onResponse: async (response, responseModel) => { + cacheTouchedAtMs = Date.now(); + await options?.onResponse?.(response, responseModel); + }, + }); + const outer = new AssistantMessageEventStream(); + const armRefresh = (message: AssistantMessage): void => { + if ( + message.stopReason === "error" || + message.stopReason === "aborted" || + message.usage.cacheRead + message.usage.cacheWrite <= 0 || + cacheTouchedAtMs === undefined || + capturedPayload === undefined || + !hasShortAnthropicMessageBreakpoint(capturedPayload) + ) { + return; + } + refreshState.arm(createAnthropicCacheRefreshPlan(model, context, options, capturedPayload), cacheTouchedAtMs); + }; + + void (async () => { + try { + for await (const event of inner) { + if (event.type === "done") armRefresh(event.message); + outer.push(event); + if (outer.done) return; + } + if (!outer.done) { + const result = await inner.result(); + armRefresh(result); + outer.end(result); + } + } catch (error) { + outer.fail(error); + } + })(); + return outer; +} + export function streamSimple( model: Model, context: Context, options?: SimpleStreamOptions, +): AssistantMessageEventStream { + return streamSimpleWithAnthropicCacheRefresh(model, context, options); +} + +function streamSimpleRequest( + model: Model, + context: Context, + options?: SimpleStreamOptions, ): AssistantMessageEventStream { const inputOptions = (options || {}) as SimpleStreamOptions; const baseOptions = { ...inputOptions, fetch: inputOptions.fetch ?? defaultFetchForModel(model) }; @@ -1147,7 +1435,7 @@ export function streamSimple( }; try { - const inner = streamSimple(model, context, { ...requestOptions, apiKey }); + const inner = streamSimpleRequest(model, context, { ...requestOptions, apiKey }); for await (const event of inner) { if (!emittedReplayUnsafeEvent && event.type === "start") { bufferedEvents.push(event); @@ -1613,6 +1901,7 @@ function mapOptionsForApi( fetch: options?.fetch, fallbacks: options?.fallbacks, acceptEmptyResponse: options?.acceptEmptyResponse, + anthropicCacheRefreshRequest: options?.anthropicCacheRefreshRequest, ...simpleProviderOptions, }; diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index 073303089..43ae36965 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -412,6 +412,16 @@ export interface StreamOptions { signal?: AbortSignal; apiKey?: string; cacheRetention?: CacheRetention; + /** + * Keep Anthropic's 5-minute prompt cache warm across bounded idle gaps. + * + * This is an ownership flag, not a general provider default: exactly one + * primary agent loop sharing `providerSessionState` should enable it. + * Side-channel and advisor requests must leave it unset. + */ + anthropicCacheRefresh?: boolean; + /** @internal Marks a replay-only Anthropic request that must use non-streaming `max_tokens: 0`. */ + anthropicCacheRefreshRequest?: boolean; /** * Additional headers to include in provider requests. * These are merged on top of model-defined headers. @@ -559,10 +569,10 @@ export interface StreamOptions { */ providerRetryWait?: (delayMs: number, signal?: AbortSignal) => Promise; /** - * Accept a Google `STOP` response with no visible text or tool call as a - * successful completion. Passive callers such as advisors use this because - * silence is a valid result; interactive agent turns retain empty-response - * retries by default. Ignored by non-Google providers. + * Accept a normal provider stop with no visible text or tool call as a + * successful completion. Passive callers and zero-output cache refreshes use + * this because silence is their expected result; interactive agent turns + * retain empty-response retries by default. */ acceptEmptyResponse?: boolean; /** diff --git a/packages/ai/src/utils/empty-completion-retry.ts b/packages/ai/src/utils/empty-completion-retry.ts index f37a4d498..65c6d8b9d 100644 --- a/packages/ai/src/utils/empty-completion-retry.ts +++ b/packages/ai/src/utils/empty-completion-retry.ts @@ -63,6 +63,7 @@ function isMeaningfulCompletionEvent(event: AssistantMessageEvent): boolean { interface EmptyCompletionRetryOptions { signal?: AbortSignal; providerRetryWait?: (delayMs: number, signal?: AbortSignal) => Promise; + acceptEmptyResponse?: boolean; } /** @@ -82,7 +83,7 @@ export function withEmptyCompletionRetry { for (const event of buffered) outer.push(event); @@ -117,6 +118,7 @@ export function withEmptyCompletionRetry { expect(options.defaultHeaders["anthropic-beta"]).not.toContain("context-1m-2025-08-07"); }); - it("caches the stable prefix and the trailing block while leaving billing + CC identity uncached", () => { - const blocks = buildAnthropicSystemBlocks(["Stay concise."], { - includeClaudeCodeInstruction: true, - extraInstructions: ["Use citations when possible"], - cacheControl: { type: "ephemeral" }, - }); - - expect(blocks).toHaveLength(4); - // OAuth cloak blocks stay uncached: the billing header is a per-request - // fingerprint and the CC identity block mimics Claude Code. - expect(blocks?.[0].text).toStartWith("x-anthropic-billing-header:"); - expect(blocks?.[0].cache_control).toBeUndefined(); - expect(blocks?.[1].text).toBe(claudeCodeSystemInstruction); - expect(blocks?.[1].cache_control).toBeUndefined(); - // Stable-prefix breakpoint on the block before the trailing footer, plus a - // full-match breakpoint on the trailing block itself (#7324). - expect(blocks?.[2]).toEqual({ - type: "text", - text: "Use citations when possible", - cache_control: { type: "ephemeral" }, - }); - expect(blocks?.[3]).toEqual({ - type: "text", - text: "Stay concise.", - cache_control: { type: "ephemeral" }, - }); - }); - - it("keeps the stable-prefix breakpoint when the trailing project footer (cwd/date) changes (#7324)", () => { - const staticInstructions = "STATIC INSTRUCTIONS BLOCK"; - const runA = buildAnthropicSystemBlocks([staticInstructions, "PROJECT\nToday is 2026-08-01, cwd '/tmp/a'."], { - includeClaudeCodeInstruction: true, - cacheControl: { type: "ephemeral" }, - }); - const runB = buildAnthropicSystemBlocks([staticInstructions, "PROJECT\nToday is 2026-08-02, cwd '/tmp/b'."], { - includeClaudeCodeInstruction: true, - cacheControl: { type: "ephemeral" }, - }); - - for (const blocks of [runA, runB]) { - expect(blocks).toHaveLength(4); - expect(blocks?.[0].cache_control).toBeUndefined(); - expect(blocks?.[1].cache_control).toBeUndefined(); - // The static block carries a breakpoint whose prefix excludes the - // volatile footer, so a cwd/date change reuses it instead of - // re-writing the whole system cache. - expect(blocks?.[2].text).toBe(staticInstructions); - expect(blocks?.[2].cache_control).toEqual({ type: "ephemeral" }); - expect(blocks?.[3].cache_control).toEqual({ type: "ephemeral" }); - } - expect(runA?.[2].text).toBe(runB?.[2].text); - }); - - it("caches before the project footer when active-repo context follows it (#7324)", () => { - const staticInstructions = "STATIC INSTRUCTIONS BLOCK"; - const projectFooter = "PROJECT\nToday is 2026-08-01, cwd '/tmp'."; - const activeRepoContext = "The active repository is './repo'."; - const blocks = buildAnthropicSystemBlocks([staticInstructions, projectFooter, activeRepoContext], { - includeClaudeCodeInstruction: true, - cacheControl: { type: "ephemeral" }, - }); - - // blocks: [billing, CC identity, static, project footer, active-repo context] - expect(blocks).toHaveLength(5); - expect(blocks?.[0].cache_control).toBeUndefined(); - expect(blocks?.[1].cache_control).toBeUndefined(); - expect(blocks?.[2]).toEqual({ - type: "text", - text: staticInstructions, - cache_control: { type: "ephemeral" }, - }); - expect(blocks?.[3]).toEqual({ - type: "text", - text: projectFooter, - cache_control: { type: "ephemeral" }, - }); - expect(blocks?.[4]).toEqual({ - type: "text", - text: activeRepoContext, - cache_control: { type: "ephemeral" }, - }); - }); - - it("caches Claude Code context and the last user block in OAuth request payloads", async () => { + it("places a short breakpoint only on the trailing message in a one-message OAuth request", async () => { const payload = (await captureAnthropicPayload(ANTHROPIC_MODEL, { systemPrompt: ["Stay concise."], messages: [{ role: "user", content: "Hi", timestamp: Date.now() }], @@ -355,12 +271,11 @@ describe("Anthropic request fingerprint alignment", () => { expect(payload.system?.[0]?.cache_control).toBeUndefined(); expect(payload.system?.[1]?.text).toBe(claudeCodeSystemInstruction); expect(payload.system?.[1]?.cache_control).toBeUndefined(); - expect(payload.system?.[2]?.cache_control).toEqual({ type: "ephemeral", ttl: "1h" }); + expect(payload.system?.[2]?.cache_control).toBeUndefined(); const content = payload.messages?.[0]?.content; expect(Array.isArray(content)).toBe(true); expect(Array.isArray(content) ? content[0]?.cache_control : undefined).toEqual({ type: "ephemeral", - ttl: "1h", }); }); @@ -380,7 +295,6 @@ describe("Anthropic request fingerprint alignment", () => { const content = payload.messages?.[0]?.content; expect(Array.isArray(content) ? content[0]?.cache_control : undefined).toEqual({ type: "ephemeral", - ttl: "1h", }); }); @@ -416,15 +330,23 @@ describe("Anthropic request fingerprint alignment", () => { timestamp: Date.now(), }, ], - })) as { messages?: Array<{ content?: Array<{ type?: string; cache_control?: unknown }> | string }> }; + })) as { + system?: Array<{ cache_control?: unknown }>; + messages?: Array<{ content?: Array<{ type?: string; cache_control?: unknown }> | string }>; + }; + expect(payload.system?.some(block => block.cache_control != null)).toBe(false); const messages = payload.messages ?? []; - const lastContent = messages[messages.length - 1]?.content; - expect(Array.isArray(lastContent)).toBe(true); - expect(Array.isArray(lastContent) ? lastContent[0]?.type : undefined).toBe("tool_result"); - expect(Array.isArray(lastContent) ? lastContent[0]?.cache_control : undefined).toEqual({ + expect(messages[0]?.content).toBe("Use the tool"); + const assistantContent = messages.at(-2)?.content; + expect(Array.isArray(assistantContent) ? assistantContent.at(-1)?.type : undefined).toBe("tool_use"); + expect(Array.isArray(assistantContent) ? assistantContent.at(-1)?.cache_control : undefined).toEqual({ + type: "ephemeral", + }); + const lastContent = messages.at(-1)?.content; + expect(Array.isArray(lastContent) ? lastContent.at(-1)?.type : undefined).toBe("tool_result"); + expect(Array.isArray(lastContent) ? lastContent.at(-1)?.cache_control : undefined).toEqual({ type: "ephemeral", - ttl: "1h", }); }); @@ -504,7 +426,7 @@ describe("Anthropic request fingerprint alignment", () => { expect(last?.content).toBe("Continue."); }); - it("caches the real assistant before a synthetic Continue pad when the breakpoint budget is tight", async () => { + it("caches the last two real messages and ignores a synthetic Continue pad", async () => { const assistant: AssistantMessage = { role: "assistant", content: [{ type: "text", text: "real assistant answer" }], @@ -534,11 +456,14 @@ describe("Anthropic request fingerprint alignment", () => { messages?: Array<{ role: string; content: string | Array<{ cache_control?: unknown }> }>; }; - expect(payload.system?.filter(block => block.cache_control != null)).toHaveLength(3); + expect(payload.system?.some(block => block.cache_control != null)).toBe(false); + const userContent = payload.messages?.[0]?.content; + expect(Array.isArray(userContent) ? userContent[0]?.cache_control : undefined).toEqual({ + type: "ephemeral", + }); const assistantContent = payload.messages?.find(message => message.role === "assistant")?.content; expect(Array.isArray(assistantContent) ? assistantContent[0]?.cache_control : undefined).toEqual({ type: "ephemeral", - ttl: "1h", }); const pad = payload.messages?.at(-1); expect(pad?.content).toBe("Continue."); @@ -635,7 +560,7 @@ describe("Anthropic request fingerprint alignment", () => { efforts: [Effort.Minimal, Effort.Low, Effort.Medium, Effort.High, Effort.XHigh], }, }); - let capturedParams: MessageCreateParamsStreaming | undefined; + let capturedParams: MessageCreateParams | undefined; let capturedOptions: { headers?: Record } | undefined; await streamAnthropic( adaptiveModel, @@ -673,7 +598,7 @@ describe("Anthropic request fingerprint alignment", () => { expect(capturedOptions?.headers?.["anthropic-beta"] ?? "").toContain("effort-2025-11-24"); }); - it("adds the extended-cache-ttl beta to API-key requests that default to 1h caching", async () => { + it("adds the extended-cache-ttl beta only when 1h caching is requested", async () => { const captureBeta = () => { let captured: string | undefined; const fetchMock = (async (_input: string | URL | Request, init?: RequestInit) => { @@ -690,18 +615,25 @@ describe("Anthropic request fingerprint alignment", () => { messages: [{ role: "user", content: "Hi", timestamp: Date.now() }], }; - const canonical = captureBeta(); + const short = captureBeta(); await streamAnthropic(ANTHROPIC_MODEL, cacheContext, { apiKey: "sk-ant-api-test", - fetch: canonical.fetchMock, + fetch: short.fetchMock, }).result(); - expect(canonical.beta()).toContain("extended-cache-ttl-2025-04-11"); + expect(short.beta()).not.toContain("extended-cache-ttl-2025-04-11"); + + const long = captureBeta(); + await streamAnthropic(ANTHROPIC_MODEL, cacheContext, { + apiKey: "sk-ant-api-test", + cacheRetention: "long", + fetch: long.fetchMock, + }).result(); + expect(long.beta()).toContain("extended-cache-ttl-2025-04-11"); - // Endpoints without long-cache support never send `ttl: "1h"`, so the - // companion beta must stay off the wire too. const proxy = captureBeta(); await streamAnthropic(UMANS_ANTHROPIC_MODEL, cacheContext, { apiKey: "sk-umans-test", + cacheRetention: "long", fetch: proxy.fetchMock, }).result(); expect(proxy.beta()).not.toContain("extended-cache-ttl-2025-04-11"); @@ -832,7 +764,7 @@ describe("Anthropic request fingerprint alignment", () => { expect(extractSuffix(billingWithDev)).toBe(extractSuffix(billingUserOnly)); }); - it("caches the trailing and stable-prefix system blocks on API-key requests (#7324)", async () => { + it("leaves system blocks uncached on API-key requests", async () => { const payload = (await captureAnthropicPayload( ANTHROPIC_MODEL, { @@ -843,10 +775,8 @@ describe("Anthropic request fingerprint alignment", () => { )) as { system?: Array<{ type: string; text?: string; cache_control?: unknown }> }; expect(payload.system).toEqual([ - // Stable-prefix breakpoint: reused when the trailing block changes. - { type: "text", text: "stable system", cache_control: { type: "ephemeral", ttl: "1h" } }, - // Canonical Anthropic API-key requests default to the 1h breakpoint. - { type: "text", text: "stable durable context", cache_control: { type: "ephemeral", ttl: "1h" } }, + { type: "text", text: "stable system" }, + { type: "text", text: "stable durable context" }, ]); }); diff --git a/packages/ai/test/anthropic-cache-refresh.test.ts b/packages/ai/test/anthropic-cache-refresh.test.ts new file mode 100644 index 000000000..1db8475fb --- /dev/null +++ b/packages/ai/test/anthropic-cache-refresh.test.ts @@ -0,0 +1,287 @@ +import { afterEach, describe, expect, it, vi } from "bun:test"; +import { streamSimple } from "@oh-my-pi/pi-ai"; +import type { MessageCreateParams } from "@oh-my-pi/pi-ai/providers/anthropic-wire"; +import type { Context, FetchImpl, Model, ProviderSessionState } from "@oh-my-pi/pi-ai/types"; +import { buildModel } from "@oh-my-pi/pi-catalog/build"; + +const CACHE_REFRESH_DELAY_MS = 5 * 60_000 - 15_000; +const CACHE_TOKENS = 1_200; + +const model: Model<"anthropic-messages"> = buildModel({ + id: "claude-sonnet-4-6", + name: "Claude Sonnet 4.6", + api: "anthropic-messages", + provider: "anthropic", + baseUrl: "https://api.anthropic.com", + reasoning: false, + input: ["text"], + cost: { input: 3, output: 15, cacheRead: 0.3, cacheWrite: 3.75 }, + contextWindow: 200_000, + maxTokens: 8_192, +}); + +const thinkingModel: Model<"anthropic-messages"> = buildModel({ + ...model, + reasoning: true, +}); + +const context: Context = { + messages: [{ role: "user", content: "Keep this prefix warm.", timestamp: 1 }], +}; + +type ResponseMode = "ordinary-write" | "ordinary-roll" | "refresh-read" | "thinking-refresh"; + +interface FetchCapture { + bodies: MessageCreateParams[]; + thinkingRefreshAborted: boolean; +} + +const stateMaps: Array> = []; + +function createProviderSessionState(): Map { + const states = new Map(); + stateMaps.push(states); + return states; +} + +function sseResponse(events: Array>): Response { + const body = `${events.map(event => `event: ${String(event.type)}\ndata: ${JSON.stringify(event)}`).join("\n\n")}\n\n`; + return new Response(body, { + status: 200, + headers: { "Content-Type": "text/event-stream", "request-id": "req_cache_refresh" }, + }); +} + +function usage(cacheRead: number, cacheWrite: number, output: number): Record { + return { + input_tokens: 0, + output_tokens: output, + cache_read_input_tokens: cacheRead, + cache_creation_input_tokens: cacheWrite, + cache_creation: { + ephemeral_5m_input_tokens: cacheWrite, + ephemeral_1h_input_tokens: 0, + }, + }; +} + +function ordinaryResponse(mode: "ordinary-write" | "ordinary-roll"): Response { + const cacheRead = mode === "ordinary-roll" ? CACHE_TOKENS : 0; + return sseResponse([ + { + type: "message_start", + message: { + id: "msg_ordinary", + usage: usage(cacheRead, CACHE_TOKENS, 0), + }, + }, + { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "ok" } }, + { type: "content_block_stop", index: 0 }, + { + type: "message_delta", + delta: { stop_reason: "end_turn" }, + usage: usage(cacheRead, CACHE_TOKENS, 1), + }, + { type: "message_stop" }, + ]); +} + +function refreshResponse(): Response { + return new Response( + JSON.stringify({ + id: "msg_refresh", + type: "message", + role: "assistant", + model: model.id, + content: [], + stop_reason: "end_turn", + usage: usage(CACHE_TOKENS, 0, 0), + }), + { + status: 200, + headers: { "Content-Type": "application/json", "request-id": "req_cache_refresh" }, + }, + ); +} + +function thinkingRefreshResponse(signal: AbortSignal | null | undefined, capture: FetchCapture): Response { + const encoder = new TextEncoder(); + const body = new ReadableStream({ + start(controller) { + const events = [ + { + type: "message_start", + message: { id: "msg_thinking_refresh", usage: usage(CACHE_TOKENS, 0, 0) }, + }, + { + type: "content_block_start", + index: 0, + content_block: { type: "thinking", thinking: "", signature: "" }, + }, + ]; + controller.enqueue( + encoder.encode( + `${events.map(event => `event: ${event.type}\ndata: ${JSON.stringify(event)}`).join("\n\n")}\n\n`, + ), + ); + const closeOnAbort = () => { + capture.thinkingRefreshAborted = true; + controller.close(); + }; + if (signal?.aborted) closeOnAbort(); + else signal?.addEventListener("abort", closeOnAbort, { once: true }); + }, + cancel() { + capture.thinkingRefreshAborted = true; + }, + }); + return new Response(body, { + status: 200, + headers: { "Content-Type": "text/event-stream", "request-id": "req_thinking_refresh" }, + }); +} + +function createFetch(modes: ResponseMode[], capture: FetchCapture): FetchImpl { + return async (input, init) => { + const body: MessageCreateParams = JSON.parse(String(init?.body ?? "{}")); + capture.bodies.push(body); + const mode = modes[capture.bodies.length - 1]; + switch (mode) { + case "ordinary-write": + case "ordinary-roll": + return ordinaryResponse(mode); + case "refresh-read": + return refreshResponse(); + case "thinking-refresh": + return thinkingRefreshResponse(input instanceof Request ? input.signal : init?.signal, capture); + } + }; +} + +interface FinishRequestOptions { + anthropicCacheRefresh?: boolean; + model?: Model<"anthropic-messages">; + sessionId?: string; +} + +async function finishRequest( + fetch: FetchImpl, + providerSessionState: Map, + options: FinishRequestOptions = {}, +): Promise { + const requestModel = options.model ?? model; + const stream = streamSimple(requestModel, context, { + fetch, + anthropicCacheRefresh: options.anthropicCacheRefresh ?? true, + providerSessionState, + sessionId: options.sessionId ?? "cache-refresh-test-session", + }); + for await (const _event of stream) { + // Drain the public response before the idle gap begins. + } + await stream.result(); +} + +async function drainUntil(predicate: () => boolean, message: string): Promise { + for (let attempt = 0; attempt < 1_000; attempt++) { + if (predicate()) return; + await Promise.resolve(); + } + throw new Error(message); +} + +async function advanceToRefresh(capture: FetchCapture, expectedRequests: number): Promise { + vi.advanceTimersByTime(CACHE_REFRESH_DELAY_MS); + await drainUntil( + () => capture.bodies.length >= expectedRequests, + `Expected ${expectedRequests} Anthropic requests, saw ${capture.bodies.length}`, + ); +} + +afterEach(() => { + for (const states of stateMaps.splice(0)) { + for (const state of states.values()) state.close(); + states.clear(); + } + vi.useRealTimers(); + vi.restoreAllMocks(); +}); + +describe("Anthropic prompt-cache refresh", () => { + it("replays max_tokens=0 once per interval and stops after three refreshes", async () => { + vi.useFakeTimers(); + const capture: FetchCapture = { bodies: [], thinkingRefreshAborted: false }; + const fetch = createFetch(["ordinary-write", "refresh-read", "refresh-read", "refresh-read"], capture); + const states = createProviderSessionState(); + + await finishRequest(fetch, states); + for (let requestCount = 2; requestCount <= 4; requestCount++) { + await advanceToRefresh(capture, requestCount); + } + vi.advanceTimersByTime(CACHE_REFRESH_DELAY_MS * 2); + await Promise.resolve(); + + expect(capture.bodies).toHaveLength(4); + for (const refresh of capture.bodies.slice(1)) { + expect(refresh.max_tokens).toBe(0); + expect(refresh.stream).toBe(false); + } + }); + + it("resets the idle gap when another normal request starts", async () => { + vi.useFakeTimers(); + const capture: FetchCapture = { bodies: [], thinkingRefreshAborted: false }; + const fetch = createFetch(["ordinary-write", "ordinary-roll", "refresh-read"], capture); + const states = createProviderSessionState(); + + await finishRequest(fetch, states); + vi.advanceTimersByTime(CACHE_REFRESH_DELAY_MS - 1); + await finishRequest(fetch, states); + vi.advanceTimersByTime(CACHE_REFRESH_DELAY_MS - 1); + await Promise.resolve(); + expect(capture.bodies).toHaveLength(2); + + vi.advanceTimersByTime(1); + await drainUntil(() => capture.bodies.length === 3, "Replacement idle timer did not refresh"); + expect(capture.bodies).toHaveLength(3); + }); + + it("keeps refresh ownership with the main turn when a side request shares provider state", async () => { + vi.useFakeTimers(); + const capture: FetchCapture = { bodies: [], thinkingRefreshAborted: false }; + const fetch = createFetch(["ordinary-write", "ordinary-roll", "refresh-read"], capture); + const states = createProviderSessionState(); + const halfInterval = Math.floor(CACHE_REFRESH_DELAY_MS / 2); + + await finishRequest(fetch, states); + vi.advanceTimersByTime(halfInterval); + await finishRequest(fetch, states, { + anthropicCacheRefresh: false, + sessionId: "cache-refresh-test-session:side:1", + }); + vi.advanceTimersByTime(CACHE_REFRESH_DELAY_MS - halfInterval); + await drainUntil(() => capture.bodies.length === 3, "Main idle timer did not refresh"); + + vi.advanceTimersByTime(halfInterval); + await Promise.resolve(); + expect(capture.bodies).toHaveLength(3); + }); + + it("treats omitted adaptive thinking as active and aborts at the first generated block", async () => { + vi.useFakeTimers(); + const capture: FetchCapture = { bodies: [], thinkingRefreshAborted: false }; + const fetch = createFetch(["ordinary-write", "thinking-refresh"], capture); + const states = createProviderSessionState(); + + await finishRequest(fetch, states, { model: thinkingModel }); + await advanceToRefresh(capture, 2); + await drainUntil(() => capture.thinkingRefreshAborted, "Thinking refresh was not aborted"); + + expect(capture.bodies[1]?.thinking).toBeUndefined(); + expect(capture.bodies[1]?.output_config?.effort).toBe("low"); + expect(capture.bodies[1]?.max_tokens).toBeGreaterThan(0); + expect(capture.bodies[1]?.stream).toBe(true); + expect(capture.thinkingRefreshAborted).toBe(true); + }); +}); diff --git a/packages/ai/test/anthropic-stream-envelope.test.ts b/packages/ai/test/anthropic-stream-envelope.test.ts index 255d5a886..f6c2f9364 100644 --- a/packages/ai/test/anthropic-stream-envelope.test.ts +++ b/packages/ai/test/anthropic-stream-envelope.test.ts @@ -1735,7 +1735,7 @@ describe("anthropic stream envelope handling", () => { expect(cacheControls[2]).toEqual({ type: "ephemeral" }); }); - it("defaults API-key requests to 1h cache TTL where long retention is supported", async () => { + it("defaults Anthropic requests to 5m writes and keeps 1h retention opt-in", async () => { type CapturedParams = { messages: Array<{ content: unknown }> }; const payloads: CapturedParams[] = []; vi.spyOn(AnthropicMessages.prototype, "create").mockImplementation((params: unknown) => { @@ -1758,7 +1758,7 @@ describe("anthropic stream envelope handling", () => { await drain(model); await drain(proxyModel); - await withEnv({ PI_CACHE_RETENTION: "short" }, () => drain(model)); + await withEnv({ PI_CACHE_RETENTION: "long" }, () => drain(model)); const cacheControls = payloads.map(payload => { const content = payload.messages.at(-1)?.content; @@ -1766,13 +1766,8 @@ describe("anthropic stream envelope handling", () => { const lastBlock: { cache_control?: { ttl?: string; type: string } } | undefined = content.at(-1); return lastBlock?.cache_control; }); - // Agent sessions idle past 5 minutes on background jobs; the canonical - // Anthropic API defaults to the 1h breakpoint so resume doesn't cold-miss - // the whole prefix. - expect(cacheControls[0]).toEqual({ type: "ephemeral", ttl: "1h" }); - // Endpoints without long-cache support keep the plain 5m breakpoint. + expect(cacheControls[0]).toEqual({ type: "ephemeral" }); expect(cacheControls[1]).toEqual({ type: "ephemeral" }); - // PI_CACHE_RETENTION=short opts back out of the 1h default. - expect(cacheControls[2]).toEqual({ type: "ephemeral" }); + expect(cacheControls[2]).toEqual({ type: "ephemeral", ttl: "1h" }); }); }); diff --git a/packages/catalog/CHANGELOG.md b/packages/catalog/CHANGELOG.md index 1b2732e40..b745d544c 100644 --- a/packages/catalog/CHANGELOG.md +++ b/packages/catalog/CHANGELOG.md @@ -5,9 +5,11 @@ ### Added - Added first-party OpenAI Daybreak Blue, Daybreak Red, and GPT-5.6 Cyber models with their documented API pricing, token limits, tools, and full `off`/`low`/`medium`/`high`/`xhigh`/`max` reasoning controls. +- Added `calculateUncachedInputCost()` for pricing a prompt against its active context-length tier without prompt caching. ### Fixed +- Fixed Anthropic cache-write pricing to honor mixed 5-minute and 1-hour TTL usage instead of charging every write at the 5-minute rate. - Fixed Ollama Cloud DeepSeek V4 Flash (including dated/preview tags like `deepseek-v4-flash:0731`) exposing the generic `minimal`/`low`/`medium`/`high`/`xhigh` effort ladder without `max`; the `ollama-chat` transport now applies the DeepSeek effort contract (Flash → `low`/`high`/`max`, V4 Pro and older reasoners → `high`/`max`), matching the direct API and every other host ([#8334](https://github.com/can1357/oh-my-pi/issues/8334)). - Bounded OpenAI-compatible model discovery with a default request timeout so a stalled provider `/models` endpoint can no longer hang startup indefinitely in `resolveModelDiscoveryFallback` ([#8315](https://github.com/can1357/oh-my-pi/issues/8315)). - Fixed Codex-discovered `gpt-daybreak-*` aliases being treated as unknown models, restoring the GPT-5.6 `low`/`medium`/`high`/`xhigh`/`max` effort ladder and its 372K fallback only when the Codex registry omits `context_window`. diff --git a/packages/catalog/src/models.ts b/packages/catalog/src/models.ts index dd0f498f7..69b239eec 100644 --- a/packages/catalog/src/models.ts +++ b/packages/catalog/src/models.ts @@ -42,18 +42,23 @@ export function getBundledModels(provider: GeneratedProvider): Model[] { const models = getProviderModels(provider); return models ? (Array.from(models.values()) as Model[]) : []; } -function resolveTokenCost(cost: Model["cost"], usage: Usage): TokenCost { +function resolveTokenCost(cost: Model["cost"], promptInputTokens: number): TokenCost { const longContext = cost.longContext; if (!longContext) return cost; - const orchestration = usage.orchestration; - const promptInputTokens = - usage.input + usage.cacheRead + usage.cacheWrite + (orchestration?.input ?? 0) + (orchestration?.cacheRead ?? 0); return promptInputTokens > longContext.inputThreshold ? longContext : cost; } +/** Price a prompt as fully uncached input under its active context-length tier. */ +export function calculateUncachedInputCost(cost: Model["cost"], promptInputTokens: number): number { + const rates = resolveTokenCost(cost, promptInputTokens); + return (rates.input / 1_000_000) * promptInputTokens; +} + export function calculateCost(model: Model, usage: Usage): Usage["cost"] { - const rates = resolveTokenCost(model.cost, usage); const orchestration = usage.orchestration; + const promptInputTokens = + usage.input + usage.cacheRead + usage.cacheWrite + (orchestration?.input ?? 0) + (orchestration?.cacheRead ?? 0); + const rates = resolveTokenCost(model.cost, promptInputTokens); usage.cost.input = (rates.input / 1000000) * (usage.input + (orchestration?.input ?? 0)); usage.cost.output = (rates.output / 1000000) * (usage.output + (orchestration?.output ?? 0)); usage.cost.cacheRead = (rates.cacheRead / 1000000) * (usage.cacheRead + (orchestration?.cacheRead ?? 0)); @@ -66,11 +71,11 @@ export function calculateCost(model: Model, usage: Usage * Price cache-write tokens, honoring the TTL breakdown when the provider reports one. * * `rates.cacheWrite` is the 5-minute write rate (Anthropic bills 5m writes at - * 1.25x base input). When `usage.cttl` is present the write mixes 5m and 1h - * breakpoints — omp defaults to 1h retention on first-party Anthropic, and 1h writes - * bill at 2x base input — so each component is priced at its own rate instead of the - * flat 5m rate. Deriving 1h from `input * 2` (Anthropic's published multiplier) is - * model-independent and stays correct even for legacy entries whose stored + * 1.25x base input). When `usage.cttl` is present the write can mix 5m and 1h + * breakpoints, and 1h writes bill at 2x base input, so each component is + * priced at its own rate instead of the flat 5m rate. Deriving 1h from + * `input * 2` (Anthropic's published multiplier) is model-independent and + * stays correct even for legacy entries whose stored * `cacheWrite` scalar drifts from 1.25x input. Providers that omit `cttl` * (everyone but Anthropic) keep the flat-rate calculation. * diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 74b13c390..ccb689906 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -6,11 +6,13 @@ - Added Astral `ty` as a built-in Python primary LSP server (`ty server`), ordered behind `pyright`/`basedpyright`/`pylsp` so it becomes the primary Python LSP only when the existing servers are unavailable. `ruff` remains the Python linter and coexists alongside `ty` ([#4617](https://github.com/can1357/oh-my-pi/issues/4617)). - Added first-party Nix support with reproducible source builds for Linux and macOS on x86-64 and ARM64, a pinned development shell, an overlay, NixOS and Home Manager modules, offline Bun dependencies, and lightweight flake evaluation in CI. Nix-managed installs now direct updates back through Nix instead of replacing store-managed executables. +- `omp update` and the startup version check now follow an `omp.rename` pointer in the published npm manifest, preparing existing installs for the upcoming npm package rename. Migration is transactional: the renamed agent/natives packages are installed first (npm uses `--force` to take over the `omp` bin), so an install failure leaves the old install untouched; the old-name globals are removed only afterwards, and a broken bin link is restored by re-running the idempotent install before verification decides the outcome. ### Changed - `/usage`, `omp usage`, and the status line now show authoritative OpenCode Go quota from the official `GET /zen/go/v1/usage` endpoint — including usage made outside OMP — instead of dollar estimates summed from OMP-observed request costs. The status line renders all three windows (`5h` / `7d` / `mo`), and the per-turn cost recording special case for `opencode-go` sessions is gone along with the "OMP-observed spend only" disclaimer ([#8337](https://github.com/can1357/oh-my-pi/pull/8337) by [@will-bogusz](https://github.com/will-bogusz)). - Clarified that the production collab relay source and binaries are not currently published, and documented the source-available local protocol relay ([#8165](https://github.com/can1357/oh-my-pi/issues/8165)). +- Enabled bounded Anthropic prompt-cache refreshes for the main agent loop while keeping advisor and side-channel requests from taking over the shared refresh timer. - Fixed snapcompact compaction shipping its redundant frame archive out of `SessionMaintenance.compact()` on both the manual RPC response (which hard-failed protocol v1 with a transport error after the compaction had already persisted) and the `auto_compaction_end` event payload (which forced the shrink ladder on every unattended pass); the archive is now stripped from both exits while the persisted compaction entry keeps it ([#8168](https://github.com/can1357/oh-my-pi/issues/8168)). - Fixed the edit tool showing no diff preview in `apply_patch` mode: the built-in `edit` tool presents on the wire as `apply_patch`, but the renderer-provenance gate did not resolve that alias to its built-in owner, so the edit renderer was skipped ([#8184](https://github.com/can1357/oh-my-pi/issues/8184)). diff --git a/packages/coding-agent/src/cli/stats-cli.ts b/packages/coding-agent/src/cli/stats-cli.ts index 06d9856b4..fc453afbf 100644 --- a/packages/coding-agent/src/cli/stats-cli.ts +++ b/packages/coding-agent/src/cli/stats-cli.ts @@ -170,6 +170,7 @@ async function printStatsSummary(): Promise { console.log(` Input Tokens: ${formatNumber(overall.totalInputTokens)}`); console.log(` Output Tokens: ${formatNumber(overall.totalOutputTokens)}`); console.log(` Cache Rate: ${formatPercent(overall.cacheRate)}`); + console.log(` Cache Savings: ${formatPercent(overall.cacheSavings)}`); console.log(` Total Cost: ${formatCost(overall.totalCost)}`); console.log(` Premium Requests: ${formatNumber(normalizePremiumRequests(overall.totalPremiumRequests ?? 0))}`); console.log(` Avg Duration: ${overall.avgDuration !== null ? formatDuration(overall.avgDuration) : "-"}`); @@ -182,7 +183,7 @@ async function printStatsSummary(): Promise { console.log(chalk.bold("\nBy Model:")); for (const m of byModel.slice(0, 10)) { console.log( - ` ${m.model}: ${formatNumber(m.totalRequests)} reqs, ${formatCost(m.totalCost)}, ${formatPercent(m.cacheRate)} cache`, + ` ${m.model}: ${formatNumber(m.totalRequests)} reqs, ${formatCost(m.totalCost)}, ${formatPercent(m.cacheRate)} cache rate, ${formatPercent(m.cacheSavings)} cache savings`, ); } } diff --git a/packages/coding-agent/src/cli/update-cli.ts b/packages/coding-agent/src/cli/update-cli.ts index ca56d17ce..af5ba846c 100644 --- a/packages/coding-agent/src/cli/update-cli.ts +++ b/packages/coding-agent/src/cli/update-cli.ts @@ -67,11 +67,27 @@ function currentNativeTag(): string { /** Distribution channel advertised by a release's published npm manifest. */ export type ReleaseDist = "npm" | "binary"; -interface ReleaseInfo { +/** npm package names a release installs: the agent package and its natives companion. */ +export interface ReleasePackages { + pkg: string; + natives: string; +} + +/** Parsed `omp.rename` pointer: the new agent package name and optional new natives name. */ +export interface ReleaseRename { + pkg: string; + natives?: string; +} + +const CURRENT_PACKAGES: ReleasePackages = { pkg: PACKAGE, natives: NATIVES_PACKAGE }; + +export interface ReleaseInfo { tag: string; version: string; /** Parsed `omp.dist` from the registry manifest; undefined when absent. */ dist?: ReleaseDist; + /** npm names to install, resolved after following any `omp.rename` pointers. */ + packages: ReleasePackages; } export interface ReleaseBinaryAsset { @@ -103,6 +119,34 @@ export function resolveReleaseDist(manifest: unknown): ReleaseDist | undefined { return dist === "npm" ? "npm" : "binary"; } +/** + * Parse the `omp.rename` pointer from a published package manifest. + * + * Forward-compatibility contract for renaming the npm package: the final + * version published under an old name is a stub whose manifest carries + * `"omp": { "rename": { "package": "", "natives": "" }, "dist": "binary" }`. + * Updaters that understand `rename` follow the pointer and resolve the + * release from the renamed package instead ({@link getLatestRelease}); + * older deployed updaters ignore it and take the `dist: "binary"` escape + * hatch, replacing the install with the GitHub release binary rather than + * installing the stub via bun/npm. + * + * The renamed package's own manifest MUST declare `"dist": "npm"` (so + * package-manager installs stay package-managed across a major bump) and + * MUST continue the old version line (a version reset would compare as + * "already up to date" against the running build). + */ +export function resolveReleaseRename(manifest: unknown): ReleaseRename | undefined { + if (!isRecord(manifest) || !isRecord(manifest.omp)) return undefined; + const rename = manifest.omp.rename; + if (!isRecord(rename) || typeof rename.package !== "string" || rename.package.length === 0) return undefined; + const natives = rename.natives; + return { + pkg: rename.package, + natives: typeof natives === "string" && natives.length > 0 ? natives : undefined, + }; +} + function majorVersion(version: string): number { const major = Number.parseInt(version, 10); return Number.isNaN(major) ? 0 : major; @@ -529,36 +573,63 @@ async function resolveUpdateTarget(options: { allowPackageManagers: boolean }): throw new Error(`Could not resolve ${APP_NAME} binary path in PATH`); } -/** - * Get the latest release info from the npm registry. - * Uses npm instead of GitHub API to avoid unauthenticated rate limiting. - */ -async function getLatestRelease(): Promise { +/** Bound on `omp.rename` hops so a broken pointer chain cannot loop forever. */ +const MAX_RENAME_HOPS = 3; + +async function fetchLatestManifest( + pkg: string, + timeoutMs: number, +): Promise<{ version: string; manifest: Record }> { let response: Response; try { - response = await fetch(`${NPM_REGISTRY}${PACKAGE}/latest`, { - signal: withTimeoutSignal(RELEASE_METADATA_TIMEOUT_MS), + response = await fetch(`${NPM_REGISTRY}${pkg}/latest`, { + signal: withTimeoutSignal(timeoutMs), }); } catch (err) { if (isTimeoutError(err)) { - throw new Error("Timed out fetching release info after 30s", { cause: err }); + throw new Error(`Timed out fetching release info for ${pkg} after ${Math.round(timeoutMs / 1000)}s`, { + cause: err, + }); } throw err; } if (!response.ok) { - throw new Error(`Failed to fetch release info: ${response.statusText}`); + throw new Error(`Failed to fetch release info for ${pkg}: ${response.statusText}`); } const data: unknown = await response.json(); if (!isRecord(data) || typeof data.version !== "string") { - throw new Error("Malformed npm registry response: missing version"); + throw new Error(`Malformed npm registry response for ${pkg}: missing version`); + } + return { version: data.version, manifest: data }; +} + +/** + * Get the latest release info from the npm registry, following `omp.rename` + * pointers ({@link resolveReleaseRename}) when the package has moved to a new + * npm name. Version, dist, and install names all come from the final manifest + * in the chain. Uses npm instead of GitHub API to avoid unauthenticated rate + * limiting. + */ +export async function getLatestRelease(options: { timeoutMs?: number } = {}): Promise { + const timeoutMs = options.timeoutMs ?? RELEASE_METADATA_TIMEOUT_MS; + const packages: ReleasePackages = { ...CURRENT_PACKAGES }; + const visited = new Set([packages.pkg]); + let latest = await fetchLatestManifest(packages.pkg, timeoutMs); + for (let hop = 0; hop < MAX_RENAME_HOPS; hop++) { + const rename = resolveReleaseRename(latest.manifest); + if (!rename || visited.has(rename.pkg)) break; + visited.add(rename.pkg); + packages.pkg = rename.pkg; + if (rename.natives) packages.natives = rename.natives; + latest = await fetchLatestManifest(packages.pkg, timeoutMs); } - const version = data.version; return { - tag: `v${version}`, - version, - dist: resolveReleaseDist(data), + tag: `v${latest.version}`, + version: latest.version, + dist: resolveReleaseDist(latest.manifest), + packages, }; } @@ -994,10 +1065,14 @@ export async function replaceBinaryForUpdate(options: BinaryReplacementOptions): } } -function buildVersionedPackageInstallArgs(expectedVersion: string, nativeTag: string): string[] { - const args = [`${PACKAGE}@${expectedVersion}`, `${NATIVES_PACKAGE}@${expectedVersion}`]; +function buildVersionedPackageInstallArgs( + expectedVersion: string, + nativeTag: string, + packages: ReleasePackages, +): string[] { + const args = [`${packages.pkg}@${expectedVersion}`, `${packages.natives}@${expectedVersion}`]; if (SUPPORTED_NATIVE_TAGS.has(nativeTag)) { - args.push(`${NATIVES_PACKAGE}-${nativeTag}@${expectedVersion}`); + args.push(`${packages.natives}-${nativeTag}@${expectedVersion}`); } return args; } @@ -1032,25 +1107,41 @@ function buildVersionedPackageInstallArgs(expectedVersion: string, nativeTag: st * the original "no matching version" message instead of `EBADPLATFORM`. * See #1824. */ -export function buildBunInstallArgs(expectedVersion: string, nativeTag: string = currentNativeTag()): string[] { +export function buildBunInstallArgs( + expectedVersion: string, + nativeTag: string = currentNativeTag(), + packages: ReleasePackages = CURRENT_PACKAGES, +): string[] { return [ "install", "-g", "--no-cache", `--registry=${NPM_REGISTRY}`, - ...buildVersionedPackageInstallArgs(expectedVersion, nativeTag), + ...buildVersionedPackageInstallArgs(expectedVersion, nativeTag, packages), ]; } -/** Build the npm argv used to update npm-managed global installs. */ -export function buildNpmInstallArgs(expectedVersion: string, nativeTag: string = currentNativeTag()): string[] { - const args = [ +/** + * Build the npm argv used to update npm-managed global installs. + * + * `force` is set only for rename migrations: npm refuses to write the `omp` + * bin while the old package still owns it (`EEXIST`), and the migration + * installs the new package BEFORE removing the old one so a failed install + * never leaves the user without a working `omp`. + */ +export function buildNpmInstallArgs( + expectedVersion: string, + nativeTag: string = currentNativeTag(), + packages: ReleasePackages = CURRENT_PACKAGES, + flags: { force?: boolean } = {}, +): string[] { + return [ "install", "-g", + ...(flags.force ? ["--force"] : []), `--registry=${NPM_REGISTRY}`, - ...buildVersionedPackageInstallArgs(expectedVersion, nativeTag), + ...buildVersionedPackageInstallArgs(expectedVersion, nativeTag, packages), ]; - return args; } export function buildHomebrewUpdateArgs(force: boolean): string[] { @@ -1066,17 +1157,122 @@ export function buildMiseForceInstallArgs(expectedVersion: string): string[] { } /** - * Update via package manager. + * Old-name globals a rename migration removes after the new install exists: + * the set difference between the old install's top-level globals + * ({@link buildVersionedPackageInstallArgs} installs the agent, natives core, + * and platform leaf explicitly) and the resolved install's. An agent-only + * rename keeps the natives names, and removing them would strip the addon + * the new install just pinned. */ -async function updateViaBun(expectedVersion: string): Promise { - console.log(chalk.dim("Updating via bun...")); - const args = buildBunInstallArgs(expectedVersion); - const result = await $`bun ${args}`.nothrow(); - if (result.exitCode !== 0) { - throw new Error(`bun install failed with exit code ${result.exitCode}`); +export function buildRenameCleanupPackages( + packages: ReleasePackages, + nativeTag: string = currentNativeTag(), +): string[] { + const old = [PACKAGE, NATIVES_PACKAGE]; + if (SUPPORTED_NATIVE_TAGS.has(nativeTag)) { + old.push(`${NATIVES_PACKAGE}-${nativeTag}`); + } + const newLeaf = `${packages.natives}-${nativeTag}`; + return old.filter(name => name !== packages.pkg && name !== packages.natives && name !== newLeaf); +} + +/** Injectable shell steps for {@link migrateRenamedInstall}; commands return process exit codes. */ +export interface RenameMigrationSteps { + /** Globally install the new package names. MUST be idempotent: re-running re-links the `omp` bin. */ + install(): Promise; + /** Remove the old-name globals. */ + removeOld(): Promise; + /** Check the PATH-resolved `omp` against the expected version. */ + verify(): Promise; +} + +/** Production {@link RenameMigrationSteps}: bun/npm global installs plus PATH verification. */ +function packageManagerMigrationSteps(manager: "bun" | "npm", release: ReleaseInfo): RenameMigrationSteps { + const nativeTag = currentNativeTag(); + return { + async install() { + if (manager === "bun") { + const args = buildBunInstallArgs(release.version, nativeTag, release.packages); + return (await $`bun ${args}`.nothrow()).exitCode; + } + const args = buildNpmInstallArgs(release.version, nativeTag, release.packages, { force: true }); + return (await $`npm ${args}`.nothrow()).exitCode; + }, + async removeOld() { + // One invocation per package: a single batched remove fails wholesale + // when any name is absent (e.g. the platform leaf on an old install), + // which would skip the agent package that actually owns the bin. + let agentExit = 0; + for (const pkg of buildRenameCleanupPackages(release.packages, nativeTag)) { + const result = + manager === "bun" + ? await $`bun remove -g ${pkg}`.quiet().nothrow() + : await $`npm uninstall -g ${pkg}`.quiet().nothrow(); + if (pkg === PACKAGE) agentExit = result.exitCode; + } + return agentExit; + }, + verify: () => verifyInstalledVersion(release.version), + }; +} + +/** + * Migrate a package-manager install across an `omp.rename` hop without a + * window where no working `omp` exists: + * + * 1. Install the new package FIRST. Nothing has been removed yet, so a + * failure here leaves the old install fully functional. + * 2. Remove the old-name globals. Failure is non-fatal: a stale package + * wastes disk, but the bin already points at the new install. + * 3. Verify the PATH-resolved `omp`. If the removal deleted the shared bin + * link (manager-dependent), re-run the idempotent install to restore it + * and verify again; only a repeated failure aborts, with a recovery hint. + */ +export async function migrateRenamedInstall(release: ReleaseInfo, steps: RenameMigrationSteps): Promise { + console.log(chalk.dim(`npm package renamed to ${release.packages.pkg}; migrating this install.`)); + const installExit = await steps.install(); + if (installExit !== 0) { + throw new Error( + `install of ${release.packages.pkg} failed with exit code ${installExit}; the existing install was left untouched`, + ); } - await printVerification(expectedVersion); + const removeExit = await steps.removeOld(); + if (removeExit !== 0) { + console.log(chalk.yellow(`Warning: could not remove the old ${PACKAGE} package; remove it manually later.`)); + } + + let verification = await steps.verify(); + if (!verification.ok) { + // Removing the old package may have taken the shared bin link with it; + // reinstalling the new package restores the link. + if ((await steps.install()) === 0) { + verification = await steps.verify(); + } + } + if (!verification.ok) { + throw new Error( + `${formatVerificationFailure(verification, release.version)}; reinstall with: curl -fsSL https://omp.sh/install | sh`, + ); + } + printVerifiedVersion(release.version); +} + +/** + * Update via package manager. + */ +async function updateViaBun(release: ReleaseInfo): Promise { + console.log(chalk.dim("Updating via bun...")); + if (release.packages.pkg !== PACKAGE) { + await migrateRenamedInstall(release, packageManagerMigrationSteps("bun", release)); + } else { + const args = buildBunInstallArgs(release.version, currentNativeTag(), release.packages); + const result = await $`bun ${args}`.nothrow(); + if (result.exitCode !== 0) { + throw new Error(`bun install failed with exit code ${result.exitCode}`); + } + await printVerification(release.version); + } try { const pruneResult = await pruneBunCacheAfterGlobalInstall(); if (pruneResult && pruneResult.removedEntries > 0) { @@ -1087,15 +1283,19 @@ async function updateViaBun(expectedVersion: string): Promise { } } -async function updateViaNpm(expectedVersion: string): Promise { +async function updateViaNpm(release: ReleaseInfo): Promise { console.log(chalk.dim("Updating via npm...")); - const args = buildNpmInstallArgs(expectedVersion); + if (release.packages.pkg !== PACKAGE) { + await migrateRenamedInstall(release, packageManagerMigrationSteps("npm", release)); + return; + } + const args = buildNpmInstallArgs(release.version, currentNativeTag(), release.packages); const result = await $`npm ${args}`.nothrow(); if (result.exitCode !== 0) { throw new Error(`npm install failed with exit code ${result.exitCode}`); } - await printVerification(expectedVersion); + await printVerification(release.version); } async function updateViaHomebrew(expectedVersion: string, force: boolean): Promise { @@ -1345,6 +1545,9 @@ export async function runUpdateCommand(opts: { force: boolean; check: boolean }) } else { console.log(chalk.yellow(`Forcing reinstall of ${release.version}`)); } + if (release.packages.pkg !== PACKAGE) { + console.log(chalk.cyan(`The npm package moved to ${release.packages.pkg}; updating migrates this install.`)); + } if (opts.check) { // Just check, don't install @@ -1379,9 +1582,9 @@ export async function runUpdateCommand(opts: { force: boolean; check: boolean }) ), ); } else if (target.method === "bun") { - await updateViaBun(release.version); + await updateViaBun(release); } else { - await updateViaNpm(release.version); + await updateViaNpm(release); } } else { if (forceBinary && target.replacesSymlink) { diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index 139a7d1c7..240dab98d 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -29,6 +29,7 @@ import { processFileArguments } from "./cli/file-processor"; import { buildInitialMessage } from "./cli/initial-message"; import { selectSession } from "./cli/session-picker"; import { applyStartupCwd } from "./cli/startup-cwd"; +import { getLatestRelease } from "./cli/update-cli"; import { findConfigFile } from "./config"; import { ModelRegistry } from "./config/model-registry"; import { @@ -95,7 +96,6 @@ import { concreteThinkingLevel, parseConfiguredThinkingLevel } from "./thinking" import type { LspStartupServerInfo } from "./tools"; import { getChangelogPath, resolveStartupChangelogForDisplay, type StartupChangelogSelection } from "./utils/changelog"; import { EventBus } from "./utils/event-bus"; -import { withTimeoutSignal } from "./utils/fetch-timeout"; type RunAcpMode = (createSession: AcpSessionFactory) => Promise; type RunPrintMode = (session: AgentSession, options: PrintModeOptions) => Promise; @@ -115,19 +115,8 @@ async function checkForNewVersion(currentVersion: string): Promise 0) { - return latestVersion; - } - - return undefined; + const release = await getLatestRelease({ timeoutMs: 5_000 }); + return Bun.semver.order(release.version, currentVersion) > 0 ? release.version : undefined; } catch { return undefined; } diff --git a/packages/coding-agent/src/sdk.ts b/packages/coding-agent/src/sdk.ts index 6d1f16b8f..1f487efaf 100644 --- a/packages/coding-agent/src/sdk.ts +++ b/packages/coding-agent/src/sdk.ts @@ -3247,6 +3247,7 @@ async function createAgentSessionScoped(options: CreateAgentSessionOptions): Pro supportsExternalThinking(streamModel); return settingsAwareStreamFn(streamModel, context, { ...streamOptions, + anthropicCacheRefresh: true, forceReasoningOff: externalThinking || streamOptions?.forceReasoningOff, }); }, diff --git a/packages/coding-agent/src/web/search/providers/anthropic.ts b/packages/coding-agent/src/web/search/providers/anthropic.ts index ebf84d7c7..6f4012f12 100644 --- a/packages/coding-agent/src/web/search/providers/anthropic.ts +++ b/packages/coding-agent/src/web/search/providers/anthropic.ts @@ -130,7 +130,6 @@ function buildSystemBlocks( return buildAnthropicSystemBlocks(systemPrompt ? [systemPrompt] : undefined, { includeClaudeCodeInstruction: includeClaudeCode, extraInstructions, - cacheControl: { type: "ephemeral" }, }); } diff --git a/packages/coding-agent/test/cli/update-rename-migration.integration.test.ts b/packages/coding-agent/test/cli/update-rename-migration.integration.test.ts new file mode 100644 index 000000000..1e6f4a8a1 --- /dev/null +++ b/packages/coding-agent/test/cli/update-rename-migration.integration.test.ts @@ -0,0 +1,155 @@ +/** + * Real package-manager seam for `omp.rename` migrations. + * + * The unit tests in test/update-cli.test.ts prove the orchestration order of + * migrateRenamedInstall with injected steps; these fixtures prove the two + * empirical assumptions that orchestration stands on, against the actual + * package managers in isolated temp prefixes: + * + * - npm refuses to overwrite a bin owned by another package (EEXIST) and + * `--force` takes ownership of it — and `npm uninstall -g ` deletes + * the shared bin even though it points at the new package, so the repair + * reinstall inside migrateRenamedInstall is the NORMAL npm path, not an + * edge case. + * - bun clobbers the bin on install without force, and `bun remove -g ` + * re-links the bin to the surviving package. + * + * Each scenario runs the full install-new/remove-old/verify transaction and + * asserts the resulting launcher executes the NEW version. + */ +import { afterEach, beforeAll, describe, expect, it, vi } from "bun:test"; +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import { $which, removeWithRetries } from "@oh-my-pi/pi-utils"; +import { $ } from "bun"; +import { + type InstalledVersionVerification, + migrateRenamedInstall, + type ReleaseInfo, + type RenameMigrationSteps, +} from "../../src/cli/update-cli"; +import { initTheme } from "../../src/modes/theme/theme"; + +const OLD_PKG = "omp-rename-fixture-old"; +const NEW_PKG = "omp-rename-fixture-new"; +const OLD_VERSION = "1.0.0"; +const NEW_VERSION = "2.0.0"; + +const tempDirs: string[] = []; + +// printVerifiedVersion renders theme glyphs; the update command initializes +// the theme before calling into update-cli, so the tests must too. +beforeAll(async () => { + await initTheme(); +}); + +afterEach(async () => { + vi.restoreAllMocks(); + await Promise.all(tempDirs.splice(0).map(dir => removeWithRetries(dir))); +}); + +/** Two installable packages that both expose an `omp` bin, plus an empty prefix. */ +async function makeFixtures(): Promise<{ root: string; oldDir: string; newDir: string }> { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "omp-rename-itest-")); + tempDirs.push(root); + const mkpkg = async (name: string, version: string): Promise => { + const dir = path.join(root, name); + await Bun.write(path.join(dir, "package.json"), JSON.stringify({ name, version, bin: { omp: "cli.js" } })); + const cli = path.join(dir, "cli.js"); + await Bun.write(cli, `#!/usr/bin/env bun\nconsole.log("omp/${version}");\n`); + await fs.chmod(cli, 0o755); + return dir; + }; + return { root, oldDir: await mkpkg(OLD_PKG, OLD_VERSION), newDir: await mkpkg(NEW_PKG, NEW_VERSION) }; +} + +/** Run the installed launcher and parse its reported version, mirroring verifyBinaryAtPath. */ +async function verifyLauncher(binDir: string, expectedVersion: string): Promise { + const launcher = path.join(binDir, "omp"); + const result = await $`${launcher}`.quiet().nothrow(); + if (result.exitCode !== 0) return { ok: false, path: launcher }; + const actual = result.text().match(/\/(\d+\.\d+\.\d+)/)?.[1]; + return { ok: actual === expectedVersion, actual, path: launcher }; +} + +const RELEASE: ReleaseInfo = { + tag: `v${NEW_VERSION}`, + version: NEW_VERSION, + packages: { pkg: NEW_PKG, natives: "@oh-my-pi/pi-natives" }, +}; + +describe.skipIf(process.platform === "win32" || !$which("npm"))("rename migration over real npm", () => { + it("takes bin ownership with --force, survives the uninstall deleting the bin, and lands on the new version", async () => { + vi.spyOn(console, "log").mockImplementation(() => {}); + const { root, oldDir, newDir } = await makeFixtures(); + const prefix = path.join(root, "npm-prefix"); + const binDir = path.join(prefix, "bin"); + const env = { + ...process.env, + npm_config_cache: path.join(root, "npm-cache"), + npm_config_update_notifier: "false", + npm_config_fund: "false", + npm_config_audit: "false", + }; + + const seed = await $`npm install -g --prefix ${prefix} ${oldDir}`.env(env).quiet().nothrow(); + expect(seed.exitCode).toBe(0); + expect(await verifyLauncher(binDir, OLD_VERSION)).toMatchObject({ ok: true, actual: OLD_VERSION }); + + // The load-bearing precondition for --force: while the old package owns + // the bin, a plain install of the new package fails instead of clobbering. + const plain = await $`npm install -g --prefix ${prefix} ${newDir}`.env(env).quiet().nothrow(); + expect(plain.exitCode).not.toBe(0); + expect(await verifyLauncher(binDir, OLD_VERSION)).toMatchObject({ ok: true, actual: OLD_VERSION }); + + const steps: RenameMigrationSteps = { + async install() { + return (await $`npm install -g --force --prefix ${prefix} ${newDir}`.env(env).quiet().nothrow()).exitCode; + }, + async removeOld() { + return (await $`npm uninstall -g --prefix ${prefix} ${OLD_PKG}`.env(env).quiet().nothrow()).exitCode; + }, + verify: () => verifyLauncher(binDir, NEW_VERSION), + }; + await migrateRenamedInstall(RELEASE, steps); + + expect(await verifyLauncher(binDir, NEW_VERSION)).toMatchObject({ ok: true, actual: NEW_VERSION }); + const globalPackages = await fs.readdir(path.join(prefix, "lib", "node_modules")); + expect(globalPackages).toContain(NEW_PKG); + expect(globalPackages).not.toContain(OLD_PKG); + }, 120_000); +}); + +describe.skipIf(process.platform === "win32")("rename migration over real bun", () => { + it("clobbers the old bin on install, survives removing the old package, and lands on the new version", async () => { + vi.spyOn(console, "log").mockImplementation(() => {}); + const { root, oldDir, newDir } = await makeFixtures(); + const binDir = path.join(root, "bun-bin"); + await fs.mkdir(binDir, { recursive: true }); + const env = { + ...process.env, + BUN_INSTALL_GLOBAL_DIR: path.join(root, "bun-global"), + BUN_INSTALL_BIN: binDir, + }; + + const seed = await $`bun add -g file:${oldDir}`.env(env).quiet().nothrow(); + expect(seed.exitCode).toBe(0); + expect(await verifyLauncher(binDir, OLD_VERSION)).toMatchObject({ ok: true, actual: OLD_VERSION }); + + const steps: RenameMigrationSteps = { + async install() { + return (await $`bun add -g file:${newDir}`.env(env).quiet().nothrow()).exitCode; + }, + async removeOld() { + return (await $`bun remove -g ${OLD_PKG}`.env(env).quiet().nothrow()).exitCode; + }, + verify: () => verifyLauncher(binDir, NEW_VERSION), + }; + await migrateRenamedInstall(RELEASE, steps); + + expect(await verifyLauncher(binDir, NEW_VERSION)).toMatchObject({ ok: true, actual: NEW_VERSION }); + const globalManifest = await Bun.file(path.join(root, "bun-global", "package.json")).json(); + expect(Object.keys(globalManifest.dependencies ?? {})).toEqual([NEW_PKG]); + }, 120_000); +}); diff --git a/packages/coding-agent/test/update-cli.test.ts b/packages/coding-agent/test/update-cli.test.ts index dc414db8f..608071d8b 100644 --- a/packages/coding-agent/test/update-cli.test.ts +++ b/packages/coding-agent/test/update-cli.test.ts @@ -291,13 +291,23 @@ describe("update-cli npm rename contract", () => { expect(buildNpmInstallArgs("16.3.15", "win32-x64")).not.toContain("--force"); }); - it("removes the old agent package and its natives companions, including the supported platform leaf", () => { - expect(buildRenameCleanupPackages("darwin-arm64")).toEqual([ + it("removes the old agent package and its natives companions when both names moved", () => { + const packages = { pkg: "@new/omp", natives: "@new/natives" }; + expect(buildRenameCleanupPackages(packages, "darwin-arm64")).toEqual([ "@oh-my-pi/pi-coding-agent", "@oh-my-pi/pi-natives", "@oh-my-pi/pi-natives-darwin-arm64", ]); - expect(buildRenameCleanupPackages("linux-arm")).toEqual(["@oh-my-pi/pi-coding-agent", "@oh-my-pi/pi-natives"]); + expect(buildRenameCleanupPackages(packages, "linux-arm")).toEqual([ + "@oh-my-pi/pi-coding-agent", + "@oh-my-pi/pi-natives", + ]); + }); + + it("keeps the natives packages on an agent-only rename so cleanup cannot strip the addon the new install pinned", () => { + const packages = { pkg: "@new/omp", natives: "@oh-my-pi/pi-natives" }; + expect(buildRenameCleanupPackages(packages, "darwin-arm64")).toEqual(["@oh-my-pi/pi-coding-agent"]); + expect(buildRenameCleanupPackages(packages, "linux-arm")).toEqual(["@oh-my-pi/pi-coding-agent"]); }); }); diff --git a/packages/natives/scripts/build-bindings.ts b/packages/natives/scripts/build-bindings.ts index b480e6440..2c0182294 100644 --- a/packages/natives/scripts/build-bindings.ts +++ b/packages/natives/scripts/build-bindings.ts @@ -11,7 +11,7 @@ import * as fs from "node:fs/promises"; import { createRequire } from "node:module"; import * as path from "node:path"; import { $ } from "bun"; -import { detectHostAvx2Support } from "../../../scripts/host-detect"; +import { detectHostAvx2Support, resolveLocalHostAddon } from "../../../scripts/host-detect"; import { generateEnumExports } from "./gen-enums"; // pcre2-sys prefers a system libpcre2 when pkg-config finds one. Keep the @@ -61,10 +61,12 @@ const rustDir = path.join(repoRoot, "crates/pi-natives"); const nativeDir = path.join(import.meta.dir, "../native"); const packageJsonPath = path.join(import.meta.dir, "../package.json"); -type X64Variant = "modern" | "baseline"; - -const effectiveVariant: X64Variant | null = - process.arch === "x64" ? (detectHostAvx2Support() ? "modern" : "baseline") : null; +const localAddon = resolveLocalHostAddon({ + platform: process.platform, + arch: process.arch, + avx2: detectHostAvx2Support(), +}); +const effectiveVariant = localAddon.x64Variant; const variantSuffix = effectiveVariant ? `-${effectiveVariant}` : ""; // Pin Rust target-cpu so x64 baseline/modern variants get a reproducible ISA floor @@ -166,7 +168,7 @@ async function installGeneratedBindings(outputDir: string): Promise { } } -const canonicalAddonFilename = `pi_natives.${process.platform}-${process.arch}${variantSuffix}.node`; +const canonicalAddonFilename = localAddon.filename; const canonicalAddonPath = path.join(nativeDir, canonicalAddonFilename); console.log(`Building pi-natives bindings for ${process.platform}-${process.arch}${variantSuffix} (local)…`); diff --git a/packages/stats/CHANGELOG.md b/packages/stats/CHANGELOG.md index 171945d60..f53d0bf2d 100644 --- a/packages/stats/CHANGELOG.md +++ b/packages/stats/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Added + +- Added cost-weighted `cacheSavings` alongside `cacheRate`, including cache-read discounts and 5-minute/1-hour write premiums against the equivalent uncached prompt cost. + ### Fixed - Made the embedded dashboard archive byte-reproducible by sorting entries and zeroing tar and gzip timestamps before it is compiled into OMP. diff --git a/packages/stats/README.md b/packages/stats/README.md index c1d252ef6..ef7cc411e 100644 --- a/packages/stats/README.md +++ b/packages/stats/README.md @@ -15,6 +15,7 @@ Local observability dashboard for AI usage statistics. |--------|-------------| | Tokens/s | `output_tokens / (duration / 1000)` | | Cache Rate | `cache_read / (input + cache_read) * 100` | +| Cache Savings | `(uncached prompt cost - actual prompt cost) / uncached prompt cost * 100` | | Error Rate | `count(stopReason=error) / total_calls * 100` | | Total Cost | Sum of `usage.cost.total` | | Avg Latency | Mean of `duration` | @@ -71,7 +72,7 @@ console.log(stats.byModel[0].avgTokensPerSecond); The web dashboard provides: -- Overall metrics cards (requests, cost, cache rate, error rate, duration, tokens/s) +- Overall metrics cards (requests, cost, cache rate, cache savings, error rate, duration, tokens/s) - Time series chart showing requests and errors over time - Per-model breakdown table - Per-folder breakdown table diff --git a/packages/stats/src/client/routes/ModelsRoute.tsx b/packages/stats/src/client/routes/ModelsRoute.tsx index 26d9869c1..b8b730237 100644 --- a/packages/stats/src/client/routes/ModelsRoute.tsx +++ b/packages/stats/src/client/routes/ModelsRoute.tsx @@ -322,7 +322,7 @@ function ModelsTable({
-
Quality
+
Efficiency
Error rate @@ -336,8 +336,18 @@ function ModelsTable({
Cache rate - - {(model.cacheRate * 100).toFixed(1)}% + {(model.cacheRate * 100).toFixed(1)}% +
+
+ Cache savings + + {(model.cacheSavings * 100).toFixed(1)}%
diff --git a/packages/stats/src/client/routes/ProjectsRoute.tsx b/packages/stats/src/client/routes/ProjectsRoute.tsx index fe4e78caf..92529ac0a 100644 --- a/packages/stats/src/client/routes/ProjectsRoute.tsx +++ b/packages/stats/src/client/routes/ProjectsRoute.tsx @@ -87,8 +87,16 @@ export function ProjectsRoute({ active, range, refreshTrigger }: ProjectsRoutePr key: "cacheRate", header: "Cache Rate", numeric: true, + render: (item: FolderRowView) => {formatPercent(item.cacheRate)}, + }, + { + key: "cacheSavings", + header: "Cache Savings", + numeric: true, render: (item: FolderRowView) => ( - {formatPercent(item.cacheRate)} + + {formatPercent(item.cacheSavings)} + ), }, { @@ -129,9 +137,13 @@ export function ProjectsRoute({ active, range, refreshTrigger }: ProjectsRoutePr
{formatCost(item.totalCost)}
-
Cache
+
Cache Rate
{formatPercent(item.cacheRate)}
+
+
Cache Savings
+
{formatPercent(item.cacheSavings)}
+
Duration
{formatDurationMs(item.avgDuration)}
diff --git a/packages/stats/src/client/styles.css b/packages/stats/src/client/styles.css index ffa46681b..b511656dd 100644 --- a/packages/stats/src/client/styles.css +++ b/packages/stats/src/client/styles.css @@ -563,7 +563,7 @@ .stats-metric-primary-grid { display: grid; - grid-template-columns: repeat(4, 1fr); + grid-template-columns: repeat(5, minmax(0, 1fr)); gap: 16px; } diff --git a/packages/stats/src/client/ui/MetricCluster.tsx b/packages/stats/src/client/ui/MetricCluster.tsx index f9edad1af..faaaa604a 100644 --- a/packages/stats/src/client/ui/MetricCluster.tsx +++ b/packages/stats/src/client/ui/MetricCluster.tsx @@ -29,7 +29,17 @@ export function MetricCluster({ stats }: MetricClusterProps) {
Requests
{formatInteger(stats.totalRequests)}
-
+
+
Cache Savings
+
{formatPercent(stats.cacheSavings)}
+
+
Cache Rate
{formatPercent(stats.cacheRate)}
diff --git a/packages/stats/src/db.ts b/packages/stats/src/db.ts index 9858e16de..d1fb5c1c6 100644 --- a/packages/stats/src/db.ts +++ b/packages/stats/src/db.ts @@ -2,7 +2,7 @@ import { Database } from "bun:sqlite"; import * as fs from "node:fs/promises"; import type { Usage } from "@oh-my-pi/pi-ai"; import type { GeneratedProvider } from "@oh-my-pi/pi-catalog/models"; -import { getBundledModel } from "@oh-my-pi/pi-catalog/models"; +import { calculateUncachedInputCost, getBundledModel } from "@oh-my-pi/pi-catalog/models"; import { getConfigRootDir, getStatsDbPath } from "@oh-my-pi/pi-utils"; import { classifyAgentType } from "./parser"; import type { @@ -33,7 +33,7 @@ import type { type ModelCost = { input: number; output: number; cacheRead: number; cacheWrite: number }; type UsageCost = Usage["cost"]; -type CostTokens = Pick; +type CostTokens = Pick; const ZERO_USAGE_COST: UsageCost = { input: 0, @@ -53,6 +53,42 @@ interface CostBackfillRow { cache_write_tokens: number; } +interface NoCacheInputCostBackfillRow { + id: number; + provider: string; + model: string; + input_tokens: number; + cache_read_tokens: number; + cache_write_tokens: number; +} + +interface AggregatedStatsRow { + total_requests: number; + failed_requests: number | null; + total_input_tokens: number | null; + total_output_tokens: number | null; + total_cache_read_tokens: number | null; + total_cache_write_tokens: number | null; + total_premium_requests: number | null; + total_cost: number | null; + total_cached_prompt_cost: number | null; + total_no_cache_input_cost: number | null; + avg_duration: number | null; + avg_ttft: number | null; + avg_tokens_per_second: number | null; + first_timestamp: number | null; + last_timestamp: number | null; +} + +interface ModelStatsRow extends AggregatedStatsRow { + model: string; + provider: string; +} + +interface FolderStatsRow extends AggregatedStatsRow { + folder: string; +} + let db: Database | null = null; const BACKFILL_COMPLETE = "complete"; @@ -112,6 +148,7 @@ export async function initDb(): Promise { cost_cache_read REAL NOT NULL, cost_cache_write REAL NOT NULL, cost_total REAL NOT NULL, + cost_no_cache_input REAL, agent_type TEXT NOT NULL DEFAULT 'main', UNIQUE(session_file, entry_id) ); @@ -183,6 +220,9 @@ export async function initDb(): Promise { if (!messageColumns.some(column => column.name === "premium_requests")) { db.run("ALTER TABLE messages ADD COLUMN premium_requests REAL NOT NULL DEFAULT 0"); } + if (!messageColumns.some(column => column.name === "cost_no_cache_input")) { + db.run("ALTER TABLE messages ADD COLUMN cost_no_cache_input REAL"); + } db.run("UPDATE messages SET premium_requests = 0 WHERE premium_requests IS NULL"); // Token-usage-by-agent: each message is classified main / subagent / advisor // from its transcript path. A brand-new table gets the column from CREATE @@ -265,6 +305,7 @@ export async function initDb(): Promise { backfillPriorityPremiumRequests(db); backfillAgentType(db); backfillMissingCatalogCosts(db); + backfillNoCacheInputCosts(db); backfillForkDuplicates(db); return db; } @@ -323,6 +364,18 @@ function resolveStoredCost(stats: MessageStats): UsageCost { return calculateCatalogCost(stats.provider, stats.model, stats.usage) ?? storedCost ?? ZERO_USAGE_COST; } +function calculateNoCacheInputCost(provider: string, modelId: string, tokens: CostTokens): number | null { + const cost = getCatalogCost(provider, modelId); + if (!cost) return null; + const promptInputTokens = + tokens.input + + tokens.cacheRead + + tokens.cacheWrite + + (tokens.orchestration?.input ?? 0) + + (tokens.orchestration?.cacheRead ?? 0); + return calculateUncachedInputCost(cost, promptInputTokens); +} + function backfillMissingCatalogCosts(database: Database): void { const rows = database .prepare(` @@ -358,6 +411,31 @@ function backfillMissingCatalogCosts(database: Database): void { applyBackfill(); } +function backfillNoCacheInputCosts(database: Database): void { + const rows = database + .prepare(` + SELECT id, provider, model, input_tokens, cache_read_tokens, cache_write_tokens + FROM messages + WHERE cost_no_cache_input IS NULL + `) + .all() as NoCacheInputCostBackfillRow[]; + if (rows.length === 0) return; + + const update = database.prepare("UPDATE messages SET cost_no_cache_input = ? WHERE id = ?"); + const applyBackfill = database.transaction(() => { + for (const row of rows) { + const cost = calculateNoCacheInputCost(row.provider, row.model, { + input: row.input_tokens, + output: 0, + cacheRead: row.cache_read_tokens, + cacheWrite: row.cache_write_tokens, + }); + update.run(cost ?? 0, row.id); + } + }); + applyBackfill(); +} + /** * Get the stored offset for a session file. */ @@ -406,9 +484,9 @@ export function insertMessageStats(stats: MessageStats[]): number { session_file, entry_id, folder, model, provider, api, timestamp, duration, ttft, stop_reason, error_message, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, total_tokens, premium_requests, - cost_input, cost_output, cost_cache_read, cost_cache_write, cost_total, agent_type + cost_input, cost_output, cost_cache_read, cost_cache_write, cost_total, cost_no_cache_input, agent_type ) - SELECT ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? + SELECT ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? WHERE NOT EXISTS ( SELECT 1 FROM messages WHERE entry_id = ? AND timestamp = ? AND session_file <> ? @@ -422,6 +500,7 @@ export function insertMessageStats(stats: MessageStats[]): number { const insert = db.transaction(() => { for (const s of stats) { const cost = resolveStoredCost(s); + const noCacheInputCost = calculateNoCacheInputCost(s.provider, s.model, s.usage) ?? 0; const result = stmt.run( s.sessionFile, s.entryId, @@ -445,6 +524,7 @@ export function insertMessageStats(stats: MessageStats[]): number { cost.cacheRead, cost.cacheWrite, cost.total, + noCacheInputCost, s.agentType, // `WHERE NOT EXISTS` binds: skip when a different session_file // already holds this (entry_id, timestamp). @@ -463,7 +543,7 @@ export function insertMessageStats(stats: MessageStats[]): number { /** * Build aggregated stats from query results. */ -function buildAggregatedStats(rows: any[]): AggregatedStats { +function buildAggregatedStats(rows: AggregatedStatsRow[]): AggregatedStats { if (rows.length === 0) { return { totalRequests: 0, @@ -475,6 +555,7 @@ function buildAggregatedStats(rows: any[]): AggregatedStats { totalCacheReadTokens: 0, totalCacheWriteTokens: 0, cacheRate: 0, + cacheSavings: 0, totalCost: 0, totalPremiumRequests: 0, avgDuration: null, @@ -492,6 +573,8 @@ function buildAggregatedStats(rows: any[]): AggregatedStats { const totalInputTokens = row.total_input_tokens || 0; const totalCacheReadTokens = row.total_cache_read_tokens || 0; const totalPremiumRequests = row.total_premium_requests || 0; + const noCacheInputCost = row.total_no_cache_input_cost || 0; + const cachedPromptCost = row.total_cached_prompt_cost || 0; return { totalRequests, @@ -506,6 +589,7 @@ function buildAggregatedStats(rows: any[]): AggregatedStats { totalInputTokens + totalCacheReadTokens > 0 ? totalCacheReadTokens / (totalInputTokens + totalCacheReadTokens) : 0, + cacheSavings: noCacheInputCost > 0 ? (noCacheInputCost - cachedPromptCost) / noCacheInputCost : 0, totalCost: row.total_cost || 0, totalPremiumRequests, avgDuration: row.avg_duration, @@ -533,6 +617,10 @@ export function getOverallStats(cutoff?: number): AggregatedStats { SUM(cache_write_tokens) as total_cache_write_tokens, SUM(premium_requests) as total_premium_requests, SUM(cost_total) as total_cost, + SUM(CASE WHEN cost_no_cache_input > 0 + THEN cost_input + cost_cache_read + cost_cache_write + ELSE 0 END) as total_cached_prompt_cost, + SUM(cost_no_cache_input) as total_no_cache_input_cost, AVG(duration) as avg_duration, AVG(ttft) as avg_ttft, AVG(CASE WHEN duration > 0 THEN output_tokens * 1000.0 / duration ELSE NULL END) as avg_tokens_per_second, @@ -543,7 +631,7 @@ export function getOverallStats(cutoff?: number): AggregatedStats { `); const rows = hasCutoff ? stmt.all(cutoff) : stmt.all(); - return buildAggregatedStats(rows as any[]); + return buildAggregatedStats(rows as AggregatedStatsRow[]); } /** * Get stats grouped by model. @@ -564,6 +652,10 @@ export function getStatsByModel(cutoff?: number): ModelStats[] { SUM(cache_write_tokens) as total_cache_write_tokens, SUM(premium_requests) as total_premium_requests, SUM(cost_total) as total_cost, + SUM(CASE WHEN cost_no_cache_input > 0 + THEN cost_input + cost_cache_read + cost_cache_write + ELSE 0 END) as total_cached_prompt_cost, + SUM(cost_no_cache_input) as total_no_cache_input_cost, AVG(duration) as avg_duration, AVG(ttft) as avg_ttft, AVG(CASE WHEN duration > 0 THEN output_tokens * 1000.0 / duration ELSE NULL END) as avg_tokens_per_second, @@ -575,7 +667,7 @@ export function getStatsByModel(cutoff?: number): ModelStats[] { ORDER BY total_requests DESC `); - const rows = (hasCutoff ? stmt.all(cutoff) : stmt.all()) as any[]; + const rows = (hasCutoff ? stmt.all(cutoff) : stmt.all()) as ModelStatsRow[]; return rows.map(row => ({ model: row.model, provider: row.provider, @@ -601,6 +693,10 @@ export function getStatsByFolder(cutoff?: number): FolderStats[] { SUM(cache_write_tokens) as total_cache_write_tokens, SUM(premium_requests) as total_premium_requests, SUM(cost_total) as total_cost, + SUM(CASE WHEN cost_no_cache_input > 0 + THEN cost_input + cost_cache_read + cost_cache_write + ELSE 0 END) as total_cached_prompt_cost, + SUM(cost_no_cache_input) as total_no_cache_input_cost, AVG(duration) as avg_duration, AVG(ttft) as avg_ttft, AVG(CASE WHEN duration > 0 THEN output_tokens * 1000.0 / duration ELSE NULL END) as avg_tokens_per_second, @@ -612,7 +708,7 @@ export function getStatsByFolder(cutoff?: number): FolderStats[] { ORDER BY total_requests DESC `); - const rows = (hasCutoff ? stmt.all(cutoff) : stmt.all()) as any[]; + const rows = (hasCutoff ? stmt.all(cutoff) : stmt.all()) as FolderStatsRow[]; return rows.map(row => ({ folder: row.folder, ...buildAggregatedStats([row]), diff --git a/packages/stats/src/index.ts b/packages/stats/src/index.ts index 5894f73fb..cf7e42d5a 100755 --- a/packages/stats/src/index.ts +++ b/packages/stats/src/index.ts @@ -68,6 +68,7 @@ async function printStats(): Promise { console.log(` Input Tokens: ${formatNumber(overall.totalInputTokens)}`); console.log(` Output Tokens: ${formatNumber(overall.totalOutputTokens)}`); console.log(` Cache Rate: ${formatPercent(overall.cacheRate)}`); + console.log(` Cache Savings: ${formatPercent(overall.cacheSavings)}`); console.log(` Total Cost: ${formatCost(overall.totalCost)}`); console.log(` Premium Requests: ${formatNumber(normalizePremiumRequests(overall.totalPremiumRequests ?? 0))}`); console.log(` Avg Duration: ${overall.avgDuration !== null ? formatDuration(overall.avgDuration) : "-"}`); @@ -80,7 +81,7 @@ async function printStats(): Promise { console.log("\nBy Model:"); for (const m of byModel.slice(0, 10)) { console.log( - ` ${m.model}: ${formatNumber(m.totalRequests)} reqs, ${formatCost(m.totalCost)}, ${formatPercent(m.cacheRate)} cache`, + ` ${m.model}: ${formatNumber(m.totalRequests)} reqs, ${formatCost(m.totalCost)}, ${formatPercent(m.cacheRate)} cache rate, ${formatPercent(m.cacheSavings)} cache savings`, ); } } diff --git a/packages/stats/src/shared-types.ts b/packages/stats/src/shared-types.ts index 1bcb2639b..27e2ce59a 100644 --- a/packages/stats/src/shared-types.ts +++ b/packages/stats/src/shared-types.ts @@ -25,8 +25,13 @@ export interface AggregatedStats { totalCacheReadTokens: number; /** Total cache write tokens */ totalCacheWriteTokens: number; - /** Cache hit rate (0-1) */ + /** Percentage of prompt input tokens served from cache (0-1). */ cacheRate: number; + /** + * Prompt-input cost saved relative to billing the same tokens uncached + * (0-1; negative when cache writes cost more than reads save). + */ + cacheSavings: number; /** Total cost */ totalCost: number; /** Total premium requests */ diff --git a/packages/stats/test/db-cost.test.ts b/packages/stats/test/db-cost.test.ts index 8b66ef4eb..08fdef86c 100644 --- a/packages/stats/test/db-cost.test.ts +++ b/packages/stats/test/db-cost.test.ts @@ -1,6 +1,6 @@ import { Database } from "bun:sqlite"; import { describe, expect, it } from "bun:test"; -import { closeDb, getRecentRequests, initDb, insertMessageStats } from "@oh-my-pi/omp-stats/db"; +import { closeDb, getOverallStats, getRecentRequests, initDb, insertMessageStats } from "@oh-my-pi/omp-stats/db"; import type { MessageStats } from "@oh-my-pi/omp-stats/types"; import { getBundledModel } from "@oh-my-pi/pi-catalog/models"; import { getStatsDbPath } from "@oh-my-pi/pi-utils"; @@ -46,6 +46,32 @@ function expectedCodexGptCost() { }; } +function createAnthropicCacheStats(entryId: string, cacheRead: number, cacheWrite: number): MessageStats { + const input = 1_000 - cacheRead - cacheWrite; + return { + sessionFile: "/tmp/anthropic-session.jsonl", + entryId, + folder: "/tmp/project", + model: "claude-sonnet-4-6", + provider: "anthropic", + api: "anthropic-messages", + timestamp: Date.now(), + duration: 1000, + ttft: 100, + stopReason: "stop", + errorMessage: null, + usage: { + input, + output: 0, + cacheRead, + cacheWrite, + totalTokens: 1_000, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + agentType: "main", + }; +} + describe("stats GPT cost correction", () => { it("stores catalog-derived cost when OpenAI Codex session usage has zero cost", async () => { await initDb(); @@ -107,3 +133,55 @@ describe("stats GPT cost correction", () => { expect(request?.usage.cost.total).toBeCloseTo(expectedCodexGptCost().total, 8); }); }); + +describe("stats cache metrics", () => { + it("subtracts 5-minute writes from the savings produced by cache reads", async () => { + await initDb(); + insertMessageStats([createAnthropicCacheStats("mixed-cache", 800, 100)]); + + // 100 uncached + 800 reads at 0.1x + 100 writes at 1.25x = 305, + // versus 1,000 tokens at the uncached input rate. + expect(getOverallStats().cacheSavings).toBeCloseTo(0.695, 8); + expect(getOverallStats().cacheRate).toBeCloseTo(800 / 900, 8); + }); + + it("reports cache writes without reads as negative savings", async () => { + await initDb(); + insertMessageStats([createAnthropicCacheStats("cache-write", 0, 1_000)]); + + expect(getOverallStats().cacheSavings).toBeCloseTo(-0.25, 8); + }); + + it("charges 1-hour cache writes at their full overhead", async () => { + await initDb(); + const stats = createAnthropicCacheStats("one-hour-write", 0, 1_000); + stats.usage.cost = { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0.006, + total: 0.006, + }; + insertMessageStats([stats]); + + expect(getOverallStats().cacheSavings).toBeCloseTo(-1, 8); + }); + + it("excludes unpriced custom models from the savings ratio", async () => { + await initDb(); + const known = createAnthropicCacheStats("known", 800, 100); + const unpriced = createAnthropicCacheStats("unpriced", 0, 0); + unpriced.provider = "custom"; + unpriced.model = "custom-model"; + unpriced.usage.cost = { + input: 1, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 1, + }; + insertMessageStats([known, unpriced]); + + expect(getOverallStats().cacheSavings).toBeCloseTo(0.695, 8); + }); +}); diff --git a/packages/stats/test/overview-token-labels.test.tsx b/packages/stats/test/overview-token-labels.test.tsx index 03f49635f..d96397dd1 100644 --- a/packages/stats/test/overview-token-labels.test.tsx +++ b/packages/stats/test/overview-token-labels.test.tsx @@ -14,6 +14,7 @@ const stats: AggregatedStats = { totalCacheReadTokens: 300, totalCacheWriteTokens: 40, cacheRate: 0.75, + cacheSavings: 0.695, totalCost: 0, totalPremiumRequests: 0, avgDuration: 1000, @@ -31,6 +32,11 @@ describe("overview token metrics", () => { expect(html).toContain("Cache Read"); expect(html).toContain("Conversation Total"); expect(html).toContain("Uncached input + cache reads + cache writes + output"); + expect(html).toContain("Cache Rate"); + expect(html).toContain("Cache Savings"); + expect(html).toContain("75.0%"); + expect(html).toContain("69.5%"); + expect(html).toContain("cache writes can make this negative"); const expectedTotal = formatCompact( stats.totalInputTokens + diff --git a/scripts/bazel-natives.ts b/scripts/bazel-natives.ts index 17725627c..b60fe92fd 100755 --- a/scripts/bazel-natives.ts +++ b/scripts/bazel-natives.ts @@ -25,13 +25,16 @@ * (packages/natives/scripts/build-bindings.ts) against the installed VS Build * Tools; every other target on a win32 host fails fast with guidance. * + * Set `OMP_NATIVE_BUILD_BACKEND=cargo` to route the host target through the + * same local N-API build on systems where Bazel's prebuilt host tools cannot run. + * * Note: musl addons intentionally reuse the plain linux- filenames, so a * `linux-all` copy overwrites the gnu addon with the musl one (and vice versa); * CI jobs that ship files always request an explicit disjoint target set. */ import * as fs from "node:fs/promises"; import * as path from "node:path"; -import { detectHostAvx2Support } from "./host-detect"; +import { detectHostAvx2Support, resolveLocalHostAddon } from "./host-detect"; const repoRoot = path.join(import.meta.dir, ".."); @@ -221,14 +224,10 @@ async function installAddon(sourcePath: string, destPath: string): Promise } } -/** - * win32-host path for the `host` pseudo-target: the bazel msvc cross toolchain - * cannot run here, but real MSVC can — build the addon via the napi local - * build and install it into destDir like the bazel path would. - */ -async function buildWindowsHostAddon(host: HostInfo, destDir: string): Promise { +/** Build and install the host addon through the local Cargo/N-API path. */ +async function buildLocalHostAddon(host: HostInfo, destDir: string): Promise { const script = path.join(repoRoot, "packages/natives/scripts/build-bindings.ts"); - console.log(`win32 host: bazel msvc toolchain is linux/mac-only; building via ${path.relative(repoRoot, script)}`); + console.log(`local host build: using ${path.relative(repoRoot, script)}`); const proc = Bun.spawn([process.execPath, script], { cwd: repoRoot, stdout: "inherit", @@ -237,7 +236,7 @@ async function buildWindowsHostAddon(host: HostInfo, destDir: string): Promise { const host: HostInfo = { platform: process.platform, arch: process.arch, avx2: detectHostAvx2Support() }; const destDir = options.dest ? path.resolve(options.dest) : path.join(repoRoot, "packages/natives/native"); - if (host.platform === "win32" && !options.source) { + if ((host.platform === "win32" || Bun.env.OMP_NATIVE_BUILD_BACKEND === "cargo") && !options.source) { if (options.targets.length !== 1 || options.targets[0] !== "host") { - throw new Error( - `Cannot bazel-build [${options.targets.join(", ")}] on a Windows host: the msvc cross ` + - "toolchain (bazel/toolchains/msvc) only runs on linux/mac exec hosts. Use `host` here " + - "(local napi build via VS Build Tools), or run this script from WSL/linux for cross targets.", - ); + if (host.platform === "win32") { + throw new Error( + `Cannot bazel-build [${options.targets.join(", ")}] on a Windows host: the msvc cross ` + + "toolchain (bazel/toolchains/msvc) only runs on linux/mac exec hosts. Use `host` here " + + "(local napi build via VS Build Tools), or run this script from WSL/linux for cross targets.", + ); + } + throw new Error("OMP_NATIVE_BUILD_BACKEND=cargo supports only the host target"); } - await buildWindowsHostAddon(host, destDir); + await buildLocalHostAddon(host, destDir); return; } let outputs: string[]; diff --git a/scripts/host-detect.ts b/scripts/host-detect.ts index 138e3ef3c..b722a882a 100644 --- a/scripts/host-detect.ts +++ b/scripts/host-detect.ts @@ -10,7 +10,27 @@ function runCommand(command: string, args: string[]): string | null { return null; } } +/** Local N-API addon identity derived from a host platform and ISA. */ +export interface LocalHostAddon { + readonly filename: string; + readonly x64Variant: "modern" | "baseline" | null; +} +/** Resolve the exact filename and x86-64 ISA emitted by the local N-API build. */ +export function resolveLocalHostAddon(host: { + readonly platform: string; + readonly arch: string; + readonly avx2: boolean; +}): LocalHostAddon { + const x64Variant = host.arch === "x64" ? (host.avx2 ? "modern" : "baseline") : null; + const variantSuffix = x64Variant ? `-${x64Variant}` : ""; + return { + filename: `pi_natives.${host.platform}-${host.arch}${variantSuffix}.node`, + x64Variant, + }; +} + +/** Detect whether this x86-64 host can run the modern AVX2 addon. */ export function detectHostAvx2Support(): boolean { if (process.arch !== "x64") return false;