From 4318e0d8d2a0005911bb617598cebbb499d729b7 Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Sat, 7 Nov 2026 12:40:08 +0000 Subject: [PATCH 1/2] fix(usage): ignore summed usage from server-side fallback replies Anthropic's server-side fallback can run one request twice when a content filter refuses the first model. The reply then reports the usage of both runs added together. Magic Context read that sum as the size of the prompt, so a session at 58% could show 110% and enter the force band, hold the turn for the historian, refuse it and fold early. A reply that carries the anthropic-auth fallback marker now keeps the previous pressure reading and only moves the response time, the same as a completed reply without input usage. OpenCode 1 is unchanged: its message events carry no reply content, so the marker does not reach the usage handler there. --- .../provider-response-completion.test.ts | 30 ++++++++++ .../shared/provider-response-completion.ts | 16 +++++ packages/plugin/src/v2/hooks/context.ts | 1 + .../v2/hooks/server-fallback-usage.test.ts | 58 +++++++++++++++++++ .../plugin/src/v2/hooks/usage-persist.test.ts | 42 ++++++++++++++ packages/plugin/src/v2/hooks/usage-persist.ts | 11 +++- packages/plugin/src/v2/hooks/usage-reading.ts | 8 ++- 7 files changed, 164 insertions(+), 2 deletions(-) create mode 100644 packages/plugin/src/shared/provider-response-completion.test.ts create mode 100644 packages/plugin/src/v2/hooks/server-fallback-usage.test.ts diff --git a/packages/plugin/src/shared/provider-response-completion.test.ts b/packages/plugin/src/shared/provider-response-completion.test.ts new file mode 100644 index 0000000000..6ba9219dad --- /dev/null +++ b/packages/plugin/src/shared/provider-response-completion.test.ts @@ -0,0 +1,30 @@ +import { expect, test } from "bun:test"; +import * as completion from "./provider-response-completion"; + +test("detects server-fallback usage only in the marked reasoning part", () => { + const content = [ + { type: "reasoning", text: "thinking", native: { signature: "real-signature" } }, + { + type: "reasoning", + text: " ", + native: { + signature: "cortexkit-server-fallback-v1:claude-opus-5-5|claude-opus-4-8", + }, + }, + { type: "text", text: "answer" }, + { type: "tool", name: "read" }, + ]; + expect(completion.hasServerFallbackUsage(content)).toBe(true); + expect(completion.hasServerFallbackUsage([content[0], content[2]])).toBe(false); + for (const invalid of [ + undefined, + null, + {}, + [null, {}, { type: "reasoning", native: { signature: 3 } }], + [{ type: "text", native: content[1]!.native }], + [{ type: "reasoning", native: { signature: "cortexkit-server-fallback-v10:models" } }], + [{ type: "reasoning", native: { signature: "other-cortexkit-server-fallback-v1:models" } }], + ]) { + expect(completion.hasServerFallbackUsage(invalid)).toBe(false); + } +}); diff --git a/packages/plugin/src/shared/provider-response-completion.ts b/packages/plugin/src/shared/provider-response-completion.ts index 22e0c15c3a..f7079c3e5f 100644 --- a/packages/plugin/src/shared/provider-response-completion.ts +++ b/packages/plugin/src/shared/provider-response-completion.ts @@ -1,3 +1,19 @@ +import { isRecord } from "./record-type-guard"; + +export function hasServerFallbackUsage(content: unknown): boolean { + return ( + Array.isArray(content) && + content.some( + (part) => + isRecord(part) && + part.type === "reasoning" && + isRecord(part.native) && + typeof part.native.signature === "string" && + part.native.signature.startsWith("cortexkit-server-fallback-v1:"), + ) + ); +} + /** Failure markers take precedence over a completion timestamp or token usage. */ export function providerResponseFailed(input: { finish?: unknown; error?: unknown }): boolean { if (input.error !== undefined && input.error !== null && input.error !== "") return true; diff --git a/packages/plugin/src/v2/hooks/context.ts b/packages/plugin/src/v2/hooks/context.ts index ce8f77afd7..f988f6c021 100644 --- a/packages/plugin/src/v2/hooks/context.ts +++ b/packages/plugin/src/v2/hooks/context.ts @@ -1517,6 +1517,7 @@ export async function registerContext(context: V2Context) { rowModel: latest?.data.model, draftModel: { providerID: draft.model.providerID, id: draft.model.id }, tokens: latest?.data.tokens, + content: latest?.data.content, // Completion is authoritative. Legacy finish-only rows use a // stable stored message time, never the time of this reread. completed: diff --git a/packages/plugin/src/v2/hooks/server-fallback-usage.test.ts b/packages/plugin/src/v2/hooks/server-fallback-usage.test.ts new file mode 100644 index 0000000000..62e6d52c4c --- /dev/null +++ b/packages/plugin/src/v2/hooks/server-fallback-usage.test.ts @@ -0,0 +1,58 @@ +import { expect, test } from "bun:test"; +import { getOrCreateSessionMeta } from "../../features/magic-context/storage"; +import { oc2ContextFixture } from "./__tests__/oc2-context.fixture"; + +for (const marked of [true, false]) { + test(`OC2 newest ${marked ? "server-fallback" : "normal control"} reply ${marked ? "keeps pressure and admits without execute" : "enters the force band"}`, async () => { + const f = await oc2ContextFixture(marked ? 801 : 802, { + area: "server-fallback-usage", + provider: "anthropic", + model: "claude-opus-5-5", + cacheTtl: "1h", + }); + try { + await f.pass(); + f.respond(1, 507_000); + await f.pass(); + const before = getOrCreateSessionMeta(f.db, f.session); + expect(before.lastInputTokens).toBe(507_000); + f.respond(1, 963_300); + const latest = f.rows.at(-1)!; + const row = f.raw + .prepare("SELECT data FROM session_message WHERE id=?") + .get(latest.id) as { data: string }; + const data = JSON.parse(row.data); + data.content.unshift({ + type: "reasoning", + text: " ", + native: { + signature: marked + ? "cortexkit-server-fallback-v1:claude-opus-5-5|claude-opus-4-8" + : "real-signature", + }, + }); + f.raw + .prepare("UPDATE session_message SET data=? WHERE id=?") + .run(JSON.stringify(data), latest.id); + const result = await f.attempt(); + const logs = result.logs.join("\n"); + const meta = getOrCreateSessionMeta(f.db, f.session); + if (marked) { + expect(result.error).toBeUndefined(); + expect(meta.lastInputTokens).toBe(507_000); + expect(meta.lastContextPercentage).toBe(before.lastContextPercentage); + expect(meta.lastUsageContextLimit).toBe(before.lastUsageContextLimit); + expect(meta.lastResponseTime).toBe(latest.created + 1000); + expect(logs).toMatch(/transform scheduler: .*decision=defer/); + expect(logs).not.toMatch(/transform scheduler: .*decision=execute/); + expect(logs).toContain("ignored server-fallback reply inputTokens=963300"); + } else { + expect(meta.lastInputTokens).toBe(963_300); + expect(meta.lastContextPercentage).toBeGreaterThan(95); + expect(logs).toMatch(/transform scheduler: .*decision=execute/); + } + } finally { + await f.cleanup(); + } + }); +} diff --git a/packages/plugin/src/v2/hooks/usage-persist.test.ts b/packages/plugin/src/v2/hooks/usage-persist.test.ts index 49f06e3bbf..6a556af889 100644 --- a/packages/plugin/src/v2/hooks/usage-persist.test.ts +++ b/packages/plugin/src/v2/hooks/usage-persist.test.ts @@ -65,6 +65,48 @@ async function configure272kWindow(): Promise { } describe("persistV2UsageReading", () => { + it("keeps pressure and the map entry for a server-fallback reply but moves completion time", () => { + process.env.XDG_DATA_HOME = makeTempDir("v2-fallback-usage-"); + const db = openDatabase(); + const sessionID = "v2-fallback-usage"; + const contextUsageMap: TransformDeps["contextUsageMap"] = new Map(); + const draftModel = { providerID: "anthropic", id: "claude-opus-5-5" }; + const persist = (input: number, completed: number, marked: boolean) => { + const reading = resolveUsageReading({ + draftModel, + tokens: { input }, + completed, + finish: "stop", + content: marked + ? [ + { + type: "reasoning", + text: " ", + native: { + signature: "cortexkit-server-fallback-v1:primary|fallback", + }, + }, + ] + : [{ type: "reasoning", native: { signature: "real-signature" } }], + limitFor: () => 872_000, + })!; + persistV2UsageReading({ db, sessionID, draftModel, reading, contextUsageMap }); + }; + persist(507_000, 1000, false); + const previous = getOrCreateSessionMeta(db, sessionID); + const entry = contextUsageMap.get(sessionID); + persist(963_300, 2000, true); + expect(getOrCreateSessionMeta(db, sessionID)).toEqual({ + ...previous, + lastResponseTime: 2000, + }); + expect(contextUsageMap.get(sessionID)).toBe(entry); + persist(510_000, 3000, false); + expect(getOrCreateSessionMeta(db, sessionID).lastInputTokens).toBe(510_000); + expect(contextUsageMap.get(sessionID)?.usage.inputTokens).toBe(510_000); + expect(getOrCreateSessionMeta(db, sessionID).lastResponseTime).toBe(3000); + }); + it("dates provider usage by completion when the store is read after a fold", () => { process.env.XDG_DATA_HOME = makeTempDir("v2-usage-age-"); const db = openDatabase(); diff --git a/packages/plugin/src/v2/hooks/usage-persist.ts b/packages/plugin/src/v2/hooks/usage-persist.ts index e1ea6447a1..3748b4d559 100644 --- a/packages/plugin/src/v2/hooks/usage-persist.ts +++ b/packages/plugin/src/v2/hooks/usage-persist.ts @@ -22,7 +22,7 @@ export interface PersistV2UsageReadingArgs { /** * Record the usage OpenCode stored for the latest completed reply. * - * The reading is the prompt size of a request the provider accepted, so it is + * An unmarked `reading` is the prompt size of a request the provider accepted, so it is * real pressure at any size. It is never compared with the configured window: * that window can be smaller than what the model actually serves, and a reading * past it is real overflow of the user's limit for the scheduler to handle. @@ -68,6 +68,15 @@ export function persistV2UsageReading(args: PersistV2UsageReadingArgs): void { !providerResponseFailed(completion) && (reading.inputTokens > 0 || isSuccessfulProviderCompletion(completion)); const responseTime = served && reading.completed; + // Server fallback sums two runs. It does not measure one context size. + if (reading.serverFallback) { + if (responseTime) updateSessionMeta(db, sessionID, { lastResponseTime: responseTime }); + sessionLog( + sessionID, + `v2 usage: ignored server-fallback reply inputTokens=${reading.inputTokens} (sum of two server runs); keeping previous reading`, + ); + return; + } if (reading.inputTokens <= 0 || !Number.isFinite(reading.limit) || reading.limit <= 0) { if (responseTime) updateSessionMeta(db, sessionID, { lastResponseTime: responseTime }); return; diff --git a/packages/plugin/src/v2/hooks/usage-reading.ts b/packages/plugin/src/v2/hooks/usage-reading.ts index bdefa8b664..5350d08c4c 100644 --- a/packages/plugin/src/v2/hooks/usage-reading.ts +++ b/packages/plugin/src/v2/hooks/usage-reading.ts @@ -1,4 +1,7 @@ -import { isSuccessfulProviderCompletion } from "../../shared/provider-response-completion"; +import { + hasServerFallbackUsage, + isSuccessfulProviderCompletion, +} from "../../shared/provider-response-completion"; export interface UsageReadingInput { rowModel?: { providerID?: unknown; id?: unknown }; @@ -11,6 +14,7 @@ export interface UsageReadingInput { completed?: number; finish?: string; error?: unknown; + content?: unknown; /** Resolve the output-reserved usable window for a model. */ limitFor: (providerID: string, modelID: string) => number; } @@ -23,6 +27,7 @@ export interface UsageReading { admissionLimit: number; /** Absent when a legacy row carries no model metadata. */ modelKey?: string; + serverFallback?: boolean; completed?: number; finish?: string; error?: unknown; @@ -80,5 +85,6 @@ export function resolveUsageReading(input: UsageReadingInput): UsageReading | un : {}), ...(input.finish !== undefined ? { finish: input.finish } : {}), ...(input.error !== undefined ? { error: input.error } : {}), + ...(hasServerFallbackUsage(input.content) ? { serverFallback: true } : {}), }; } From e739317ee25edcb1b68e0b1869a717382c77b8a7 Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Mon, 9 Nov 2026 09:09:44 +0000 Subject: [PATCH 2/2] fix(usage): read fallback pressure from per-run usage anthropic-auth 2.5.8 and later put a per-run usage marker on every step whose usage Anthropic summed: the handoff step and later steps that stay on the fallback model. On OpenCode 2 the context size now comes from the last fallback_message entry of that marker (input + cache read + cache write, a missing count is 0). The billed top-level usage is not changed. A reply with only the boundary marker (an older anthropic-auth) keeps the previous reading and moves only the response time. A malformed usage marker is treated the same way and logged. Replies without a marker are unchanged. The signature is read from native.signature (OpenCode 2.2+) and from state.signature (2.0.x). Refs #877 --- .../provider-response-completion.test.ts | 15 ++++ .../shared/provider-response-completion.ts | 68 ++++++++++++--- .../v2/hooks/server-fallback-usage.test.ts | 77 ++++++++++++++--- .../plugin/src/v2/hooks/usage-persist.test.ts | 56 +++++++++++++ packages/plugin/src/v2/hooks/usage-persist.ts | 4 +- .../plugin/src/v2/hooks/usage-reading.test.ts | 82 +++++++++++++++++++ packages/plugin/src/v2/hooks/usage-reading.ts | 15 +++- 7 files changed, 292 insertions(+), 25 deletions(-) diff --git a/packages/plugin/src/shared/provider-response-completion.test.ts b/packages/plugin/src/shared/provider-response-completion.test.ts index 6ba9219dad..e052beef09 100644 --- a/packages/plugin/src/shared/provider-response-completion.test.ts +++ b/packages/plugin/src/shared/provider-response-completion.test.ts @@ -1,6 +1,21 @@ import { expect, test } from "bun:test"; import * as completion from "./provider-response-completion"; +for (const location of ["native", "state"]) { + test(`detects usage and boundary markers in ${location}.signature`, () => { + for (const signature of [ + 'cortexkit-server-fallback-usage-v1:{"iterations":[]}', + "cortexkit-server-fallback-v1:primary|fallback", + ]) { + expect( + completion.hasServerFallbackUsage([ + { type: "reasoning", [location]: { signature } }, + ]), + ).toBe(true); + } + }); +} + test("detects server-fallback usage only in the marked reasoning part", () => { const content = [ { type: "reasoning", text: "thinking", native: { signature: "real-signature" } }, diff --git a/packages/plugin/src/shared/provider-response-completion.ts b/packages/plugin/src/shared/provider-response-completion.ts index f7079c3e5f..1e98dea87c 100644 --- a/packages/plugin/src/shared/provider-response-completion.ts +++ b/packages/plugin/src/shared/provider-response-completion.ts @@ -1,17 +1,65 @@ import { isRecord } from "./record-type-guard"; +const SERVER_FALLBACK_USAGE_SIGNATURE_PREFIX = "cortexkit-server-fallback-usage-v1:"; +const SERVER_FALLBACK_SIGNATURE_PREFIX = "cortexkit-server-fallback-v1:"; + export function hasServerFallbackUsage(content: unknown): boolean { - return ( - Array.isArray(content) && - content.some( - (part) => - isRecord(part) && - part.type === "reasoning" && - isRecord(part.native) && - typeof part.native.signature === "string" && - part.native.signature.startsWith("cortexkit-server-fallback-v1:"), + return readServerFallbackUsage(content) !== undefined; +} + +type ServerFallbackUsage = + | { kind: "usage"; inputTokens: number } + | { kind: "boundary"; invalidUsage: boolean }; + +function parseServerFallbackInputTokens(json: string): number | undefined { + try { + const payload: unknown = JSON.parse(json); + if (!isRecord(payload) || !Array.isArray(payload.iterations)) return undefined; + let last: unknown; + for (const entry of payload.iterations) { + if (isRecord(entry) && entry.type === "fallback_message") last = entry; + } + if (!isRecord(last)) return undefined; + const counts = [ + last.input_tokens, + last.cache_read_input_tokens ?? 0, + last.cache_creation_input_tokens ?? 0, + ]; + if ( + !counts.every( + (count) => typeof count === "number" && Number.isFinite(count) && count >= 0, + ) ) - ); + return undefined; + const inputTokens = counts.reduce((sum, count) => sum + Number(count), 0); + return Number.isFinite(inputTokens) ? inputTokens : undefined; + } catch { + // Invalid usage must not replace the previous pressure measurement. + return undefined; + } +} + +export function readServerFallbackUsage(content: unknown): ServerFallbackUsage | undefined { + if (!Array.isArray(content)) return undefined; + let boundary = false; + let invalidUsage = false; + for (const part of content) { + if (!isRecord(part) || part.type !== "reasoning") continue; + for (const location of [part.native, part.state]) { + if (!isRecord(location) || typeof location.signature !== "string") continue; + const signature = location.signature; + if (signature.startsWith(SERVER_FALLBACK_USAGE_SIGNATURE_PREFIX)) { + const inputTokens = parseServerFallbackInputTokens( + signature.slice(SERVER_FALLBACK_USAGE_SIGNATURE_PREFIX.length), + ); + if (inputTokens !== undefined) return { kind: "usage", inputTokens }; + invalidUsage = true; + } else if (signature.startsWith(SERVER_FALLBACK_SIGNATURE_PREFIX)) { + boundary = true; + } + } + } + return boundary || invalidUsage ? { kind: "boundary", invalidUsage } : undefined; } /** Failure markers take precedence over a completion timestamp or token usage. */ diff --git a/packages/plugin/src/v2/hooks/server-fallback-usage.test.ts b/packages/plugin/src/v2/hooks/server-fallback-usage.test.ts index 62e6d52c4c..314692a22a 100644 --- a/packages/plugin/src/v2/hooks/server-fallback-usage.test.ts +++ b/packages/plugin/src/v2/hooks/server-fallback-usage.test.ts @@ -2,9 +2,19 @@ import { expect, test } from "bun:test"; import { getOrCreateSessionMeta } from "../../features/magic-context/storage"; import { oc2ContextFixture } from "./__tests__/oc2-context.fixture"; -for (const marked of [true, false]) { - test(`OC2 newest ${marked ? "server-fallback" : "normal control"} reply ${marked ? "keeps pressure and admits without execute" : "enters the force band"}`, async () => { - const f = await oc2ContextFixture(marked ? 801 : 802, { +const markers = [ + "boundary", + "usage-native", + "usage-state", + "invalid-json", + "missing-fallback", + "none", +]; +for (const marker of markers) { + const marked = marker !== "none"; + const usage = marker.startsWith("usage"); + test(`OC2 newest ${marker} reply ${marked ? "admits at per-run pressure without execute" : "enters the force band"}`, async () => { + const f = await oc2ContextFixture(801 + markers.indexOf(marker), { area: "server-fallback-usage", provider: "anthropic", model: "claude-opus-5-5", @@ -25,12 +35,42 @@ for (const marked of [true, false]) { data.content.unshift({ type: "reasoning", text: " ", - native: { - signature: marked - ? "cortexkit-server-fallback-v1:claude-opus-5-5|claude-opus-4-8" - : "real-signature", + [marker === "usage-state" ? "state" : "native"]: { + signature: usage + ? `cortexkit-server-fallback-usage-v1:${JSON.stringify({ + iterations: [ + { + type: "message", + model: "primary", + input_tokens: 453_300, + output_tokens: 50, + }, + { + type: "fallback_message", + model: "fallback", + input_tokens: 100_000, + output_tokens: 50, + cache_read_input_tokens: 400_000, + cache_creation_input_tokens: 10_000, + }, + ], + })}` + : marker === "invalid-json" + ? "cortexkit-server-fallback-usage-v1:{" + : marker === "missing-fallback" + ? 'cortexkit-server-fallback-usage-v1:{"iterations":[]}' + : marked + ? "cortexkit-server-fallback-v1:claude-opus-5-5|claude-opus-4-8" + : "real-signature", }, }); + if (usage) + data.content.unshift({ + type: "reasoning", + native: { + signature: "cortexkit-server-fallback-v1:primary|fallback", + }, + }); f.raw .prepare("UPDATE session_message SET data=? WHERE id=?") .run(JSON.stringify(data), latest.id); @@ -39,13 +79,30 @@ for (const marked of [true, false]) { const meta = getOrCreateSessionMeta(f.db, f.session); if (marked) { expect(result.error).toBeUndefined(); - expect(meta.lastInputTokens).toBe(507_000); - expect(meta.lastContextPercentage).toBe(before.lastContextPercentage); + expect(meta.lastInputTokens).toBe(usage ? 510_000 : 507_000); + expect(meta.lastContextPercentage).toBeCloseTo( + usage + ? (510_000 / before.lastUsageContextLimit) * 100 + : before.lastContextPercentage, + 5, + ); expect(meta.lastUsageContextLimit).toBe(before.lastUsageContextLimit); expect(meta.lastResponseTime).toBe(latest.created + 1000); expect(logs).toMatch(/transform scheduler: .*decision=defer/); expect(logs).not.toMatch(/transform scheduler: .*decision=execute/); - expect(logs).toContain("ignored server-fallback reply inputTokens=963300"); + if (!usage) + expect(logs).toContain("ignored server-fallback reply inputTokens=963300"); + if (marker === "invalid-json" || marker === "missing-fallback") { + expect( + result.logs.filter((line) => + line.includes("invalid server-fallback usage marker"), + ), + ).toHaveLength(1); + const repeat = await f.attempt(); + expect(repeat.logs.join("\n")).not.toContain( + "invalid server-fallback usage marker", + ); + } } else { expect(meta.lastInputTokens).toBe(963_300); expect(meta.lastContextPercentage).toBeGreaterThan(95); diff --git a/packages/plugin/src/v2/hooks/usage-persist.test.ts b/packages/plugin/src/v2/hooks/usage-persist.test.ts index 6a556af889..4cc63052a3 100644 --- a/packages/plugin/src/v2/hooks/usage-persist.test.ts +++ b/packages/plugin/src/v2/hooks/usage-persist.test.ts @@ -65,6 +65,62 @@ async function configure272kWindow(): Promise { } describe("persistV2UsageReading", () => { + it("records last-iteration pressure instead of billed usage and accepts the next normal reply", () => { + process.env.XDG_DATA_HOME = makeTempDir("v2-per-run-usage-"); + const db = openDatabase(); + const sessionID = "v2-per-run-usage"; + const contextUsageMap: TransformDeps["contextUsageMap"] = new Map(); + const draftModel = { providerID: "anthropic", id: "claude-opus-5-5" }; + const tokens = { input: 963_300 }; + const persist = (completed: number, content: unknown) => { + const reading = resolveUsageReading({ + draftModel, + tokens, + completed, + finish: "stop", + content, + limitFor: () => 872_000, + })!; + persistV2UsageReading({ db, sessionID, draftModel, reading, contextUsageMap }); + }; + persist(1000, [ + { + type: "reasoning", + native: { + signature: `cortexkit-server-fallback-usage-v1:${JSON.stringify({ + iterations: [ + { + type: "message", + model: "primary", + input_tokens: 453_300, + output_tokens: 50, + }, + { + type: "fallback_message", + model: "fallback", + input_tokens: 100_000, + output_tokens: 50, + cache_read_input_tokens: 400_000, + cache_creation_input_tokens: 10_000, + }, + ], + })}`, + }, + }, + ]); + const meta = getOrCreateSessionMeta(db, sessionID); + expect(meta.lastInputTokens).toBe(510_000); + expect(meta.lastContextPercentage).toBeCloseTo(58.4862, 3); + expect(meta.lastResponseTime).toBe(1000); + expect(contextUsageMap.get(sessionID)?.usage.inputTokens).toBe(510_000); + expect(contextUsageMap.get(sessionID)?.usage.percentage).toBeCloseTo(58.4862, 3); + expect(tokens.input).toBe(963_300); + tokens.input = 520_000; + persist(2000, []); + expect(getOrCreateSessionMeta(db, sessionID).lastInputTokens).toBe(520_000); + expect(contextUsageMap.get(sessionID)?.usage.inputTokens).toBe(520_000); + }); + it("keeps pressure and the map entry for a server-fallback reply but moves completion time", () => { process.env.XDG_DATA_HOME = makeTempDir("v2-fallback-usage-"); const db = openDatabase(); diff --git a/packages/plugin/src/v2/hooks/usage-persist.ts b/packages/plugin/src/v2/hooks/usage-persist.ts index 3748b4d559..2d736b4473 100644 --- a/packages/plugin/src/v2/hooks/usage-persist.ts +++ b/packages/plugin/src/v2/hooks/usage-persist.ts @@ -68,12 +68,12 @@ export function persistV2UsageReading(args: PersistV2UsageReadingArgs): void { !providerResponseFailed(completion) && (reading.inputTokens > 0 || isSuccessfulProviderCompletion(completion)); const responseTime = served && reading.completed; - // Server fallback sums two runs. It does not measure one context size. + // Without usage for each run, the server sum does not measure one context size. if (reading.serverFallback) { if (responseTime) updateSessionMeta(db, sessionID, { lastResponseTime: responseTime }); sessionLog( sessionID, - `v2 usage: ignored server-fallback reply inputTokens=${reading.inputTokens} (sum of two server runs); keeping previous reading`, + `v2 usage: ignored server-fallback reply inputTokens=${reading.inputTokens} (${reading.serverFallbackInvalidUsage ? "invalid server-fallback usage marker" : "sum of two server runs"}); keeping previous reading`, ); return; } diff --git a/packages/plugin/src/v2/hooks/usage-reading.test.ts b/packages/plugin/src/v2/hooks/usage-reading.test.ts index c641406246..4ad0815ae1 100644 --- a/packages/plugin/src/v2/hooks/usage-reading.test.ts +++ b/packages/plugin/src/v2/hooks/usage-reading.test.ts @@ -4,6 +4,88 @@ import { resolveUsageReading, usageReadingMatchesDraft } from "./usage-reading"; const windows: Record = { old: 200_000, new: 1_000_000 }; const limitFor = (_providerID: string, modelID: string) => windows[modelID] ?? 0; +for (const location of ["native", "state"]) { + for (const cache of [true, false]) { + test(`reads the last fallback iteration from ${location} with cache=${cache}`, () => { + const tokens = { input: 963_300, output: 100, cache: { read: 1, write: 2 } }; + const reading = resolveUsageReading({ + draftModel: { providerID: "p", id: "new" }, + tokens, + completed: 123, + content: [ + { + type: "reasoning", + [location]: { + signature: `cortexkit-server-fallback-usage-v1:${JSON.stringify({ + iterations: [ + { + type: "message", + model: "primary", + input_tokens: 400_000, + output_tokens: 10, + }, + { + type: "fallback_message", + model: "fallback", + input_tokens: 500_000, + output_tokens: 20, + }, + { + type: "fallback_message", + model: "fallback", + input_tokens: 100_000, + output_tokens: 30, + ...(cache + ? { + cache_read_input_tokens: 400_000, + cache_creation_input_tokens: 10_000, + } + : {}), + }, + { + type: "message", + model: "primary", + input_tokens: 900_000, + output_tokens: 40, + }, + ], + })}`, + }, + }, + ], + limitFor, + }); + expect(reading?.inputTokens).toBe(cache ? 510_000 : 100_000); + expect(reading?.serverFallback).not.toBe(true); + expect(tokens).toEqual({ input: 963_300, output: 100, cache: { read: 1, write: 2 } }); + }); + } +} + +for (const payload of [ + "{", + "null", + '{"iterations":[]}', + '{"iterations":[{"type":"message"}]}', + '{"iterations":[{"type":"fallback_message","input_tokens":"bad"}]}', +]) { + test(`invalid usage marker keeps the boundary-only path: ${payload}`, () => { + const reading = resolveUsageReading({ + draftModel: { providerID: "p", id: "new" }, + tokens: { input: 963_300 }, + completed: 123, + content: [ + { + type: "reasoning", + state: { signature: `cortexkit-server-fallback-usage-v1:${payload}` }, + }, + ], + limitFor, + }); + expect(reading?.serverFallback).toBe(true); + }); +} + test("a same-model reading uses one window for attribution and admission", () => { const reading = resolveUsageReading({ rowModel: { providerID: "p", id: "old" }, diff --git a/packages/plugin/src/v2/hooks/usage-reading.ts b/packages/plugin/src/v2/hooks/usage-reading.ts index 5350d08c4c..da517a254a 100644 --- a/packages/plugin/src/v2/hooks/usage-reading.ts +++ b/packages/plugin/src/v2/hooks/usage-reading.ts @@ -1,6 +1,6 @@ import { - hasServerFallbackUsage, isSuccessfulProviderCompletion, + readServerFallbackUsage, } from "../../shared/provider-response-completion"; export interface UsageReadingInput { @@ -28,6 +28,7 @@ export interface UsageReading { /** Absent when a legacy row carries no model metadata. */ modelKey?: string; serverFallback?: boolean; + serverFallbackInvalidUsage?: boolean; completed?: number; finish?: string; error?: unknown; @@ -63,8 +64,11 @@ export function resolveUsageReading(input: UsageReadingInput): UsageReading | un const rowModelID = typeof input.rowModel?.id === "string" ? input.rowModel.id : undefined; const measuredProviderID = rowProviderID ?? input.draftModel.providerID; const measuredModelID = rowModelID ?? input.draftModel.id; + const fallback = readServerFallbackUsage(input.content); const inputTokens = - numeric(tokens?.input) + numeric(tokens?.cache?.read) + numeric(tokens?.cache?.write); + fallback?.kind === "usage" + ? fallback.inputTokens + : numeric(tokens?.input) + numeric(tokens?.cache?.read) + numeric(tokens?.cache?.write); const limit = input.limitFor(measuredProviderID, measuredModelID); if ((!Number.isFinite(limit) || limit <= 0) && !successfulCompletion) return undefined; const sameModel = @@ -85,6 +89,11 @@ export function resolveUsageReading(input: UsageReadingInput): UsageReading | un : {}), ...(input.finish !== undefined ? { finish: input.finish } : {}), ...(input.error !== undefined ? { error: input.error } : {}), - ...(hasServerFallbackUsage(input.content) ? { serverFallback: true } : {}), + ...(fallback?.kind === "boundary" + ? { + serverFallback: true, + ...(fallback.invalidUsage ? { serverFallbackInvalidUsage: true } : {}), + } + : {}), }; }