diff --git a/docs/billing-reports.md b/docs/billing-reports.md new file mode 100644 index 0000000..5ee7ae0 --- /dev/null +++ b/docs/billing-reports.md @@ -0,0 +1,164 @@ +# Vault-served request billing + +OpenCode 2 HTTP, its xAI Responses WebSocket frames, and Pi send optional +`kind: billing` reports through `@cortexkit/common-auth/routing-accessor`. +OpenCode 1 does not report. The plugin creates an accounting record only after +a request carrying the plugin's placeholder key (not the user's own key) +receives a vault receipt identifying the credential authorized for that send, +and the process-global +`Symbol.for('cortexkit.routing')` accessor is callable. Without it, billing does +not wrap responses, acquire readers, or look up usage semantics. Existing +credential-redaction and host lifecycle wrappers still apply. + +Each reported model-call attempt has its own random UUID, including retries and +account reroutes. Same-origin redirect hops that core follows are not model +calls and produce no report; only the terminal send that core returns to the +host, or a send that ends in an error, reports. HTTP reports settle when the body ends, +errors or is cancelled, not when headers arrive. Successful Pi sends wait for +pi-ai's final assistant message to supply the stored `responseId` and outcome; +their usage comes from the provider reply through the same table as OpenCode 2, +not pi-ai's normalized counters (which supply zeroes for missing fields). +Unsuccessful sends settle before replacement sends. xAI WebSocket reports settle +at the terminal Responses frame, or on the host's attempt-end callback for an +abandoned/failed attempt. The socket can remain open between sends. + +Reports carry the receipt's exact credential id and, if present, account identity. +`provider` is the **host catalog id**, not the provider id in the vault's fleet +roster: Kimi is +`kimi-code-plan-global` on OpenCode 2 and `kimi-coding` on Pi. **Pricing gap:** +Pi's `kimi-coding` id does not match the pricing catalog's +`kimi-code-plan-global`; that catalog mapping must be resolved on the pricing +side, without changing the host id in the report. The model comes +from the hook's `modelID` or Pi's `model.id`; no request body is parsed to find it. +Reply model names are ignored: providers often echo aliases, prefixes or dated +versions of the sent model. The plugin never sends `usageByModel`, because those +names are not a reliable reroute signal and may lack pricing catalog coverage. OpenCode's `kind` maps directly to `primary`, `title`, `compaction`, or +`generate`; Pi uses `primary`. Host-provided retry attempt numbers are retained +when available; Pi numbers reported model-call attempts within its stream. + +Outcomes follow the openai-auth convention: a provider error or non-2xx status +is `failed`; a completed reply whose body ended is `completed`; otherwise a +response with output is `cut` and one without output is `failed`. +When a host cancels after a provider terminal signal (Chat `finish_reason` or +`[DONE]`, Responses completed/incomplete, or parsed JSON), the answer is already +complete, so it reports `completed` even without byte EOF. Cancellation before +terminal and transport read errors remain cut/failed. If usage would have +arrived in a later chunk, it stays absent rather than being invented. Optional +accounting never awaits report delivery. Non-streaming JSON observation retains +at most 1 MiB; larger replies still report, without parsed usage. + +## Usage semantics and sources + +The `billing` rows in `packages/core/src/providers.ts` are the extraction and +semantics table for both hosts. Field paths below are relative to `usage` unless +specified otherwise. Missing counts stay absent; zero is retained only when +reported. Each stream retains the latest raw value of every reported counter; +cache subtraction, flooring at zero and write-lifetime summation happen once +when the report is built, so counts may arrive in different chunks. OpenCode's +cache-exclusive token namespace remains separate from generic prompt totals. +When prompt counts include cache but the reply supplies no cache +counter, `input` is the reported prompt count and **may include cache hits**. +The plugin never changes `stream_options`, `perf_metrics_in_response`, or any +other request field to obtain accounting. + +**Standard OpenAI assumption:** where provider documentation names standard +OpenAI fields but does not define their inclusion semantics, use OpenAI's +[Chat Completions usage](https://platform.openai.com/docs/api-reference/chat/object#chat/object-usage) +and [Responses usage](https://platform.openai.com/docs/api-reference/responses/object#responses/object-usage): +`prompt_tokens_details.cached_tokens` / `input_tokens_details.cached_tokens` are +subsets of prompt/input; `completion_tokens_details.reasoning_tokens` / +`output_tokens_details.reasoning_tokens` are included in output. Such rows are +marked **assumed (OpenAI field semantics)** below. Nonstandard undocumented +counts are not guessed. The generic OpenAI fallback reads either the Chat +`prompt_tokens` / `completion_tokens` pair or Responses `input_tokens` / +`output_tokens`, their standard cached/reasoning details, and no cache-write +field unless explicitly listed. + +| Provider | Usage fields / inclusion rules | Stream availability; automatic cache | Charge basis and official sources | +| --- | --- | --- | --- | +| DeepSeek | `prompt_tokens` includes `prompt_cache_hit_tokens` (preferred to standard cached detail); `completion_tokens` includes standard reasoning detail. Cache-read and reasoning inclusion **verified live** for `deepseek-flash` on 2026-11-11; see evidence below. No documented write count. | Usage on the final finish/content chunk even without opt-in. **Implicit cache**. | `metered_api`. [API](https://api-docs.deepseek.com/api/create-chat-completion), [context caching](https://api-docs.deepseek.com/guides/kv_cache), [pricing](https://api-docs.deepseek.com/quick_start/pricing). | +| Cerebras | `prompt_tokens` includes `prompt_tokens_details.cached_tokens`; `completion_tokens` includes standard reasoning detail, **verified live** for `qwen-3.8-27b` on 2026-11-11. Cache inclusion was untested in that run (no hit). No documented write count. | Documented stream includes usage without opt-in. **Implicit cache**. | Omitted: the same key can span trial, PayGo and Code Pro. [API](https://inference-docs.cerebras.ai/api-reference/chat-completions), [cache](https://inference-docs.cerebras.ai/capabilities/prompt-caching), [pricing](https://www.cerebras.ai/pricing), [Code plan](https://www.cerebras.ai/code). | +| OpenRouter | Generic OpenAI fields; documented `prompt_tokens_details.cache_write_tokens` also subtracted from prompt, floored at zero. **Assumed: details fields are a breakdown of prompt_tokens (writes included)**. Cache/reasoning inclusion otherwise uses **assumed (OpenAI field semantics)**. | Automatic final-chunk usage; `include_usage` is deprecated. **No universal implicit cache claim**: depends on routed model/provider. | `metered_api`. [API](https://openrouter.ai/docs/api_reference/overview), [usage](https://openrouter.ai/docs/cookbook/administration/usage-accounting), [cache](https://openrouter.ai/docs/guides/best-practices/prompt-caching), [pricing](https://openrouter.ai/pricing). | +| Meta | Chat/Responses generic OpenAI usage fields; cached/reasoning inclusion is **assumed (OpenAI field semantics)**. No documented write count. | **Implicit prefix cache**, no flag required. Protocol-dependent usage location. | `metered_api` for Model API, not the separate Muse Code subscription. [cache and usage fields](https://dev.meta.ai/docs/prompt-caching), [Responses](https://dev.meta.ai/docs/protocols/responses), [pricing](https://dev.meta.ai/docs/pricing-rate-limits). | +| OpenCode Go / Zen | Prefer `opencode.tokens.{input,cache_read,cache_write_5m,cache_write_1h,output,reasoning}`. Input already excludes reads and writes; output includes reasoning. Report both write lifetimes plus their sum. Generic OpenAI fallback is **assumed (OpenAI field semantics)**. | Chat's final usage chunk needs opt-in; Responses uses `response.completed`. Automatic caching **UNSTATED**, so no implicit claim. | Go omitted: allowance plus optional PayGo Extra Usage share a key. Zen `metered_api`. [inference API](https://opencode.ai/v2/docs/console/api/inference/), [Go](https://opencode.ai/docs/go/), [Zen](https://opencode.ai/docs/zen/). | +| Ollama Cloud | Cloud `/v1` usage details **UNSTATED**; generic OpenAI fields are **assumed (OpenAI field semantics)** when present. Do not borrow native `/api/chat`'s `prompt_eval_count` / `eval_count`. | Supports Chat `include_usage`; requirement and automatic cache for Cloud `/v1` **UNSTATED**. | Omitted: generic key spans pay-as-you-go and Pro/Max included credits. [OpenAI compatibility](https://docs.ollama.com/api/openai-compatibility), [pricing](https://ollama.com/pricing). | +| xAI | Responses `input_tokens`, `input_tokens_details.cached_tokens`, `output_tokens`, `output_tokens_details.reasoning_tokens` (Chat equivalents also accepted). Inclusion is **assumed (OpenAI field semantics)**. | Usage is taken from the terminal response object, including the frames already observed on WebSocket. **Implicit prefix cache**. | `metered_api`. [Responses](https://docs.x.ai/docs/api-reference#responses), [automatic caching](https://docs.x.ai/developers/advanced-api-usage/prompt-caching), [cache usage](https://docs.x.ai/developers/advanced-api-usage/prompt-caching/usage-and-pricing), [pricing](https://docs.x.ai/developers/pricing). | +| Fireworks | `prompt_tokens`, `completion_tokens`; cache read from final chunk's **`perf_metrics.cached-prompt-tokens`**, falling back to standard cached detail. Prompt includes cache. Standard reasoning detail is **assumed (OpenAI field semantics)** if present. No documented write count. | Usage included by default. Cache metrics require `perf_metrics_in_response`, which the plugin **does not add**; without them input may include cache hits. **Implicit cache**. | `metered_api`. [Chat API](https://docs.fireworks.ai/api-reference/post-chatcompletions), [cache](https://docs.fireworks.ai/guides/prompt-caching), [pricing](https://docs.fireworks.ai/serverless/pricing). | +| Inception | `prompt_tokens`, `completion_tokens` (includes reasoning). Accept `cached_input_tokens` or standard cached detail; `reasoning_tokens` or standard reasoning detail. Cache subtraction is **assumed (OpenAI field semantics)**. No documented write count. | Final usage-only chunk requires `include_usage`. Automatic cache **UNSTATED**. | `metered_api`. [API (both spellings)](https://docs.inceptionlabs.ai/api-reference/chat/create-a-chat-completion), [pricing](https://docs.inceptionlabs.ai/get-started/models.md). | +| Kimi Code | Kimi Chat fields: prompt includes cached reads **and** `prompt_tokens_details.cache_write_tokens`; completion includes standard reasoning detail. Cache-read and reasoning inclusion **verified live** for Code model `k3` on 2026-11-11. Cache-write inclusion remains **assumed applicable to Code plan** from regular Kimi Chat docs, not Code's endpoint page. | Final usage chunk requires `include_usage`. **Implicit cache**, also **assumed for the Code plan endpoint**. | `subscription_included` for the membership key, not a Platform key. [Code](https://www.kimi.com/code/docs/en), [Chat API](https://platform.moonshot.ai/docs/api/chat), [cache](https://platform.kimi.ai/docs/guide/context-caching). | +| Cline Pass | `prompt_tokens`, `completion_tokens`, standard cached detail. Generic reasoning detail, when present, and inclusion semantics are **assumed (OpenAI field semantics)**. | Exact usage opt-in requirement and automatic cache **UNSTATED**. | `subscription_included`: reference API rates are not per-request charges. [Chat API](https://docs.cline.bot/api/chat-completions.md), [ClinePass pricing](https://docs.cline.bot/getting-started/clinepass.md). | +| TypeSafe | Non-streaming JSON `usage.input_tokens` / `output_tokens`; no cache or reasoning fields claimed. | System One documents JSON, not a streamed usage contract. Automatic cache **UNSTATED**. | `metered_api`: API input tokens are billable. [System One](https://docs.typesafe.ai/api/system-one.md). | +| MiniMax Coding Plan | Generic OpenAI fields; standard cached detail documented. Reasoning detail and inclusion are **assumed (OpenAI field semantics)** if present; no guessed nonstandard write count. | **Implicit repeated-context cache**. | `subscription_included` for M Plan key, distinct from regular pay-as-you-go key. [Chat](https://platform.minimax.io/docs/api-reference/text-chat-openai.md), [cache](https://platform.minimax.io/docs/api-reference/text-prompt-caching.md), [plan key](https://platform.minimax.io/docs/m-plan/quickstart.md). | +| Synthetic | Documented `prompt_tokens`, `completion_tokens`; generic standard cache/reasoning fields, if present, are **assumed (OpenAI field semantics)**. | Opt-in requirement and automatic cache **UNSTATED**. | Omitted: subscription credits and usage-based requests/top-ups can share a key. [Chat](https://dev.synthetic.new/docs/openai/chat-completions), [pricing](https://synthetic.new/pricing), [model pricing](https://synthetic.new/pricing/models). | + +## Live evidence: 2026-11-11, OpenCode 2.2.3 + +The operator's `billing-run-1111b.log` captures the `--billing` mode against real +vault-served providers on **2026-11-11**, using **OpenCode 2.2.3**. These results +verify the listed models and fields, not every model or protocol a provider +offers. The two long prompts used fresh sessions with the same model/variant, +default agent and working directory; matching stored context totals are shown +below. This run does not establish cache-write semantics. + +### Verified live + +OpenCode's AI SDK stores visible output separately from reasoning. The report's +`output - reasoning` therefore matches stored `output` when reasoning is included +in the report's output. The original capture marked these reasoning turns FAIL +because its chars/4 heuristic underestimated short replies; the measured token +counts below establish inclusion instead. + +| Provider / model | Prompt includes cached reads | Output includes reasoning | +| --- | --- | --- | +| DeepSeek / `deepseek-flash` | **Verified:** first input **8281 = 217 + 8064** (second input + cache read). Stored context was 8281 on both long turns. | **Verified:** report output **43 - 41** reasoning **= 2** stored visible output. | +| Kimi Code / `k3` | **Verified:** first input **7709 = 0 + 7709** (second input + cache read). Stored context was 7709 on both long turns. | **Verified:** report output **53 - 39** reasoning **= 14** stored visible output. | +| Cerebras / `qwen-3.8-27b` | **Untested:** both fresh-session long turns reported cache read 0, with input/context 8567 each. No cache hit was observed. | **Verified:** report output **133 - 128** reasoning **= 5** stored visible output. | + +All three providers supplied input and output usage on all three completed +reports. These observations supplement the documentation sources above. + +### Still assumed, untested + +Cache/reasoning inclusion remains assumed where the table marks it as such: + +- **OpenRouter:** `cohere/north-mini-code:free` gave no cache hit. The first + completed report had input 7803, output 1 and cache read/write 0; the second + completed report carried **no usage**, so usage presence failed and inclusion + could not be verified. OpenCode's stored zeroes on that second turn are not + measured provider usage. +- **Fireworks:** provider-side HTTP **412** refusal. +- **Cline Pass:** provider-side HTTP **404** refusal. +- **Ollama Cloud:** provider-side HTTP **402** refusal. +- **Meta:** provider-side HTTP **403** refusal. +- **xAI:** provider-side HTTP **403** refusal (HTTP fallback was logged). +- **opencode-go:** provider-side HTTP **429** refusal. + +Refusals do not provide completed usage with which to verify inclusion semantics. + +### Not run + +- **Inception, MiniMax Coding Plan and Synthetic:** no usable account. +- **TypeSafe:** unknown to this host's catalog. +- **opencode-zen:** not served on this host. + +## Host stream options and cache policy + +The inspected OpenCode 2 source's openai-compatible protocol +(`packages/ai/src/protocols/openai-chat.ts`, `detectSupportsUsageInStreaming` and +`lowerRequest`) defaults streaming usage support to true and sends +`stream_options: { include_usage: true }`; a model compatibility override can +disable it. The `billing: a real vault-served HTTP turn` case in +`packages/opencode/test/v2/e2e/opencode2.e2e.test.ts` also verifies usage reaches +the plugin on the pinned `@opencode/cli` 2.2.0 host. +Installed pi-ai 1.3.0's `dist/api/openai-completions.js` (`buildParams` and +`getCompat`) likewise adds it unless `supportsUsageInStreaming === false`, with +support enabled for both DeepSeek and Cerebras. These are **host decisions**: +the auth plugin never adds it. A reply without usage still gets a report, +without `usage`. + +There is no keep-warm implementation. If the session's current common-auth +`sessionCachePolicy(sessionId)?.keep_warm` is set and not `off`, the report adds +`cache.warming: { on: false, stopReason: 'unsupported' }`. common-auth 0.43.0's +validator accepts warming by itself, without `implicit` or a TTL. Otherwise +warming is omitted. Only the documented-automatic providers in the table get +`cache.implicit: true`; there is no keep-warm traffic or request modification. diff --git a/docs/live-smoke.md b/docs/live-smoke.md index d0a705a..d74509e 100644 --- a/docs/live-smoke.md +++ b/docs/live-smoke.md @@ -16,7 +16,7 @@ The entry and README there are local, git-ignored operator artifacts; the shared driver is tracked at `scripts/live-smoke/run.ts` and can also be run directly with the same arguments. -For each served row with an eligible catalog model, the headless driver attempts +Without `--billing`, for each served row with an eligible catalog model, the headless driver attempts at most one provider request, with a 16-token output cap for non-reasoning models. Models without a reasoning-off control get 1024 tokens so hidden reasoning does not consume the entire answer budget. Meta always gets that @@ -43,6 +43,81 @@ driver-budget failure, not an account refusal; it is not retried. The second xAI turn and Anthropic refusal probes below remain manual checks, not extra headless requests. Raw API captures keep their sequence across host restarts. +## Billing semantics probe (OpenCode 2 only) + +Use the already-enrolled **scratch** home and test enrollment, never a personal +OpenCode profile. Preview first, then remove `--dry` to spend real provider tokens: + +```sh +LIVE_HOME= CLAUSTRUM_SUBC_CONNECTION= \ + bun scripts/live-smoke/run.ts --billing --provider deepseek --dry +# --provider all checks the runner's usual OpenCode 2 provider matrix. +``` + +`--billing` selects OpenCode 2 by default; an explicit `--host` must be +`opencode2`. It does not enable live smoke in CI. Activation, logout of only the +scratch credential activated by the runner, scratch isolation and the leak scan +remain in place. The server imports +a preload before plugin setup that installs `Symbol.for('cortexkit.routing')` +and appends **only billing report payloads** to +`$LIVE_HOME/.live-smoke/opencode2/billing.jsonl` (reset on each run). No headers, +requests or vault replies are recorded by the preload. The capture is inside +the existing scratch-home leak scan. + +For each served provider with an eligible text/tool model, the mode sends the +same fixed roughly 3,000-token prose prompt twice, ending in +“Reply with the single word OK”. Both use the same model/variant, default agent, +host and working directory. Each uses a **fresh session**: retaining the first +turn in conversation history would double the second prompt and invalidate the +cache comparison. Per-session system-prompt +differences are not observable from billing reports; compare the printed stored +agent and host context totals before interpreting a mismatch. If the catalog +advertises a reasoning-capable model, a third fresh session sends a short +arithmetic prompt with reasoning enabled at a low non-off variant when offered +and a 1,024-token output cap. No model ids are invented, retries and automatic +continuations remain blocked, and no request fields are added to obtain usage. + +The per-provider tables print every report's `outcome`, `chargeBasis`, `cache` +and usage, then verdicts from the reports (plus OpenCode's stored visible output +or visible reply length for the reasoning check): + +- **Prompt includes cache:** with a positive second-turn `cache_read`, compare + first input with second input + cache read + cache write within 2%. Reports + already subtract cache; if the first turn also hits/writes cache, restore its + cache counts before comparing too. A second total near first minus cache read + suggests the table wrongly subtracted an already cache-exclusive prompt count. + No hit, missing inputs, or no single completed primary report is **UNTESTED**. +- **Reasoning in output:** **FAIL** if reasoning exceeds output; otherwise + compare output minus reasoning with OpenCode's stored visible output within + **±1 token** when a single stored assistant output count is available. OpenCode's + AI SDK splits reasoning out of that stored output. Otherwise, **PASS** if the + difference is between half and twice visible reply characters/4, or within + **±8 tokens** of that estimate to avoid short-reply rounding failures. An empty + reply cannot pass the character fallback. Missing reasoning is **UNTESTED**. + The character fallback is heuristic, not a tokenizer or provider-contract proof. +- **Usage present:** every completed report must contain measured input and + output (including explicit zero). Missing fields are **FAIL**; no completed + reports is **UNTESTED**. The summary lists providers with missing usage. + +Beside each turn's report counts, the mode prints OpenCode's stored assistant +token counters and context size (stored input + cache.read + cache.write), +session id and model/variant. Stored visible output is used **only for the +reasoning verdict**; all other host-normalized totals remain diagnostic. +Missing reports or unavailable stored tokens are +also flagged in the ordinary smoke summary. Tables go to stdout and the usual +`results-opencode2-.md` next to the scratch home. + +**Rough cost per provider:** two long prompts contribute about **6,000 user +input tokens**, plus the host's system/tool prefix on each of two or three +requests; the reasoning prompt itself is short. Ordinary long replies have a +16-token cap, or 1,024 when the selected model needs reasoning room or is also +selected for the reasoning probe. With the +optional reasoning probe the total output ceiling is therefore usually +**1,056 tokens**, at worst **3,072**. Cache hits can reduce billed input, not +request count. Dollar cost depends on the catalog-selected models/provider +plan; all-provider mode multiplies this budget by the served providers. There +are at most three physical sends per provider, not three successful retries. + ## 0. Once, before the first provider 1. Install the published packages, or a packed build of this branch: diff --git a/packages/core/src/billing.ts b/packages/core/src/billing.ts new file mode 100644 index 0000000..934934a --- /dev/null +++ b/packages/core/src/billing.ts @@ -0,0 +1,350 @@ +import { randomUUID } from 'node:crypto' +import type { ClaustrumScopedAttempt } from '@cortexkit/common-auth/claustrum' +import { watchServerSentEvents } from '@cortexkit/common-auth/opencode2' +import { + type BillingRequestKind, + type BillingUsage, + reportRouting, + sessionCachePolicy, +} from '@cortexkit/common-auth/routing-accessor' +import type { ProviderEntry } from './providers.js' + +const object = (value: unknown): value is Record => + value !== null && typeof value === 'object' && !Array.isArray(value) + +function count( + value: unknown, + paths: readonly string[] = [], +): number | undefined { + for (const path of paths) { + let cursor = value + for (const key of path.split('.')) + cursor = object(cursor) ? cursor[key] : undefined + if ( + typeof cursor === 'number' && + Number.isSafeInteger(cursor) && + cursor >= 0 + ) + return cursor + } + return undefined +} + +interface RawBillingUsage { + counts: BillingUsage + opencode: boolean +} + +/** Retain measured totals without deriving values that later chunks can change. */ +function rawBillingUsage( + entry: ProviderEntry, + value: unknown, +): RawBillingUsage | undefined { + const semantics = entry.billing + if (!semantics) return undefined + const usage: BillingUsage = {} + const tokens = + object(value) && object(value.usage) && object(value.usage.opencode) + ? value.usage.opencode.tokens + : undefined + const opencode = !!semantics.opencodeTokens && object(tokens) + if (opencode) { + for (const [field, path] of [ + ['input', 'input'], + ['output', 'output'], + ['cacheRead', 'cache_read'], + ['cacheWrite5m', 'cache_write_5m'], + ['cacheWrite1h', 'cache_write_1h'], + ['reasoning', 'reasoning'], + ] as const) { + const measured = count(tokens, [path]) + if (measured !== undefined) usage[field] = measured + } + } else { + for (const field of [ + 'input', + 'output', + 'cacheRead', + 'cacheWrite', + 'reasoning', + ] as const) { + const measured = count(value, semantics[field]) + if (measured !== undefined) usage[field] = measured + } + } + return Object.keys(usage).length ? { counts: usage, opencode } : undefined +} + +function finalizedUsage( + entry: ProviderEntry, + counts: BillingUsage | undefined, + opencode: boolean, +): BillingUsage | undefined { + if (!counts) return undefined + const usage = { ...counts } + if (opencode) { + if (usage.cacheWrite5m !== undefined || usage.cacheWrite1h !== undefined) + usage.cacheWrite = (usage.cacheWrite5m ?? 0) + (usage.cacheWrite1h ?? 0) + usage.reasoningIncludedInOutput = true + } else { + if (usage.input !== undefined && entry.billing?.promptIncludesCache) + usage.input = Math.max( + 0, + usage.input - (usage.cacheRead ?? 0) - (usage.cacheWrite ?? 0), + ) + if (entry.billing?.reasoningIncludedInOutput !== undefined) + usage.reasoningIncludedInOutput = entry.billing.reasoningIncludedInOutput + } + return usage +} + +/** Read only documented fields; absent counters must not become measured zeroes. */ +export function billingUsage( + entry: ProviderEntry, + value: unknown, +): BillingUsage | undefined { + const raw = rawBillingUsage(entry, value) + return raw ? finalizedUsage(entry, raw.counts, raw.opencode) : undefined +} + +/** The accessor can be installed or released after the plugin starts. */ +export function hasBillingAccessor(): boolean { + try { + return ( + typeof Reflect.get(globalThis, Symbol.for('cortexkit.routing')) === + 'function' + ) + } catch { + return false + } +} + +/** One receipt and one UUID for one physical send, including redirects and retries. */ +export class BillingAttempt { + private readonly attemptId = randomUUID() + private reported = false + private completed = false + private failed = false + private outputStarted = false + private responseId?: string + private rawUsage?: BillingUsage + private rawOpenCodeUsage?: BillingUsage + private status?: number + private ended = false + + constructor( + private readonly entry: ProviderEntry, + private readonly provider: string, + private readonly sessionId: string, + private readonly requestKind: BillingRequestKind, + private readonly model: string, + private readonly receipt: ClaustrumScopedAttempt, + private readonly attempt?: number, + private readonly pi = false, + ) {} + + event(value: unknown): void { + if (!object(value)) return + if ( + value.type === 'error' || + value.type === 'response.failed' || + value.error + ) + this.failed = true + if ( + value.type === 'response.completed' || + value.type === 'response.incomplete' + ) + this.completed = true + if (typeof value.type === 'string' && value.type.endsWith('.delta')) + this.outputStarted = true + if (Array.isArray(value.choices)) { + for (const choice of value.choices) { + if (!object(choice)) continue + if (choice.finish_reason != null) this.completed = true + if ( + object(choice.delta) && + Object.keys(choice.delta).some((key) => key !== 'role') + ) + this.outputStarted = true + if (object(choice.message)) this.completed = true + } + } + const response = object(value.response) ? value.response : value + if (!this.pi && typeof response.id === 'string') + this.responseId = response.id + const raw = rawBillingUsage(this.entry, response) + if (raw?.opencode) { + // OpenCode's own input excludes cache already; do not mix it with + // generic prompt totals, even if the namespaces arrive in different chunks. + this.rawOpenCodeUsage = { ...this.rawOpenCodeUsage, ...raw.counts } + } else if (raw) { + this.rawUsage = { ...this.rawUsage, ...raw.counts } + } + } + + bodyEnd(ended: boolean, status: number): void { + this.status = status + this.ended = ended + // Pi's stored responseId and stop reason arrive after its driver consumes + // the successful body. Failed sends still settle before a retry starts. + if (!this.pi || status < 200 || status >= 300) this.end() + } + + cancel(status: number): void { + // Hosts may stop reading at the protocol terminal marker without reading + // byte EOF. Cancellation before that marker is still a cut response. + this.bodyEnd(this.completed, status) + } + + json(value: unknown): void { + this.event(value) + if (object(value) && !value.error) this.completed = true + } + + done(): void { + this.completed = true + } + + piEnd(responseId?: string, stopReason?: string): void { + // Use exactly the identifier Pi stores, not a competing wire identifier. + this.responseId = responseId + if (stopReason === 'error') this.failed = true + if (stopReason === 'aborted') this.ended = false + else if (stopReason && stopReason !== 'error') { + this.completed = true + this.ended = true + } + this.end() + } + + end( + outputStarted = this.outputStarted, + ended = this.ended, + status = this.status, + ): void { + if (this.reported) return + this.reported = true + const warming = sessionCachePolicy(this.sessionId)?.keep_warm + const implicit = this.entry.billing?.implicitCache + const usage = finalizedUsage( + this.entry, + this.rawOpenCodeUsage ?? this.rawUsage, + this.rawOpenCodeUsage !== undefined, + ) + reportRouting({ + sessionId: this.sessionId, + kind: 'billing', + attemptId: this.attemptId, + requestKind: this.requestKind, + credentialId: this.receipt.credentialId, + ...(this.receipt.accountIdentity + ? { accountId: this.receipt.accountIdentity } + : {}), + provider: this.provider, + model: this.model, + outcome: + this.failed || (status !== undefined && (status < 200 || status >= 300)) + ? 'failed' + : this.completed && ended + ? 'completed' + : outputStarted + ? 'cut' + : 'failed', + ...(status !== undefined ? { status } : {}), + ...(this.attempt !== undefined ? { attempt: this.attempt } : {}), + ...(this.responseId ? { providerResponseId: this.responseId } : {}), + ...(this.entry.chargeBasis + ? { chargeBasis: this.entry.chargeBasis } + : {}), + ...(usage ? { usage } : {}), + ...(implicit || (warming && warming !== 'off') + ? { + cache: { + ...(implicit ? { implicit } : {}), + ...(warming && warming !== 'off' + ? { warming: { on: false, stopReason: 'unsupported' } } + : {}), + }, + } + : {}), + }) + } +} + +/** Billing is strictly opt-in; do not even acquire a reader without an attempt. */ +export function observeBillingResponse( + record: BillingAttempt | undefined, + response: Response, +): Response { + if (!record) return response + if (!response.body) { + record.bodyEnd(true, response.status) + return response + } + const sse = /^text\/event-stream(?:\s*;|$)/i.test( + response.headers.get('content-type') ?? '', + ) + const decoder = new TextDecoder() + let json = '' + let oversized = false + const body = sse + ? response.body.pipeThrough( + watchServerSentEvents( + (event) => { + if (event.data === '[DONE]') record.done() + else { + try { + record.event(JSON.parse(event.data)) + } catch {} + } + }, + () => {}, + ), + ) + : response.body + const reader = body.getReader() + return new Response( + new ReadableStream( + { + async pull(controller) { + try { + const next = await reader.read() + if (next.done) { + if (!sse && !oversized) { + try { + record.json(JSON.parse(json + decoder.decode())) + } catch {} + } + record.bodyEnd(true, response.status) + controller.close() + } else { + if (!sse && !oversized) { + json += decoder.decode(next.value, { stream: true }) + // Accounting must not retain an unbounded non-streaming response. + if (json.length > 1024 * 1024) { + oversized = true + json = '' + } + } + controller.enqueue(next.value) + } + } catch (error) { + record.bodyEnd(false, response.status) + controller.error(error) + } + }, + cancel(reason) { + record.cancel(response.status) + return reader.cancel(reason) + }, + }, + { highWaterMark: 0 }, + ), + { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }, + ) +} diff --git a/packages/core/src/fetch.ts b/packages/core/src/fetch.ts index 0949c69..c9c9510 100644 --- a/packages/core/src/fetch.ts +++ b/packages/core/src/fetch.ts @@ -1,4 +1,5 @@ import { randomUUID } from 'node:crypto' +import type { ClaustrumScopedAttempt } from '@cortexkit/common-auth/claustrum' import { createRequestHistory, decideAfterLimit, @@ -8,6 +9,7 @@ import { statusVaultOrder, } from '@cortexkit/common-auth/limits' import { hasBalanceRule } from './balance.js' +import { type BillingAttempt, observeBillingResponse } from './billing.js' import { limitFromReply, limitRefusal } from './limits.js' import { ClaustrumAuthError, @@ -30,6 +32,7 @@ export interface FetchOptions extends ServingOptions { // The host supplies its own effective-key resolution, including environment // case rules and config precedence. The check is repeated on every request. effectiveCredential?: () => boolean | Promise + billing?: (receipt: ClaustrumScopedAttempt) => BillingAttempt | undefined } async function bufferBody(request: Request): Promise { @@ -100,6 +103,9 @@ export function createFetch(options: FetchOptions): CoreFetch { const tried = new Set() const rejected = new Set() const secrets: string[] = [] + let redirectBilling: + | { receipt: ClaustrumScopedAttempt; status: number } + | undefined let pending: | { response: Response @@ -257,7 +263,26 @@ export function createFetch(options: FetchOptions): CoreFetch { credentials: template.credentials, cache: template.cache, }) - return options.fetch(outbound) + let billing: BillingAttempt | undefined + try { + const reply = await options.fetch(outbound) + if ( + options.billing && + [301, 302, 303, 307, 308].includes(reply.status) && + reply.headers.has('location') + ) { + // A followed redirect is a routing hop, not a billable model + // call. Wait until the redirect checks decide to return it. + redirectBilling = { receipt, status: reply.status } + return reply + } + billing = options.billing?.(receipt) + return observeBillingResponse(billing, reply) + } catch (error) { + const failedBilling = billing ?? options.billing?.(receipt) + failedBilling?.end(false, false) + throw error + } }, { site: 'claustrum-auth', signal }, ) @@ -281,6 +306,7 @@ export function createFetch(options: FetchOptions): CoreFetch { await response.body?.cancel().catch(() => {}) throw new ClaustrumAuthError('redirect-limit') } + redirectBilling = undefined await response.body?.cancel().catch(() => {}) if ( (response.status === 303 && method !== 'HEAD') || @@ -291,6 +317,11 @@ export function createFetch(options: FetchOptions): CoreFetch { } target = next } + if (redirectBilling) { + const billing = options.billing?.(redirectBilling.receipt) + redirectBilling = undefined + response = observeBillingResponse(billing, response) + } const parsed = await limitFromReply({ account: observation?.binding.account ?? initial.account, service: options.entry.service, @@ -334,6 +365,10 @@ export function createFetch(options: FetchOptions): CoreFetch { } } } catch (error) { + if (redirectBilling) + options + .billing?.(redirectBilling.receipt) + ?.end(false, false, redirectBilling.status) await pending?.response.body?.cancel().catch(() => {}) throw redactError(error, secrets) } diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 6120da0..a592342 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -9,6 +9,12 @@ export { transitionActivation, } from './activation.js' export { hasBalanceRule, isBalanceFailure } from './balance.js' +export { + BillingAttempt, + billingUsage, + hasBillingAccessor, + observeBillingResponse, +} from './billing.js' export type { CoreFetch, FetchOptions } from './fetch.js' export { createFetch, maxBodyBytes } from './fetch.js' export { hasVaultHost } from './host.js' diff --git a/packages/core/src/providers.ts b/packages/core/src/providers.ts index 1d16d23..ef1cea1 100644 --- a/packages/core/src/providers.ts +++ b/packages/core/src/providers.ts @@ -5,6 +5,33 @@ import type { export type Host = 'opencode1' | 'opencode2' | 'pi' export type ProviderClass = 'A' | 'B' | 'C' | 'D' +export interface BillingSemantics { + input: readonly string[] + output: readonly string[] + cacheRead?: readonly string[] + cacheWrite?: readonly string[] + reasoning?: readonly string[] + promptIncludesCache: boolean + reasoningIncludedInOutput?: boolean + implicitCache?: true + opencodeTokens?: true +} +// Standard OpenAI field meanings are assumptions where a provider only names +// the fields. Provider-specific exceptions and sources: docs/billing-reports.md. +const openAIUsage = { + input: ['usage.prompt_tokens', 'usage.input_tokens'], + output: ['usage.completion_tokens', 'usage.output_tokens'], + cacheRead: [ + 'usage.prompt_tokens_details.cached_tokens', + 'usage.input_tokens_details.cached_tokens', + ], + reasoning: [ + 'usage.completion_tokens_details.reasoning_tokens', + 'usage.output_tokens_details.reasoning_tokens', + ], + promptIncludesCache: true, + reasoningIncludedInOutput: true, +} satisfies BillingSemantics export interface ProviderEntry { service: string // The fleet's provider ids for this service: the vocabulary vault rows carry @@ -24,6 +51,8 @@ export interface ProviderEntry { // handshake of the provider is refused before any vault read. handshakeOrigins?: readonly string[] accountPath?: readonly string[] + chargeBasis?: 'subscription_included' | 'metered_api' + billing?: BillingSemantics } export interface HostFacts { host: Host @@ -52,6 +81,9 @@ export interface HostFacts { export const supportedProviders: readonly ProviderEntry[] = [ { service: 'cerebras', + // One key can use trial, PayGo, or Code Pro: https://www.cerebras.ai/pricing and https://www.cerebras.ai/code + // Automatic caching: https://inference-docs.cerebras.ai/capabilities/prompt-caching + billing: { ...openAIUsage, implicitCache: true }, ids: ['cerebras'], hosts: { opencode1: 'cerebras', opencode2: 'cerebras', pi: 'cerebras' }, class: 'A', @@ -66,6 +98,9 @@ export const supportedProviders: readonly ProviderEntry[] = [ // through this exact credential id instead, only while the row has no // provider ids. service: 'cline-pass', + // Flat ClinePass key: https://docs.cline.bot/getting-started/clinepass.md + chargeBasis: 'subscription_included', + billing: { ...openAIUsage }, ids: ['cline-pass'], credentialIds: ['apikey:cline-pass'], hosts: { opencode1: 'cline-pass', opencode2: 'cline-pass' }, @@ -74,6 +109,14 @@ export const supportedProviders: readonly ProviderEntry[] = [ }, { service: 'deepseek', + // Pay-per-token API keys: https://api-docs.deepseek.com/quick_start/pricing + chargeBasis: 'metered_api', + // Automatic context caching: https://api-docs.deepseek.com/guides/kv_cache + billing: { + ...openAIUsage, + cacheRead: ['usage.prompt_cache_hit_tokens', ...openAIUsage.cacheRead], + implicitCache: true, + }, ids: ['deepseek'], hosts: { opencode1: 'deepseek', opencode2: 'deepseek', pi: 'deepseek' }, class: 'A', @@ -83,6 +126,17 @@ export const supportedProviders: readonly ProviderEntry[] = [ // The fleet id is `fireworks-ai`; OpenCode's id is `fireworks-ai` and // Pi's is `fireworks`. service: 'fireworks', + // Token-priced API: https://fireworks.ai/pricing + chargeBasis: 'metered_api', + // Automatic prefix caching: https://docs.fireworks.ai/guides/prompt-caching + billing: { + ...openAIUsage, + cacheRead: [ + 'perf_metrics.cached-prompt-tokens', + ...openAIUsage.cacheRead, + ], + implicitCache: true, + }, ids: ['fireworks-ai'], hosts: { opencode1: 'fireworks-ai', @@ -94,6 +148,13 @@ export const supportedProviders: readonly ProviderEntry[] = [ }, { service: 'inception', + // Token-priced API: https://www.inceptionlabs.ai/pricing + chargeBasis: 'metered_api', + billing: { + ...openAIUsage, + cacheRead: ['usage.cached_input_tokens', ...openAIUsage.cacheRead], + reasoning: ['usage.reasoning_tokens', ...openAIUsage.reasoning], + }, ids: ['inception'], hosts: { opencode1: 'inception', opencode2: 'inception' }, class: 'A', @@ -103,6 +164,14 @@ export const supportedProviders: readonly ProviderEntry[] = [ // OpenCode's catalog uses the fleet's global-plan id; Pi calls the same // service `kimi-coding`. Both documented regional API origins are exact. service: 'kimi-coding', + // Coding membership key: https://www.kimi.com/code/docs/en/ + chargeBasis: 'subscription_included', + // Chat caching semantics are assumed for Code: https://platform.kimi.ai/docs/guide/context-caching + billing: { + ...openAIUsage, + cacheWrite: ['usage.prompt_tokens_details.cache_write_tokens'], + implicitCache: true, + }, ids: ['kimi-code-plan-global'], hosts: { opencode1: 'kimi-code-plan-global', @@ -115,6 +184,10 @@ export const supportedProviders: readonly ProviderEntry[] = [ { // OpenCode 1 sends x-api-key; Pi does not register this provider. service: 'minimax-coding-plan', + // Coding-plan key: https://platform.minimax.io/docs/coding-plan/intro + chargeBasis: 'subscription_included', + // Automatic repeated-context caching: https://platform.minimax.io/docs/api-reference/text-prompt-caching.md + billing: { ...openAIUsage, implicitCache: true }, ids: ['minimax-coding-plan'], hosts: { opencode2: 'minimax-coding-plan' }, class: 'A', @@ -122,6 +195,8 @@ export const supportedProviders: readonly ProviderEntry[] = [ }, { service: 'ollama-cloud', + // The same key spans plan credits and PayGo: https://ollama.com/pricing + billing: { ...openAIUsage }, ids: ['ollama-cloud'], hosts: { opencode1: 'ollama-cloud', opencode2: 'ollama-cloud' }, class: 'A', @@ -129,6 +204,12 @@ export const supportedProviders: readonly ProviderEntry[] = [ }, { service: 'openrouter', + // Credit-funded API: https://openrouter.ai/pricing + chargeBasis: 'metered_api', + billing: { + ...openAIUsage, + cacheWrite: ['usage.prompt_tokens_details.cache_write_tokens'], + }, ids: ['openrouter'], hosts: { opencode1: 'openrouter', @@ -140,6 +221,8 @@ export const supportedProviders: readonly ProviderEntry[] = [ }, { service: 'synthetic', + // A key spans subscription credits and usage-based top-ups: https://synthetic.new/pricing + billing: { ...openAIUsage }, ids: ['synthetic'], hosts: { opencode1: 'synthetic', opencode2: 'synthetic' }, class: 'A', @@ -150,6 +233,13 @@ export const supportedProviders: readonly ProviderEntry[] = [ // api, which does not take a caller fetch, so the plugin's fetch never // sees a typesafe request there. service: 'typesafe', + // Input is billable API usage: https://docs.typesafe.ai/api/system-one.md + chargeBasis: 'metered_api', + billing: { + input: ['usage.input_tokens'], + output: ['usage.output_tokens'], + promptIncludesCache: false, + }, ids: ['typesafe'], hosts: { opencode1: 'typesafe', opencode2: 'typesafe' }, class: 'A', @@ -159,6 +249,10 @@ export const supportedProviders: readonly ProviderEntry[] = [ // An OAuth row: the vault serves a minted inference key with an expiry, // which custody requires of every OAuth read (docs/providers.md, "Meta"). service: 'meta', + // Token-priced Model API (not Muse Code): https://dev.meta.ai/docs/pricing-rate-limits + chargeBasis: 'metered_api', + // Automatic prefix caching: https://dev.meta.ai/docs/prompt-caching + billing: { ...openAIUsage, implicitCache: true }, ids: ['meta'], hosts: { opencode1: 'meta', opencode2: 'meta', pi: 'meta' }, class: 'A', @@ -171,6 +265,10 @@ export const supportedProviders: readonly ProviderEntry[] = [ // (docs/providers.md, "xAI on OpenCode 2"); the HTTP fallback the host // takes after a failed handshake is served like any other request. service: 'xai', + // Token-priced API: https://docs.x.ai/docs/models + chargeBasis: 'metered_api', + // Automatic matching-prefix cache: https://docs.x.ai/developers/advanced-api-usage/prompt-caching + billing: { ...openAIUsage, implicitCache: true }, ids: ['xai'], hosts: { opencode1: 'xai', opencode2: 'xai', pi: 'xai' }, class: 'A', @@ -183,6 +281,8 @@ export const supportedProviders: readonly ProviderEntry[] = [ // models send it in `x-api-key` or `x-goog-api-key` and are refused per // request (docs/providers.md, "OpenCode Go"). service: 'opencode-go', + // One key spans plan allowance and optional PayGo Extra Usage: https://opencode.ai/docs/go/ + billing: { ...openAIUsage, opencodeTokens: true }, ids: ['opencode-go'], hosts: { opencode1: 'opencode-go', @@ -198,6 +298,9 @@ export const supportedProviders: readonly ProviderEntry[] = [ // the request hooks (design section 5.2, class D). Per-model drivers as // opencode-go. service: 'opencode-zen', + // Prepaid token-priced key: https://opencode.ai/docs/zen/ + chargeBasis: 'metered_api', + billing: { ...openAIUsage, opencodeTokens: true }, ids: ['opencode'], hosts: { opencode1: 'opencode', pi: 'opencode' }, class: 'A', diff --git a/packages/core/test/billing-fixtures.ts b/packages/core/test/billing-fixtures.ts new file mode 100644 index 0000000..ee7f976 --- /dev/null +++ b/packages/core/test/billing-fixtures.ts @@ -0,0 +1,70 @@ +import { afterEach } from 'bun:test' + +const symbol = Symbol.for('cortexkit.routing') +const original = Object.getOwnPropertyDescriptor(globalThis, symbol) +afterEach(() => { + if (original) Object.defineProperty(globalThis, symbol, original) + else Reflect.deleteProperty(globalThis, symbol) +}) + +export function captureBilling(keepWarm = false) { + const reports: Record[] = [] + const sessions: string[] = [] + Object.defineProperty(globalThis, symbol, { + configurable: true, + value: (session: string) => { + sessions.push(session) + return { + report: (report: Record) => { + reports.push(report) + }, + cache: () => (keepWarm ? { keep_warm: 'always' } : null), + } + }, + }) + return { reports, sessions } +} + +// DeepSeek and Cerebras document these final-chunk fields. Cache writes are +// intentionally absent, so the expected report cannot silently invent zeroes. +export function chatChunks(provider: 'deepseek' | 'cerebras') { + return [ + { + id: 'reply-1', + model: 'served-model', + choices: [{ delta: { content: 'hello' }, finish_reason: null }], + }, + { + id: 'reply-1', + model: 'served-model', + choices: [{ delta: {}, finish_reason: 'stop' }], + usage: { + prompt_tokens: 100, + completion_tokens: 20, + ...(provider === 'deepseek' + ? { prompt_cache_hit_tokens: 40 } + : { prompt_tokens_details: { cached_tokens: 40 } }), + completion_tokens_details: { reasoning_tokens: 5 }, + }, + }, + ] +} + +export function chatResponse(provider: 'deepseek' | 'cerebras') { + return new Response( + `${chatChunks(provider) + .map((value) => `data: ${JSON.stringify(value)}\n\n`) + .join('')}data: [DONE]\n\n`, + { + headers: { 'content-type': 'text/event-stream' }, + }, + ) +} + +export const expectedUsage = { + input: 60, + output: 20, + cache_read: 40, + reasoning: 5, + reasoning_included_in_output: true, +} diff --git a/packages/core/test/billing.test.ts b/packages/core/test/billing.test.ts new file mode 100644 index 0000000..0d185dd --- /dev/null +++ b/packages/core/test/billing.test.ts @@ -0,0 +1,249 @@ +import { expect, test } from 'bun:test' +import { + BillingAttempt, + billingUsage, + observeBillingResponse, +} from '../src/billing.ts' +import { supportedProviders } from '../src/providers.ts' +import { createFetch, fixtureEntries } from '../test-entry.ts' +import { captureBilling } from './billing-fixtures.ts' +import { bearer, harness, placeholder } from './fakes.ts' + +test('disabled billing observer preserves response identity without taking a reader or parsing JSON', () => { + const response = new Response('not JSON') + Object.defineProperty(response, 'body', { + get() { + throw new Error('billing took a reader') + }, + }) + expect(observeBillingResponse(undefined, response)).toBe(response) + expect(response.bodyUsed).toBe(false) +}) + +test.each(supportedProviders)( + '$service has explicit, provider-true usage semantics', + (entry) => { + const result = billingUsage(entry, { + usage: { + prompt_tokens: 100, + input_tokens: 90, + completion_tokens: 20, + output_tokens: 15, + prompt_cache_hit_tokens: 40, + cached_input_tokens: 30, + prompt_tokens_details: { cached_tokens: 25, cache_write_tokens: 10 }, + input_tokens_details: { cached_tokens: 20 }, + completion_tokens_details: { reasoning_tokens: 5 }, + reasoning_tokens: 7, + }, + perf_metrics: { 'cached-prompt-tokens': 35 }, + }) + const expected: Record = + entry.service === 'typesafe' + ? { input: 90, output: 15 } + : { + input: 75, + output: 20, + cacheRead: 25, + reasoning: 5, + reasoningIncludedInOutput: true, + } + // These values are deliberately independent of the table's extraction code. + if (entry.service === 'deepseek') + Object.assign(expected, { input: 60, cacheRead: 40 }) + if (entry.service === 'fireworks') + Object.assign(expected, { input: 65, cacheRead: 35 }) + if (entry.service === 'inception') + Object.assign(expected, { input: 70, cacheRead: 30, reasoning: 7 }) + if (entry.service === 'openrouter' || entry.service === 'kimi-coding') + Object.assign(expected, { input: 65, cacheWrite: 10 }) + expect(result).toEqual(expected) + expect(billingUsage(entry, { usage: {} })).toBeUndefined() + }, +) + +test('OpenCode token accounting wins over generic totals and preserves write lifetimes and measured zero', () => { + const entry = supportedProviders.find((row) => row.service === 'opencode-go') + if (!entry) throw new Error('missing Go') + expect( + billingUsage(entry, { + usage: { + prompt_tokens: 999, + completion_tokens: 999, + opencode: { + tokens: { + input: 10, + cache_read: 20, + cache_write_5m: 3, + cache_write_1h: 4, + output: 8, + reasoning: 0, + }, + }, + }, + }), + ).toEqual({ + input: 10, + cacheRead: 20, + cacheWrite: 7, + cacheWrite5m: 3, + cacheWrite1h: 4, + output: 8, + reasoning: 0, + reasoningIncludedInOutput: true, + }) +}) + +test('a prompt total with missing cache metrics remains input, not an invented cache zero', () => { + const entry = supportedProviders.find((row) => row.service === 'fireworks') + if (!entry) throw new Error('missing fireworks') + expect( + billingUsage(entry, { + usage: { prompt_tokens: 100, completion_tokens: 0 }, + }), + ).toEqual({ input: 100, output: 0, reasoningIncludedInOutput: true }) +}) + +test('TypeSafe non-streaming JSON reports at body end and ignores undocumented counters', async () => { + const { reports } = captureBilling(true) + const entry = supportedProviders.find((row) => row.service === 'typesafe') + if (!entry) throw new Error('missing typesafe') + const record = new BillingAttempt( + entry, + 'typesafe', + 'session', + 'generate', + 'system-one', + { + credentialId: 'apikey:typesafe', + credentialType: 'api_key', + accountIdentitySource: 'none', + accessToken: 'secret', + recordVersion: 1, + expiresAtMs: null, + }, + ) + const response = observeBillingResponse( + record, + Response.json({ + usage: { input_tokens: 12, output_tokens: 3, cached_tokens: 999 }, + }), + ) + expect(reports).toEqual([]) + await response.json() + expect(reports).toHaveLength(1) + expect(reports[0]).toMatchObject({ + outcome: 'completed', + request_kind: 'generate', + usage: { input: 12, output: 3 }, + cache: { warming: { on: false, stop_reason: 'unsupported' } }, + }) +}) + +function streamAttempt(service: string) { + const entry = supportedProviders.find((row) => row.service === service) + if (!entry) throw new Error('missing provider') + return new BillingAttempt( + entry, + service, + 'session', + 'primary', + 'sent-model', + { + credentialId: `apikey:${service}`, + credentialType: 'api_key', + accountIdentitySource: 'none', + accessToken: 'secret', + recordVersion: 1, + expiresAtMs: null, + }, + ) +} + +test('split prompt and cached counts derive uncached input once at report time', () => { + const { reports } = captureBilling() + const record = streamAttempt('cerebras') + record.event({ usage: { prompt_tokens: 100, completion_tokens: 20 } }) + record.event({ usage: { prompt_tokens_details: { cached_tokens: 40 } } }) + record.done() + record.bodyEnd(true, 200) + expect(reports).toHaveLength(1) + expect(reports[0]?.usage).toEqual({ + input: 60, + output: 20, + cache_read: 40, + reasoning_included_in_output: true, + }) +}) + +test('split OpenCode write lifetimes are summed from their latest raw counts', () => { + const { reports } = captureBilling() + const record = streamAttempt('opencode-go') + record.event({ + usage: { opencode: { tokens: { input: 10, cache_write_5m: 3 } } }, + }) + record.event({ + usage: { opencode: { tokens: { cache_write_1h: 4, output: 8 } } }, + }) + record.done() + record.bodyEnd(true, 200) + expect(reports).toHaveLength(1) + expect(reports[0]?.usage).toEqual({ + input: 10, + output: 8, + cache_write: 7, + cache_write_5m: 3, + cache_write_1h: 4, + reasoning_included_in_output: true, + }) +}) + +test.each(['terminal-redirect', 'transport-error', 'redirect-limit'] as const)( + 'core still reports a send ending in %s', + async (mode) => { + const { reports } = captureBilling() + const h = await harness() + const entry = fixtureEntries[0] + if (!entry) throw new Error('missing fixture') + const fetch = createFetch({ + entry, + placeholder, + consumer: h.consumer, + billing: (receipt) => + new BillingAttempt( + entry, + 'fixture-a', + 'session', + 'primary', + 'sent-model', + receipt, + ), + fetch: async () => { + if (mode === 'transport-error') throw new Error('transport failed') + return new Response('redirect', { + status: 307, + headers: { + location: + mode === 'terminal-redirect' + ? 'https://other.invalid/next' + : '/again', + }, + }) + }, + }) + const reply = fetch('https://fixture.invalid/start', { headers: bearer }) + if (mode === 'terminal-redirect') await (await reply).text() + else + await expect(reply).rejects.toThrow( + mode === 'transport-error' ? 'transport failed' : 'redirect-limit', + ) + expect(reports).toHaveLength(1) + expect(reports[0]).toMatchObject({ + outcome: 'failed', + model: 'sent-model', + ...(mode !== 'transport-error' ? { status: 307 } : {}), + }) + if (mode === 'transport-error') + expect(reports[0]).not.toHaveProperty('status') + }, +) diff --git a/packages/opencode/src/v2/adapter.ts b/packages/opencode/src/v2/adapter.ts index db5dabb..7d1b2f4 100644 --- a/packages/opencode/src/v2/adapter.ts +++ b/packages/opencode/src/v2/adapter.ts @@ -1,15 +1,18 @@ import { randomUUID } from 'node:crypto' import { type AccountLimits, + BillingAttempt, ClaustrumAuthError, checkHandshakeOrigin, hasBalanceRule, + hasBillingAccessor, isBalanceFailure, isLocalLimitRefusal, type LimitObservation, limitFromReply, localLimitRefusal, localLimitRetry, + observeBillingResponse, type ProviderEntry, redactError, redactResponse, @@ -63,6 +66,8 @@ export interface VaultAttempt { handshakeURL?: string /** A provider status an error frame of this attempt carried. */ frameStatus?: number + billing?: BillingAttempt + billingAttempt?: number } /** Identifies a session's socket: the host keeps one per session and kind. */ @@ -290,6 +295,7 @@ export function createVaultAdapter( { until: number; cause?: 'rate-limit' | 'payment-balance' } >() const placeholder = placeholderSecret(options.provider) + let billingAttempts: Map | undefined const refuse = (error: unknown) => { if ( error instanceof ClaustrumAuthError && @@ -303,6 +309,11 @@ export function createVaultAdapter( ?.length ? ({ transport, data, attempt }) => { if (transport !== 'ws') return undefined + if (attempt.data?.billing) { + try { + attempt.data.billing.event(JSON.parse(data)) + } catch {} + } const verdict = inspectResponsesFrame( data, attempt.data, @@ -344,6 +355,17 @@ export function createVaultAdapter( return { providerID: options.provider, async chooseAccount(input) { + if (hasBillingAccessor() && input.hostAttempt !== undefined) { + billingAttempts ??= new Map() + billingAttempts.set( + socketKey(input.sessionID, input.kind), + input.hostAttempt, + ) + if (billingAttempts.size > 256) { + const oldest = billingAttempts.keys().next().value + if (oldest !== undefined) billingAttempts.delete(oldest) + } + } // A provider whose driver never sends its key in Authorization is // refused the same way for every account, before any vault read. if (options.entry.class === 'C' || options.entry.class === 'D') @@ -447,6 +469,7 @@ export function createVaultAdapter( async accountHeaders({ sessionID, kind, + modelID, accountId, }): Promise> { // HTTP: the vault is read in `rewriteRequest`, after the origin check. @@ -454,9 +477,14 @@ export function createVaultAdapter( // so the plugin's handshake hook checked the origin before this call // and left the URL here; only then is the vault read now. const key = socketKey(sessionID, kind) + const hostAttempt = billingAttempts?.get(key) + billingAttempts?.delete(key) const url = options.handshakes?.pending.get(key) if (url === undefined) - return { headers: {}, attempt: {} satisfies VaultAttempt } + return { + headers: {}, + attempt: { billingAttempt: hostAttempt } satisfies VaultAttempt, + } options.handshakes?.pending.delete(key) // The handshake URL has no account segment the plugin could fill. if (options.entry.class !== 'A') @@ -483,6 +511,19 @@ export function createVaultAdapter( handshakeURL: url, limits, observation, + ...(hasBillingAccessor() + ? { + billing: new BillingAttempt( + options.entry, + options.provider, + sessionID, + kind, + modelID, + receipt, + hostAttempt, + ), + } + : {}), } satisfies VaultAttempt, } }, @@ -543,6 +584,16 @@ export function createVaultAdapter( holder.consumer = consumer holder.limits = served.limits holder.observation = served.observation + if (hasBillingAccessor()) + holder.billing = new BillingAttempt( + options.entry, + options.provider, + attempt.sessionID, + attempt.kind, + attempt.modelID, + served.receipt, + holder.billingAttempt, + ) } try { return await rebuild(request, served.url, served.headers) @@ -560,7 +611,10 @@ export function createVaultAdapter( async rewriteResponse({ response, attempt }) { const secret = attempt.data?.receipt?.accessToken if (!secret) return undefined - const rewritten = await redactResponse(response, [secret]) + const rewritten = observeBillingResponse( + attempt.data?.billing, + await redactResponse(response, [secret]), + ) return isLocalLimitRefusal(response) ? localLimitRefusal(rewritten, localLimitRetry(response)) : rewritten @@ -590,6 +644,12 @@ export function createVaultAdapter( ...(inspectEvent ? { inspectEvent } : {}), async onAttemptEnd(attempt, outcome) { const data = attempt.data + if (attempt.transport === 'ws' || outcome.status === undefined) + data?.billing?.end( + outcome.outputStarted, + outcome.error === undefined, + outcome.status ?? data.frameStatus, + ) const parsed = data?.limit const mark = markFromSignal(parsed, statusVaultOrder.markers.opencode2, { now: now(), diff --git a/packages/opencode/test/v2/billing.test.ts b/packages/opencode/test/v2/billing.test.ts new file mode 100644 index 0000000..aac00a1 --- /dev/null +++ b/packages/opencode/test/v2/billing.test.ts @@ -0,0 +1,305 @@ +import { expect, test } from 'bun:test' +import { placeholderSecret } from '@cortexkit/common-auth/opencode2' +import { + type ProviderEntry, + supportedProviders, +} from '../../../core/src/providers.ts' +import { + captureBilling, + chatChunks, + chatResponse, + expectedUsage, +} from '../../../core/test/billing-fixtures.ts' +import { vaultRowFor } from '../../../core/test/fakes.ts' +import { vaultMethodId } from '../../src/v2/setup.ts' +import { scope, startServer, trigger } from './fakes.ts' + +async function setup(provider: string, tableEntry?: ProviderEntry) { + const entry = + tableEntry ?? supportedProviders.find((row) => row.service === provider) + if (!entry) throw new Error('missing provider') + const { host } = await startServer({ + rows: [vaultRowFor(entry, { accountId: 'account-1' })], + catalog: [{ id: provider }], + table: [entry], + }) + const method = host.methods.get(`${provider}:${vaultMethodId}`) + if (!method) throw new Error('missing method') + await (await method.authorize({})).callback + const scoped = { + ...scope(provider), + model: { providerID: provider, id: 'sent-model' }, + } + const request = (key = placeholderSecret(provider)) => + new Request(`${entry.origins[0]}/v1/chat/completions`, { + method: 'POST', + headers: { authorization: `Bearer ${key}` }, + body: '{}', + }) + let sent = request() + const send = async (key?: string) => { + sent = ( + await trigger(host, 'http.request', { ...scoped, request: request(key) }) + ).request + } + const response = async (reply: Response) => + ( + await trigger(host, 'http.response', { + ...scoped, + request: sent, + response: reply, + }) + ).response + return { host, scoped, send, response } +} + +for (const provider of ['deepseek', 'cerebras'] as const) { + test(`${provider} vault HTTP billing reports the documented stream exactly once at body end`, async () => { + const { reports, sessions } = captureBilling(true) + const run = await setup(provider) + await run.send() + const response = await run.response(chatResponse(provider)) + expect(reports).toEqual([]) + await response.text() + expect(reports).toEqual([ + { + kind: 'billing', + attempt_id: expect.any(String), + request_kind: 'primary', + credential_id: `apikey:${provider}`, + account_id: 'account-1', + provider, + model: 'sent-model', + outcome: 'completed', + status: 200, + provider_response_id: 'reply-1', + ...(provider === 'deepseek' ? { charge_basis: 'metered_api' } : {}), + usage: expectedUsage, + cache: { + implicit: true, + warming: { on: false, stop_reason: 'unsupported' }, + }, + }, + ]) + expect(sessions.every((session) => session === run.scoped.sessionID)).toBe( + true, + ) + }) +} + +test('a reply model alias keeps one report at the sent model without usageByModel', async () => { + const { reports } = captureBilling() + const run = await setup('deepseek') + await run.send() + const response = new Response( + `${chatChunks('deepseek') + .map( + (chunk) => + `data: ${JSON.stringify({ ...chunk, model: 'deepseek/deepseek-chat-2026-01-01' })}\n\n`, + ) + .join('')}data: [DONE]\n\n`, + { headers: { 'content-type': 'text/event-stream' } }, + ) + await (await run.response(response)).text() + expect(reports).toHaveLength(1) + expect(reports[0]).toMatchObject({ + model: 'sent-model', + usage: expectedUsage, + }) + expect(reports[0]).not.toHaveProperty('usage_by_model') + expect(reports[0]).not.toHaveProperty('usageByModel') +}) + +test('without an accessor a served host request never looks up billing semantics', async () => { + Reflect.deleteProperty(globalThis, Symbol.for('cortexkit.routing')) + const base = supportedProviders.find((entry) => entry.service === 'deepseek') + if (!base?.billing) throw new Error('missing deepseek') + let parsed = 0 + const run = await setup('deepseek', { + ...base, + billing: new Proxy(base.billing, { + get(target, key, receiver) { + parsed++ + return Reflect.get(target, key, receiver) + }, + }), + }) + parsed = 0 + await run.send() + await (await run.response(chatResponse('deepseek'))).text() + expect(parsed).toBe(0) +}) + +test('a user-key host request sends no billing report and never parses usage', async () => { + const { reports } = captureBilling() + const base = supportedProviders.find((entry) => entry.service === 'deepseek') + if (!base?.billing) throw new Error('missing deepseek') + let parsed = 0 + const run = await setup('deepseek', { + ...base, + billing: new Proxy(base.billing, { + get(target, key, receiver) { + parsed++ + return Reflect.get(target, key, receiver) + }, + }), + }) + parsed = 0 + await run.send('user-key') + const reply = chatResponse('deepseek') + expect(await run.response(reply)).toBe(reply) + await reply.text() + expect(reports).toEqual([]) + expect(parsed).toBe(0) +}) + +test('a cut HTTP stream reports cut without invented usage', async () => { + const { reports } = captureBilling() + const run = await setup('deepseek') + await run.send() + const first = chatChunks('deepseek')[0] + const reply = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode(`data: ${JSON.stringify(first)}\n\n`), + ) + }, + }), + { + headers: { 'content-type': 'text/event-stream' }, + }, + ) + const reader = (await run.response(reply)).body?.getReader() + await reader?.read() + await reader?.cancel() + expect(reports).toHaveLength(1) + expect(reports[0]).toMatchObject({ + outcome: 'cut', + provider_response_id: 'reply-1', + }) + expect(reports[0]).not.toHaveProperty('usage') +}) + +for (const withUsage of [true, false]) { + test(`cancelling after finish_reason ${withUsage ? 'with usage' : 'before trailing usage'} reports completed`, async () => { + const { reports } = captureBilling() + const run = await setup('deepseek') + await run.send() + const chunks = chatChunks('deepseek') + const terminal = chunks[1] + if (!terminal) throw new Error('missing terminal fixture') + const bytes = [ + chunks[0], + { ...terminal, usage: withUsage ? terminal.usage : undefined }, + ] + .map((chunk) => `data: ${JSON.stringify(chunk)}\n\n`) + .join('') + let cancelled = false + let releaseTrailing: (() => void) | undefined + const reply = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(bytes)) + }, + pull(controller) { + return new Promise((resolve) => { + releaseTrailing = () => { + if (!cancelled) { + controller.enqueue( + new TextEncoder().encode( + `data: ${JSON.stringify({ ...terminal, choices: [] })}\n\ndata: [DONE]\n\n`, + ), + ) + controller.close() + } + resolve() + } + }) + }, + cancel() { + cancelled = true + releaseTrailing?.() + }, + }), + { headers: { 'content-type': 'text/event-stream' } }, + ) + const reader = (await run.response(reply)).body?.getReader() + if (!reader) throw new Error('missing body') + const decoder = new TextDecoder() + let consumed = '' + while (!consumed.includes('"finish_reason":"stop"')) { + const next = await reader.read() + expect(next.done).toBe(false) + consumed += decoder.decode(next.value, { stream: true }) + } + await reader?.cancel() + expect(cancelled).toBe(true) + expect(reports).toHaveLength(1) + expect(reports[0]).toMatchObject({ outcome: 'completed', status: 200 }) + if (withUsage) expect(reports[0]?.usage).toEqual(expectedUsage) + else expect(reports[0]).not.toHaveProperty('usage') + }) +} + +test('a retry reports each physical send with a fresh attemptId', async () => { + const { reports } = captureBilling() + const run = await setup('deepseek') + await run.send() + await (await run.response(new Response('busy', { status: 429 }))).text() + await trigger(run.host, 'retry', { + ...run.scoped, + attempt: 2, + error: { type: 'provider.error', message: 'busy', status: 429 }, + decision: { retry: true, delay: 0 }, + }) + await run.send() + await (await run.response(chatResponse('deepseek'))).text() + expect(reports).toHaveLength(2) + expect(reports.map((report) => report.outcome)).toEqual([ + 'failed', + 'completed', + ]) + expect(reports[1]?.attempt).toBe(2) + expect(reports[0]?.attempt_id).not.toBe(reports[1]?.attempt_id) +}) + +test('xAI WebSocket reports the completed Responses frame once', async () => { + const { reports } = captureBilling() + const run = await setup('xai') + await trigger(run.host, 'experimental.ws.handshake', { + ...run.scoped, + url: 'wss://api.x.ai/v1/responses', + headers: { authorization: `Bearer ${placeholderSecret('xai')}` }, + }) + const frame = JSON.stringify({ + type: 'response.completed', + response: { + id: 'xai-response', + model: 'sent-model', + usage: { + input_tokens: 100, + input_tokens_details: { cached_tokens: 40 }, + output_tokens: 20, + output_tokens_details: { reasoning_tokens: 5 }, + }, + }, + }) + await trigger(run.host, 'experimental.ws.receive', { ...run.scoped, frame }) + expect(reports).toEqual([ + { + kind: 'billing', + attempt_id: expect.any(String), + request_kind: 'primary', + credential_id: 'oauth:xai', + account_id: 'account-1', + provider: 'xai', + model: 'sent-model', + outcome: 'completed', + provider_response_id: 'xai-response', + charge_basis: 'metered_api', + usage: expectedUsage, + cache: { implicit: true }, + }, + ]) +}) diff --git a/packages/opencode/test/v2/e2e/billing-preload.ts b/packages/opencode/test/v2/e2e/billing-preload.ts new file mode 100644 index 0000000..af49c85 --- /dev/null +++ b/packages/opencode/test/v2/e2e/billing-preload.ts @@ -0,0 +1,12 @@ +import { appendFileSync } from 'node:fs' + +// Loaded before plugin setup, in the real host process rather than the runner. +const destination = process.env.CLAUSTRUM_AUTH_E2E_BILLING +if (destination) { + Reflect.set(globalThis, Symbol.for('cortexkit.routing'), () => ({ + report(payload: { kind: string }) { + if (payload.kind === 'billing') + appendFileSync(destination, `${JSON.stringify(payload)}\n`) + }, + })) +} diff --git a/packages/opencode/test/v2/e2e/opencode2.e2e.test.ts b/packages/opencode/test/v2/e2e/opencode2.e2e.test.ts index 51ba3a2..597c910 100644 --- a/packages/opencode/test/v2/e2e/opencode2.e2e.test.ts +++ b/packages/opencode/test/v2/e2e/opencode2.e2e.test.ts @@ -521,6 +521,48 @@ const onWire = (run: Scenario, needle: string) => ) describe.skipIf(!ENABLED)('claustrum-auth plugin on real OpenCode 2', () => { + test('billing: a real vault-served HTTP turn reports completed provider usage', async () => { + await scenario( + [{ id: 'apikey:cerebras', providerIds: ['cerebras'] }], + async (run) => { + expect((await run.login('cerebras')).exit).toBe(0) + expect((await run.turn('cerebras')).exit).toBe(0) + const reports = readFileSync( + path.join(run.root, 'billing.jsonl'), + 'utf8', + ) + .trim() + .split('\n') + .map((line) => JSON.parse(line) as Record) + const primary = reports.filter( + (report) => report.request_kind === 'primary', + ) + expect(primary).toHaveLength(1) + expect(primary[0]).toMatchObject({ + kind: 'billing', + attempt_id: expect.any(String), + credential_id: 'apikey:cerebras', + provider: 'cerebras', + model: 'mock-model', + outcome: 'completed', + status: 200, + provider_response_id: 'chatcmpl-mock', + usage: { input: 5, output: 2, reasoning_included_in_output: true }, + cache: { implicit: true }, + }) + expect(primary[0]?.usage).not.toHaveProperty('cache_read') + }, + { + beforeStart: async (run) => { + run.env.CLAUSTRUM_AUTH_E2E_REAL_ROWS = '1' + run.env.CLAUSTRUM_AUTH_E2E_BILLING = path.join( + run.root, + 'billing.jsonl', + ) + }, + }, + ) + }) beforeAll(async () => { scratch = await mkdtemp(path.join(tmpdir(), 'claustrum-auth-oc2-cli-')) const cliDir = REUSE_CLI_DIR ?? path.join(scratch, 'cli') diff --git a/packages/opencode/test/v2/e2e/plugin-entry.ts b/packages/opencode/test/v2/e2e/plugin-entry.ts index 1846d38..6c1e87c 100644 --- a/packages/opencode/test/v2/e2e/plugin-entry.ts +++ b/packages/opencode/test/v2/e2e/plugin-entry.ts @@ -10,6 +10,7 @@ // session event with its order, and what the host's retry hook was handed. // None of these records ever holds a served secret: only provider ids, // account ids, kinds, sessions, and whether a header was the placeholder. +import './billing-preload.ts' import { appendFileSync } from 'node:fs' import path from 'node:path' import { writeEnrollmentTokenFile } from '@cortexkit/claustrum-client' diff --git a/packages/pi/src/extension.ts b/packages/pi/src/extension.ts index f225399..cfbb4ed 100644 --- a/packages/pi/src/extension.ts +++ b/packages/pi/src/extension.ts @@ -3,6 +3,7 @@ import { type ActivationEffect, type ActivationEvent, type ActivationState, + BillingAttempt, ClaustrumAuthError, type CoreFetch, type CoreOptions, @@ -10,6 +11,7 @@ import { command, type createCore, createFetch, + hasBillingAccessor, initialActivation, type ProviderEntry, piEnrollment, @@ -330,13 +332,43 @@ class PiRuntime { if (options?.apiKey !== served.placeholder) return streamSimple(model, context, options) const inner = (options.fetch ?? globalThis.fetch) as CoreFetch - return streamSimple(model, context, { + if (!hasBillingAccessor()) + return streamSimple(model, context, { + ...options, + fetch: this.ownedFetch(served, inner) as typeof fetch, + }) + let current: BillingAttempt | undefined + let attempt = 0 + const stream = streamSimple(model, context, { ...options, - fetch: this.ownedFetch(served, inner) as typeof fetch, + fetch: this.ownedFetch(served, inner, (receipt) => { + current?.end() + if (!hasBillingAccessor()) return undefined + current = new BillingAttempt( + served.entry, + model.provider, + options.sessionId ?? this.#ctx.sessionManager.getSessionId(), + 'primary', + model.id, + receipt, + ++attempt, + true, + ) + return current + }) as typeof fetch, }) + void stream.result().then( + (message) => current?.piEnd(message.responseId, message.stopReason), + () => current?.end(), + ) + return stream } - ownedFetch(served: Served, inner: CoreFetch): CoreFetch { + ownedFetch( + served: Served, + inner: CoreFetch, + billing?: Parameters[0]['billing'], + ): CoreFetch { const serve = createFetch({ entry: served.entry, placeholder: served.placeholder, @@ -345,6 +377,7 @@ class PiRuntime { fetch: inner, rateLimitMarks: served.marks, limits: () => this.#core.limits(served.provider), + billing, }) return async (input, init) => { try { diff --git a/packages/pi/test/billing.test.ts b/packages/pi/test/billing.test.ts new file mode 100644 index 0000000..afbb39e --- /dev/null +++ b/packages/pi/test/billing.test.ts @@ -0,0 +1,249 @@ +import { expect, test } from 'bun:test' +import type { AssistantMessageEventStream } from '@earendil-works/pi-ai' +import type { + ExtensionAPI, + ExtensionContext, +} from '@earendil-works/pi-coding-agent' +import { supportedProviders } from '../../core/src/providers.ts' +import { + captureBilling, + chatResponse, + expectedUsage, +} from '../../core/test/billing-fixtures.ts' +import { vaultRowFor } from '../../core/test/fakes.ts' +import { installVaultHost } from '../../core/test/host.ts' +import { createTestCore } from '../../core/test-entry.ts' +import { createPiExtension, placeholderKey } from '../src/extension.ts' +import { agentDir } from './harness.ts' + +async function turn( + provider: 'deepseek' | 'cerebras', + ownKey = false, + retry = false, + redirect = false, + onSemantics?: () => void, + onReady?: () => void, +) { + const base = supportedProviders.find((row) => row.service === provider) + if (!base?.billing) throw new Error('missing provider') + const entry = onSemantics + ? { + ...base, + billing: new Proxy(base.billing, { + get(target, key, receiver) { + onSemantics() + return Reflect.get(target, key, receiver) + }, + }), + } + : base + const rows = [ + vaultRowFor(entry, { accountId: 'account-1' }), + ...(retry + ? [ + vaultRowFor(entry, { + id: `apikey:${provider}:second`, + accountId: 'account-2', + }), + ] + : []), + ] + // The extension creates the core under test. An in-memory vault client avoids + // a second, unused core's token write and roster refresh during fixture setup. + installVaultHost({ + async listScoped() { + return { rows, view: 'view-1', grants: undefined, grantTuples: undefined } + }, + async getScoped({ credentialId }) { + const listed = rows.find((row) => row.id === credentialId) + if (!listed) + throw new Error(`unknown fixture credential: ${credentialId}`) + return { + credentialId, + material: `secret-${credentialId}-1`, + accountId: listed.accountId, + recordVersion: 1, + expiresAtMs: null, + } + }, + async reportAuthFailureScoped() {}, + close() {}, + }) + const agent = await agentDir({ + enrolled: false, + flags: { [provider]: { consented: true } }, + }) + process.env.PI_CODING_AGENT_DIR = agent.agent + let registered: Record | undefined + const handlers: Record< + string, + (event: unknown, ctx: unknown) => Promise + > = {} + const pi = { + on(name: string, handler: (event: unknown, ctx: unknown) => Promise) { + handlers[name] = handler + }, + registerCommand() {}, + unregisterProvider() {}, + registerProvider(_name: string, value: Record) { + registered = value + }, + } as unknown as ExtensionAPI + const context = { + modelRegistry: { + getAll: () => [{ provider, id: 'sent-model', api: 'openai-completions' }], + refresh: async () => {}, + }, + ui: { notify() {} }, + cwd: process.cwd(), + sessionManager: { getSessionId: () => 'pi-session' }, + } as unknown as ExtensionContext + createPiExtension({ createCore: (built) => createTestCore(built, [entry]) })( + pi, + ) + try { + await handlers.session_start?.({}, context) + onReady?.() + const stream = registered?.streamSimple as ( + ...args: unknown[] + ) => AssistantMessageEventStream + if (!stream) throw new Error('missing registered stream') + let sends = 0 + let includeUsage: unknown + const reply = stream( + { + provider, + id: 'sent-model', + api: 'openai-completions', + baseUrl: `${entry.origins[0]}/v1`, + input: ['text'], + output: ['text'], + reasoning: false, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 128000, + maxTokens: 100, + }, + { messages: [] }, + { + sessionId: 'pi-session', + apiKey: ownKey ? 'user-key' : placeholderKey(provider), + fetch: async (input: RequestInfo | URL, init?: RequestInit) => { + const request = new Request(input, init) + includeUsage = (await request.json()).stream_options + expect(request.headers.get('authorization')).toBe( + ownKey + ? 'Bearer user-key' + : `Bearer secret-apikey:${provider}${retry && sends > 0 ? ':second' : ''}-1`, + ) + if (retry && ++sends === 1) + return new Response('busy', { + status: 429, + headers: { 'retry-after': '120' }, + }) + if (redirect && ++sends === 1) + return new Response(null, { + status: 307, + headers: { location: `${entry.origins[0]}/v1/redirected` }, + }) + return chatResponse(provider) + }, + }, + ) + for await (const _event of reply) { + } + const message = await reply.result() + expect(message.stopReason).toBe('stop') + expect(includeUsage).toEqual({ include_usage: true }) + return message + } finally { + await handlers.session_shutdown?.({}, context) + } +} + +for (const provider of ['deepseek', 'cerebras'] as const) { + test(`Pi ${provider} bills provider-true usage once with Pi's final responseId`, async () => { + const { reports, sessions } = captureBilling() + let parsed = 0 + const message = await turn( + provider, + false, + false, + false, + () => { + parsed++ + }, + () => { + parsed = 0 + }, + ) + expect(parsed).toBeGreaterThan(0) + expect(message.responseId).toBe('reply-1') + expect(reports).toEqual([ + { + kind: 'billing', + attempt_id: expect.any(String), + attempt: 1, + request_kind: 'primary', + credential_id: `apikey:${provider}`, + account_id: 'account-1', + provider, + model: 'sent-model', + outcome: 'completed', + status: 200, + provider_response_id: message.responseId, + ...(provider === 'deepseek' ? { charge_basis: 'metered_api' } : {}), + usage: expectedUsage, + cache: { implicit: true }, + }, + ]) + expect(sessions.every((session) => session === 'pi-session')).toBe(true) + }) +} + +test('Pi leaves a user key unbilled even when the routing accessor is installed', async () => { + const { reports } = captureBilling() + await turn('deepseek', true) + expect(reports).toEqual([]) +}) + +test('Pi reroute bills both physical sends with different UUIDs and attempts', async () => { + const { reports } = captureBilling() + await turn('deepseek', false, true) + expect(reports).toHaveLength(2) + expect( + reports.map((report) => [report.attempt, report.status, report.outcome]), + ).toEqual([ + [1, 429, 'failed'], + [2, 200, 'completed'], + ]) + expect(reports[0]?.attempt_id).not.toBe(reports[1]?.attempt_id) +}) + +test('Pi followed same-origin redirect reports only the completed model call', async () => { + const { reports } = captureBilling() + await turn('deepseek', false, false, true) + expect(reports).toHaveLength(1) + expect(reports[0]).toMatchObject({ + status: 200, + outcome: 'completed', + usage: expectedUsage, + }) +}) + +test('Pi without an accessor never looks up usage semantics for a served request', async () => { + Reflect.deleteProperty(globalThis, Symbol.for('cortexkit.routing')) + let parsed = 0 + await turn( + 'deepseek', + false, + false, + false, + () => { + parsed++ + }, + () => { + parsed = 0 + }, + ) + expect(parsed).toBe(0) +}) diff --git a/scripts/live-smoke/billing-preload.mjs b/scripts/live-smoke/billing-preload.mjs new file mode 100644 index 0000000..f091008 --- /dev/null +++ b/scripts/live-smoke/billing-preload.mjs @@ -0,0 +1,14 @@ +import { appendFileSync } from 'node:fs' + +// Import in the server before plugin setup. Never capture requests or headers. +const destination = process.env.CLAUSTRUM_AUTH_LIVE_BILLING +if (destination) { + Reflect.set(globalThis, Symbol.for('cortexkit.routing'), () => ({ + report(payload) { + if (payload.kind === 'billing') + appendFileSync(destination, `${JSON.stringify(payload)}\n`, { + mode: 0o600, + }) + }, + })) +} diff --git a/scripts/live-smoke/billing.test.ts b/scripts/live-smoke/billing.test.ts new file mode 100644 index 0000000..035f543 --- /dev/null +++ b/scripts/live-smoke/billing.test.ts @@ -0,0 +1,202 @@ +import { describe, expect, test } from 'bun:test' +import { + type BillingProbe, + type BillingReport, + billingChecks, +} from './billing.ts' + +const report = (usage?: BillingReport['usage']): BillingReport => ({ + kind: 'billing', + request_kind: 'primary', + provider: 'synthetic', + model: 'model', + outcome: 'completed', + ...(usage ? { usage } : {}), +}) +const probe = ( + name: BillingProbe['name'], + usage?: BillingReport['usage'], + reply = 'OK', +): BillingProbe => ({ + name, + model: 'model', + session: 'session', + reports: [report(usage)], + reply, + stored: [], +}) +const cacheCheck = ( + first: BillingReport['usage'], + second: BillingReport['usage'], +) => billingChecks([probe('long-1', first), probe('long-2', second)])[0] +const reasoningCheck = (usage?: BillingReport['usage'], reply = 'OK') => + billingChecks([probe('reasoning', usage, reply)])[1] + +describe('billing report checks', () => { + test('included-cache counts PASS after restoring the subtracted hit', () => { + expect( + cacheCheck( + { input: 5000, output: 1 }, + { input: 4000, output: 1, cache_read: 1000 }, + ), + ).toMatchObject({ verdict: 'PASS' }) + }) + + test('excluded-cache counts FAIL when the table subtracted cache twice', () => { + const result = cacheCheck( + { input: 5000, output: 1 }, + { input: 3000, output: 1, cache_read: 1000 }, + ) + expect(result.verdict).toBe('FAIL') + expect(result.reason).toContain('prompt likely excludes cache') + }) + + test('no second-turn hit is UNTESTED, including an explicit zero', () => { + for (const cache_read of [undefined, 0]) { + expect( + cacheCheck( + { input: 5000, output: 1 }, + { input: 5000, output: 1, cache_read }, + ), + ).toMatchObject({ + verdict: 'UNTESTED', + reason: expect.stringContaining('no hit'), + }) + } + }) + + test('restores writes and first-turn hits before comparing identical contexts', () => { + expect( + cacheCheck( + { input: 3500, output: 1, cache_read: 1000, cache_write: 500 }, + { input: 500, output: 1, cache_read: 4000, cache_write: 500 }, + ).verdict, + ).toBe('PASS') + }) + + test('two percent is inclusive; a larger mismatch fails', () => { + for (const [input, verdict] of [ + [4100, 'PASS'], + [4101, 'FAIL'], + ] as const) { + expect( + cacheCheck({ input: 5000 }, { input, cache_read: 1000 }).verdict, + ).toBe(verdict) + } + }) + + test('a hit without both completed primary input counts is UNTESTED', () => { + expect(cacheCheck(undefined, { input: 0, cache_read: 5000 }).verdict).toBe( + 'UNTESTED', + ) + const cut = probe('long-1', { input: 5000 }) + cut.reports[0].outcome = 'cut' + expect( + billingChecks([cut, probe('long-2', { input: 0, cache_read: 5000 })])[0] + .verdict, + ).toBe('UNTESTED') + }) + + test('reasoning included in output PASSes when visible tokens match chars/4', () => { + expect( + reasoningCheck({ input: 10, output: 101, reasoning: 100 }).verdict, + ).toBe('PASS') + }) + + for (const [provider, output, reasoning, storedOutput] of [ + ['DeepSeek', 43, 41, 2], + ['Kimi', 53, 39, 14], + ['Cerebras', 133, 128, 5], + ] as const) { + test(`live ${provider} reasoning row uses stored visible output`, () => { + const turn = probe('reasoning', { output, reasoning }, '120') + turn.stored = [{ tokens: { output: storedOutput } }] + expect(billingChecks([turn])[1]).toMatchObject({ + verdict: 'PASS', + reason: expect.stringContaining('OpenCode stored visible output'), + }) + }) + } + + test('reasoning exceeding output FAILs', () => { + expect(reasoningCheck({ output: 100, reasoning: 101 })).toMatchObject({ + verdict: 'FAIL', + reason: 'reasoning 101 exceeds output 100', + }) + }) + + test('stored visible output takes precedence with an inclusive ±1 tolerance', () => { + for (const [output, verdict] of [ + [1, 'PASS'], + [3, 'PASS'], + [0, 'FAIL'], + [4, 'FAIL'], + ] as const) { + const turn = probe('reasoning', { output: 102, reasoning: 100 }) + turn.stored = [{ tokens: { output } }] + expect(billingChecks([turn])[1].verdict).toBe(verdict) + } + const zero = probe('reasoning', { output: 100, reasoning: 100 }, '') + zero.stored = [{ tokens: { output: 0 } }] + expect(billingChecks([zero])[1].verdict).toBe('PASS') + }) + + test('reasoning exceeding output FAILs even when stored output is available', () => { + const turn = probe('reasoning', { output: 100, reasoning: 101 }) + turn.stored = [{ tokens: { output: 0 } }] + expect(billingChecks([turn])[1].verdict).toBe('FAIL') + }) + + test('fallback accepts short replies within ±8 tokens, not beyond', () => { + for (const [output, verdict] of [ + [109, 'PASS'], + [110, 'FAIL'], + ] as const) { + expect(reasoningCheck({ output, reasoning: 100 }, '1234').verdict).toBe( + verdict, + ) + } + }) + + test('reasoning difference outside both fallback tolerances FAILs', () => { + for (const output of [301, 149]) + expect( + reasoningCheck({ output, reasoning: 100 }, 'x'.repeat(400)).verdict, + ).toBe('FAIL') + }) + + test('missing reasoning count is UNTESTED rather than invented zero', () => { + expect(reasoningCheck({ output: 1 }).verdict).toBe('UNTESTED') + expect(reasoningCheck({ output: 1, reasoning: 0 }).verdict).toBe('PASS') + expect(reasoningCheck({ reasoning: 100 }).verdict).toBe('UNTESTED') + expect(billingChecks([])[1].verdict).toBe('UNTESTED') + }) + + test('an empty visible reply cannot produce a reasoning PASS', () => { + expect(reasoningCheck({ output: 100, reasoning: 100 }, '').verdict).toBe( + 'FAIL', + ) + }) + + test('every completed report needs usage input and output, retaining measured zero', () => { + const reports = [report(), report({ input: 1 }), report({ output: 1 })] + expect(billingChecks([], reports)[2]).toMatchObject({ + verdict: 'FAIL', + reason: '3/3 completed reports missing usage.input or usage.output', + }) + expect( + billingChecks([], [report({ input: 0, output: 0 })])[2].verdict, + ).toBe('PASS') + reports[0].outcome = 'failed' + expect(billingChecks([], [reports[0]])[2].verdict).toBe('UNTESTED') + }) + + test('raw host counters are not cache verdict inputs', () => { + const probes = [ + probe('long-1', { input: 5000 }), + probe('long-2', { input: 4000, cache_read: 1000 }), + ] + probes[1].stored = [{ tokens: { input: 999999, output: 999999 } }] + expect(billingChecks(probes)[0].verdict).toBe('PASS') + }) +}) diff --git a/scripts/live-smoke/billing.ts b/scripts/live-smoke/billing.ts new file mode 100644 index 0000000..871da43 --- /dev/null +++ b/scripts/live-smoke/billing.ts @@ -0,0 +1,208 @@ +/** The routing accessor receives wire-format reports (snake_case), not SDK inputs. */ +export type BillingReport = { + kind: 'billing' + request_kind: string + provider: string + model: string + outcome: string + charge_basis?: string + cache?: unknown + usage?: { + input?: number + output?: number + cache_read?: number + cache_write?: number + reasoning?: number + } +} + +export type ProbeName = 'long-1' | 'long-2' | 'reasoning' +export type BillingProbe = { + name: ProbeName + model: string + session: string + reports: BillingReport[] + reply: string + stored: { agent?: string; tokens?: HostTokens }[] + captureError?: string +} +type HostTokens = { + input?: number + output?: number + reasoning?: number + cache?: { read?: number; write?: number } +} +export type Check = { + check: string + verdict: 'PASS' | 'FAIL' | 'UNTESTED' + reason: string +} + +// Roughly 3,000 tokens of fixed prose; no timestamp or provider-specific suffix +// should displace the shared prefix that automatic caches can reuse. +export const longPrompt = `${Array.from( + { length: 110 }, + () => + 'The quiet library keeps a careful record of every book. Each morning the librarian checks the shelves and returns every volume to its place.', +).join('\n')}\nReply with the single word OK` +export const reasoningPrompt = + 'Work out the smallest positive integer divisible by 6, 8, and 15. Reply with only the integer.' + +const count = (value: number | undefined): value is number => + typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 +const near = (a: number, b: number) => Math.abs(a - b) <= a * 0.02 +const primary = (probe?: BillingProbe) => { + // The accessor is session-scoped, but its payload has no session id. The + // runner associates reports by the capture file boundaries around each send. + const reports = probe?.reports.filter((r) => r.request_kind === 'primary') + return reports?.length === 1 && reports[0].outcome === 'completed' + ? reports[0] + : undefined +} + +export function billingChecks( + probes: BillingProbe[], + reports = probes.flatMap((p) => p.reports), +): Check[] { + const first = primary(probes.find((p) => p.name === 'long-1'))?.usage + const second = primary(probes.find((p) => p.name === 'long-2'))?.usage + let prompt: Check = { + check: 'Prompt includes cache', + verdict: 'UNTESTED', + reason: 'no hit (second long turn has no positive cache_read)', + } + if (second && count(second.cache_read) && second.cache_read > 0) { + if (!first || !count(first.input) || !count(second.input)) { + prompt.reason = 'missing input counts on completed long-turn reports' + } else { + // Reports already subtract cache from input. Restore the first turn's + // total too, since an automatic cache can hit even on the first probe. + const before = + first.input + (first.cache_read ?? 0) + (first.cache_write ?? 0) + const after = second.input + second.cache_read + (second.cache_write ?? 0) + prompt = { + check: prompt.check, + verdict: near(before, after) ? 'PASS' : 'FAIL', + reason: near(before, after) + ? `input+cache totals ${before} vs ${after} agree within 2%` + : `input+cache totals ${before} vs ${after} differ by >2%${near(before - second.cache_read, after) ? '; near first total minus cache_read: prompt likely excludes cache' : '; check provider semantics/context parity'}`, + } + } + } + + const reasoningProbe = probes.find((p) => p.name === 'reasoning') + const usage = primary(reasoningProbe)?.usage + let reasoning: Check = { + check: 'Reasoning in output', + verdict: 'UNTESTED', + reason: reasoningProbe + ? 'no reasoning count on a completed primary report' + : 'no eligible reasoning-capable model/probe', + } + if (usage && count(usage.reasoning)) { + if (!count(usage.output)) reasoning.reason = 'missing output count' + else if (usage.reasoning > usage.output) + reasoning = { + check: reasoning.check, + verdict: 'FAIL', + reason: `reasoning ${usage.reasoning} exceeds output ${usage.output}`, + } + else { + const visible = usage.output - usage.reasoning + const storedOutputs = reasoningProbe?.stored + .map(({ tokens }) => tokens?.output) + .filter(count) + const storedOutput = + storedOutputs?.length === 1 ? storedOutputs[0] : undefined + // OpenCode stores visible output separately from reasoning. Prefer its + // measured count to a character estimate, especially for short replies. + if (count(storedOutput)) + reasoning = { + check: reasoning.check, + verdict: Math.abs(visible - storedOutput) <= 1 ? 'PASS' : 'FAIL', + reason: `output-reasoning=${visible}; OpenCode stored visible output=${storedOutput} (±1 token tolerance)`, + } + else { + const estimate = (reasoningProbe?.reply.length ?? 0) / 4 + reasoning = { + check: reasoning.check, + verdict: + estimate > 0 && + ((visible >= estimate / 2 && visible <= estimate * 2) || + Math.abs(visible - estimate) <= 8) + ? 'PASS' + : 'FAIL', + reason: `output-reasoning=${visible}; visible reply estimate=${estimate.toFixed(2)} tokens (chars/4, factor-of-two or ±8 token tolerance; stored visible output unavailable)`, + } + } + } + } + + const completed = reports.filter((r) => r.outcome === 'completed') + const missing = completed.filter( + (r) => !count(r.usage?.input) || !count(r.usage?.output), + ) + return [ + prompt, + reasoning, + { + check: 'Usage present', + verdict: missing.length ? 'FAIL' : completed.length ? 'PASS' : 'UNTESTED', + reason: missing.length + ? `${missing.length}/${completed.length} completed reports missing usage.input or usage.output` + : completed.length + ? `all ${completed.length} completed reports carry input and output` + : 'no completed billing reports', + }, + ] +} + +/** Keep only stored accounting fields, never arbitrary message metadata. */ +export function storedCounts(value: unknown): BillingProbe['stored'] { + const messages = value as { + data?: { type?: string; agent?: string; tokens?: HostTokens }[] + } + return (messages.data ?? []) + .filter((message) => message.type === 'assistant') + .map(({ agent, tokens }) => ({ + agent, + ...(tokens + ? { + tokens: { + input: tokens.input, + output: tokens.output, + reasoning: tokens.reasoning, + cache: { read: tokens.cache?.read, write: tokens.cache?.write }, + }, + } + : {}), + })) +} + +export function billingTable( + provider: string, + probes: BillingProbe[], + reports: BillingReport[], + cell: (text: string) => string, +) { + const rows = billingChecks(probes, reports) + .map((c) => `| ${c.check} | ${c.verdict} | ${cell(c.reason)} |`) + .join('\n') + const totals = probes + .map((p) => { + const context = p.stored.map(({ tokens: t }) => + t && count(t.input) + ? t.input + (t.cache?.read ?? 0) + (t.cache?.write ?? 0) + : 'unavailable', + ) + return `| ${[p.name, p.model, p.session, JSON.stringify(p.reports.map((r) => r.usage ?? null)), JSON.stringify(p.stored), context.length ? context.join(', ') : 'unavailable', p.captureError ?? ''].map(cell).join(' | ')} |` + }) + .join('\n') + const raw = reports + .map( + (r, at) => + `| ${[String(at + 1), r.request_kind, r.model, r.outcome, r.charge_basis ?? '(absent)', JSON.stringify(r.cache ?? null), JSON.stringify(r.usage ?? null)].map(cell).join(' | ')} |`, + ) + .join('\n') + return `\n## Billing: ${cell(provider)}\n\n| check | verdict | reason |\n| --- | --- | --- |\n${rows}\n\nFresh sessions, same cwd/default agent; no conversation history reused. Stored visible output is used only for the reasoning verdict; other host totals are diagnostic. Host context = stored input + cache.read + cache.write; per-session system-prompt differences are not observable from billing reports.\n\n| turn | model/variant | session | report usage | OpenCode stored agent/tokens | host context tokens | capture error |\n| --- | --- | --- | --- | --- | --- | --- |\n${totals}\n\n| report | request kind | model | outcome | chargeBasis | cache | usage |\n| --- | --- | --- | --- | --- | --- | --- |\n${raw}\n` +} diff --git a/scripts/live-smoke/run.ts b/scripts/live-smoke/run.ts index b34738f..0001c4d 100644 --- a/scripts/live-smoke/run.ts +++ b/scripts/live-smoke/run.ts @@ -17,6 +17,16 @@ import { fileURLToPath, pathToFileURL } from 'node:url' import { freePort } from '../../packages/opencode/test/v2/e2e/host.ts' import { PiRpc } from '../../packages/pi/test/rpc.ts' import { packedRecords } from '../npm-pack-json.ts' +import { + type BillingProbe, + type BillingReport, + billingChecks, + billingTable, + longPrompt, + type ProbeName, + reasoningPrompt, + storedCounts, +} from './billing.ts' export const repo = fileURLToPath(new URL('../../', import.meta.url)) export type Host = 'opencode1' | 'opencode2' | 'pi' @@ -290,6 +300,33 @@ function reasoningRank(model: Model) { return reasoning || variants ? 2 : 1 } +function billingVariant(model: Model) { + const variants = Array.isArray(model.variants) + ? model.variants.map((v) => v.id) + : Object.keys(model.variants ?? {}) + return ['low', 'minimal', 'medium', 'high', 'xhigh', 'max'].find((v) => + variants.includes(v), + ) +} + +function reasoningModel(models: Model[], provider: string) { + const candidates = models.filter((model) => { + const capability = model.capabilities?.reasoning ?? model.reasoning + return ( + reasoningRank(model) === 2 || + capability === true || + (typeof capability === 'object' && capability.supported) || + billingVariant(model) !== undefined + ) + }) + try { + return cheapest(candidates, 'opencode2', provider) + } catch (error) { + if (error instanceof SmokeError && /^No/.test(error.code)) return undefined + throw error + } +} + export function smokeTokens(model: Model, host: Host, provider: string) { // A 16-token response can consist entirely of hidden reasoning. Meta's // captured variants lack an off control; Pi model overrides can also hide @@ -363,21 +400,27 @@ export type Options = { hosts: Host[] provider: string dry: boolean + billing?: boolean } export function options(args: string[], env = process.env): Options { - let host = 'all', + let host = '', provider = 'all', - dry = false + dry = false, + billing = false for (let i = 0; i < args.length; i++) { const arg = args[i] if (arg === '--dry') dry = true + else if (arg === '--billing') billing = true else if (arg === '--host' && args[i + 1]) host = args[++i] else if (arg === '--provider' && args[i + 1]) provider = args[++i] else throw new SmokeError( - 'Usage: --host opencode1|opencode2|pi|all --provider |all [--dry]', + 'Usage: --host opencode1|opencode2|pi|all --provider |all [--dry] [--billing (OpenCode 2 only)]', ) } + host ||= billing ? 'opencode2' : 'all' + if (billing && host !== 'opencode2') + throw new SmokeError('BillingRequiresOpenCode2') if (!env.LIVE_HOME || !env.CLAUSTRUM_SUBC_CONNECTION) throw new SmokeError('LIVE_HOME and CLAUSTRUM_SUBC_CONNECTION are required') if (host !== 'all' && !hosts.includes(host as Host)) @@ -395,6 +438,7 @@ export function options(args: string[], env = process.env): Options { hosts: selected, provider, dry, + ...(billing ? { billing } : {}), } } @@ -448,11 +492,13 @@ function plan(o: Options) { ? `${bin} auth login ${id} --method claustrum-auth --server ` : `/claustrum use ${id}${host === 'pi' ? ' (RPC consent: true)' : ''}`, `${host}/${p}: discover catalog, choose a text/tool model with reasoning off where available, cap output at 16 (1024 for reasoning), restart host, check ownership`, - host === 'opencode2' - ? `${bin} run --server --format json --title 'Live smoke' --model ${id}/ 'Reply with the single word OK.'` - : host === 'pi' - ? `RPC set_model ${id}/; prompt 'Reply with the single word OK.'` - : `HTTP POST /session//message model=${id}/ text='Reply with the single word OK.'`, + o.billing + ? `${host}/${p}: preload captures billing payloads in /.live-smoke/opencode2/billing.jsonl; two identical ~3000-token prompts in fresh sessions with the same cwd/default agent/model, then a short reasoning probe if catalog offers one; one send per probe; print report verdicts and stored host token totals` + : host === 'opencode2' + ? `${bin} run --server --format json --title 'Live smoke' --model ${id}/ 'Reply with the single word OK.'` + : host === 'pi' + ? `RPC set_model ${id}/; prompt 'Reply with the single word OK.'` + : `HTTP POST /session//message model=${id}/ text='Reply with the single word OK.'`, host === 'opencode2' ? `${bin} auth logout ${id} --server ` : `/claustrum accounts; /claustrum release ${id}`, @@ -472,6 +518,7 @@ class Runner { readonly vaultIDs = new Map() readonly children: ChildProcess[] = [] readonly pending: Promise[] = [] + readonly billingProbes = new Map() constructor( readonly o: Options, readonly host: Host, @@ -484,6 +531,16 @@ class Runner { get cwd() { return path.join(this.o.home, '.live-smoke', this.host, 'project') } + get billingFile() { + return path.join(this.o.home, '.live-smoke', this.host, 'billing.jsonl') + } + billingReports(): BillingReport[] { + if (!existsSync(this.billingFile)) return [] + return readFileSync(this.billingFile, 'utf8') + .split('\n') + .filter(Boolean) + .map((line) => JSON.parse(line) as BillingReport) + } start(argv: string[]) { const prefix = path.join( this.o.home, @@ -686,7 +743,12 @@ interface Driver { menu(text: string): Promise activate(id: string): Promise models(id: string): Promise - turn(id: string, model: Model, followup?: boolean): Promise + turn( + id: string, + model: Model, + followup?: boolean, + probe?: ProbeName, + ): Promise release(id: string): Promise stop(): Promise } @@ -943,7 +1005,7 @@ async function openCodeDriver(r: Runner): Promise { const catalog = (await api('GET', '/provider')) as { all: V1Provider[] } return v1Models(catalog.all.find((p) => p.id === id) ?? { id }) }, - async turn(id, model, followup = false) { + async turn(id, model, followup = false, probe) { if (readFileSync(ready, 'utf8') !== '1') throw new SmokeError('RetryGuardNotReady') // Bound physical sends, including output-limit continuations, not just @@ -958,12 +1020,16 @@ async function openCodeDriver(r: Runner): Promise { budget, JSON.stringify({ status: 0, - maxOutputTokens: smokeTokens(model, r.host, id), + maxOutputTokens: + probe === 'reasoning' ? 1024 : smokeTokens(model, r.host, id), }), ) if (existsSync(`${budget}.spent`)) unlinkSync(`${budget}.spent`) if (v2) { if (followup && !turnSession) throw new SmokeError('MissingSessionID') + const offset = probe ? r.billingReports().length : 0 + const variant = + probe === 'reasoning' ? billingVariant(model) : smokeVariant(model) const result = await r.run([ bin, 'run', @@ -975,8 +1041,12 @@ async function openCodeDriver(r: Runner): Promise { '--format', 'json', '--model', - `${id}/${model.id}${smokeVariant(model) ? `#${smokeVariant(model)}` : ''}`, - 'Reply with the single word OK.', + `${id}/${model.id}${variant ? `#${variant}` : ''}`, + probe === 'reasoning' + ? reasoningPrompt + : probe + ? longPrompt + : 'Reply with the single word OK.', ]) const events = result.text.split('\n').flatMap((line) => { try { @@ -985,6 +1055,43 @@ async function openCodeDriver(r: Runner): Promise { return [] } }) + const sessions = [ + ...new Set( + events + .map((e) => e.sessionID) + .filter((id): id is string => typeof id === 'string'), + ), + ] + const answer = events + .filter((e) => e.type === 'text') + .map((e) => e.part?.text ?? e.text ?? '') + .join('') + if (probe) { + const captured: BillingProbe = { + name: probe, + model: `${model.id}${variant ? `#${variant}` : ''}`, + session: sessions.length === 1 ? sessions[0] : '', + reports: r + .billingReports() + .slice(offset) + .filter((r) => r.provider === id), + reply: answer, + stored: [], + } + const probes = r.billingProbes.get(id) ?? [] + probes.push(captured) + r.billingProbes.set(id, probes) + try { + if (!captured.session) throw new SmokeError('MissingSessionID') + captured.stored = storedCounts( + await api('GET', `/api/session/${captured.session}/message`), + ) + if (!captured.stored.some((message) => message.tokens)) + throw new SmokeError('StoredTokensMissing') + } catch (error) { + captured.captureError = failure(error) + } + } if (JSON.parse(readFileSync(budget, 'utf8')).exhausted) throw new SmokeError('RequestBudgetExhausted') if (result.code !== 0 || events.some((e) => e.type === 'error')) @@ -993,24 +1100,12 @@ async function openCodeDriver(r: Runner): Promise { ? 'Timeout' : errorClass(result.text + result.error), ) - const sessions = [ - ...new Set( - events - .map((e) => e.sessionID) - .filter((id): id is string => typeof id === 'string'), - ), - ] if (sessions.length !== 1) throw new SmokeError('MissingOrMixedSessionID') if (followup && sessions[0] !== turnSession) throw new SmokeError('FollowupSessionMismatch') turnSession = sessions[0] - return ( - events - .filter((e) => e.type === 'text') - .map((e) => e.part?.text ?? e.text ?? '') - .join('') || 'NoAnswerText' - ) + return answer || 'NoAnswerText' } if (!followup) turnSession = await session() // The guard allows one provider request using an exclusive marker file. @@ -1121,6 +1216,8 @@ export type TestHooks = { onError?: (error: unknown) => void } export async function runSmoke(o: Options, hooks: TestHooks = {}) { + if (o.billing && (o.hosts.length !== 1 || o.hosts[0] !== 'opencode2')) + throw new SmokeError('BillingRequiresOpenCode2') o = { ...o, home: scratchHome(o.home) } const hostEnv = (host: Host) => ({ ...environment(o.home, o.connection, host), @@ -1146,7 +1243,7 @@ export async function runSmoke(o: Options, hooks: TestHooks = {}) { throw new SmokeError('MissingConnectionFile') const results: { host: Host; report: string; rows: Row[] }[] = [] for (const host of o.hosts) { - const env = hostEnv(host) + const env: Record = hostEnv(host) const work = path.join(o.home, '.live-smoke', host) for (const dir of [ work, @@ -1166,6 +1263,11 @@ export async function runSmoke(o: Options, hooks: TestHooks = {}) { let driver: Driver | undefined const baseline = stamp(o.home) try { + if (o.billing) { + confined(o.home, r.billingFile) + writeFileSync(r.billingFile, '', { mode: 0o600 }) + env.CLAUSTRUM_AUTH_LIVE_BILLING = r.billingFile + } const initialized = await r.run(['git', 'init', '-q', r.cwd]) if (initialized.code !== 0) throw new SmokeError('ScratchGitInitFailed') const pkg = await install(r) @@ -1244,6 +1346,7 @@ export async function runSmoke(o: Options, hooks: TestHooks = {}) { writeFileSync( path.join(guard, 'server.js'), ` + ${o.billing ? `import ${JSON.stringify(pathToFileURL(path.join(repo, 'scripts/live-smoke/billing-preload.mjs')).href)}` : ''} import original from ${JSON.stringify(original)} import {writeFileSync,openSync,closeSync} from 'node:fs' export default {...original,setup:async(ctx,options)=>{ @@ -1348,18 +1451,38 @@ export async function runSmoke(o: Options, hooks: TestHooks = {}) { const available = await driver.models(id) const models = hooks.models?.(available) ?? available const model = cheapest(models, host, p) - capModels(r, id, p, models) + const reasoning = o.billing ? reasoningModel(models, p) : undefined + capModels(r, id, p, models, reasoning) await driver.stop() driver = host === 'pi' ? await piDriver(r, plugin) : await openCodeDriver(r) - const answer = await driver.turn(id, model) + const answer = await driver.turn( + id, + model, + false, + o.billing ? 'long-1' : undefined, + ) if (!answer.trim() || answer === 'NoAnswerText') throw new SmokeError('NoAnswerText') row.turn = `${preview(answer)} (model: ${model.id})` - if (p === 'cline-pass') + if (o.billing) { + // Fresh sessions avoid doubling the prompt with conversation history. + // Reuse the host, cwd, default agent and model to minimize changes + // to the system prefix; stored token totals expose size differences. + await driver.turn(id, model, false, 'long-2') + if (reasoning) await driver.turn(id, reasoning, false, 'reasoning') + const probes = r.billingProbes.get(id) ?? [] + const checks = billingChecks(probes) + row.extras = checks.map((c) => `${c.verdict} ${c.check}`).join('; ') + if (probes.some((probe) => !probe.reports.length)) + row.extras += '; FAIL billing report missing for executed probe' + if (probes.some((probe) => probe.captureError)) + row.extras += '; FAIL stored token capture' + } + if (!o.billing && p === 'cline-pass') row.extras = 'credential-id classified; refusals clear (verify vault row has no provider id)' - if (p === 'meta') + if (!o.billing && p === 'meta') row.extras = 'one turn only; insufficient-validity is a vault defect; no 401 retry' const accounts = await driver.menu('accounts') @@ -1444,7 +1567,36 @@ export async function runSmoke(o: Options, hooks: TestHooks = {}) { path.dirname(o.home), `results-${host}-${new Date().toISOString().replace(/[:.]/g, '-')}.md`, ) - const markdown = table(host, rows) + const reports = o.billing ? r.billingReports() : [] + const missingUsage = [ + ...new Set( + reports + .filter( + (r) => + r.outcome === 'completed' && + (!r.usage || + r.usage.input === undefined || + r.usage.output === undefined), + ) + .map((r) => r.provider), + ), + ] + const markdown = + table(host, rows) + + (o.billing + ? rows + .map((row) => { + const id = providerID(host, row.provider) + return billingTable( + row.provider, + r.billingProbes.get(id) ?? [], + reports.filter((r) => r.provider === id), + safe, + ) + }) + .join('') + + `\nProviders with completed replies missing usage/input/output: ${missingUsage.length ? missingUsage.map(safe).join(', ') : 'none'}\n` + : '') writeFileSync(report, markdown, { mode: 0o600 }) console.log(markdown) results.push({ host, report, rows }) @@ -1467,7 +1619,13 @@ export async function main() { } if (import.meta.main) await main() -function capModels(r: Runner, id: string, provider: string, models: Model[]) { +function capModels( + r: Runner, + id: string, + provider: string, + models: Model[], + reasoning?: Model, +) { const file = r.host === 'pi' ? path.join(r.env.PI_CODING_AGENT_DIR, 'models.json') @@ -1483,7 +1641,13 @@ function capModels(r: Runner, id: string, provider: string, models: Model[]) { for (const model of models) config[key][id].models[model.id] = { ...config[key][id].models[model.id], - limit: { ...model.limit, output: smokeTokens(model, r.host, provider) }, + limit: { + ...model.limit, + output: + reasoning?.id === model.id + ? 1024 + : smokeTokens(model, r.host, provider), + }, } } writeFileSync(file, JSON.stringify(config))