diff --git a/.env.example b/.env.example index ea00ae63..136c4b34 100644 --- a/.env.example +++ b/.env.example @@ -58,6 +58,17 @@ LOG_TO_FILE=false # Set to true to enable file logging # GEMINI_API_KEY="AIza..." # ANTIGRAVITY_MODEL="gemini-3.1-pro-preview" +# LiteLLM (Anthropic-compatible) open-weight backend (API_BACKEND=litellm). Point +# the agent at a self-hosted LiteLLM proxy (see litellm/start-litellm.sh, which +# prints these). LITELLM_COST_LOG must be the SAME path the proxy writes its +# per-call JSONL to — that is how the run joins ACTUAL OpenRouter cost + cache back +# onto each turn; if it is unset (or points elsewhere) the run silently falls back +# to static rate-card pricing with 0 cache reads. +# LITELLM_BASE_URL="http://localhost:4000" +# LITELLM_AUTH_TOKEN="sk-..." +# LITELLM_MODEL="zai.glm-5" +# LITELLM_COST_LOG="./tmp/litellm-costs.jsonl" + # UiPath CLI plugin-discovery pin. When unset, the sandbox auto-derives # the canonical `node_modules/@uipath` from the resolved `uip` binary at setup # time. Operators on dedicated eval hosts can pin explicitly to override. diff --git a/CLAUDE.md b/CLAUDE.md index 82f395b9..16ef30cb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -25,6 +25,7 @@ coder_eval/ ├── logging_config.py # Structured logging setup ├── path_utils.py # Run ID generation, path utilities ├── pricing.py # Model pricing / cost calculation (ModelPricing, calculate_cost, register_pricing) +├── litellm_cost.py # Join proxy-captured ACTUAL per-call cost/cache onto turns (LiteLLM backend; apply_actual_cost) ├── utils.py # Version info helpers │ ├── agents/ @@ -41,7 +42,7 @@ coder_eval/ │ ├── routing.py # ApiRoute (DirectRoute/BedrockRoute) │ ├── sandbox.py # SandboxConfig, ResourceLimits │ ├── tasks.py # TaskDefinition, AgentConfig, Dataset (dataset fan-out + sample) -│ ├── telemetry.py # CommandTelemetry, CommandStatistics, TokenUsage, ReconciliationMessage, TranscriptMessage +│ ├── telemetry.py # CommandTelemetry, CommandStatistics, TokenUsage, ProviderCallCost, ReconciliationMessage, TranscriptMessage │ └── templates.py # RepoSource, TemplateDirSource, StarterFilesSource │ ├── criteria/ # Criterion checker plugins (one file per type) @@ -138,7 +139,7 @@ action.yml # Published composite GitHub Action (coder-ev - **Dataset fan-out**: `TaskDefinition.dataset` (inline rows or JSONL path) expands a single task into N row-tasks with `${row.}` substitution in `initial_prompt` and `success_criteria` string fields. Expansion runs in `task_loader.expand_dataset` **before** variant resolution, so variants cannot override the dataset. Row sampling: CLI `--sample N` (fixed-seed uniform-random N over the whole dataset) overrides `--sample-per-stratum N` / `dataset.sample_per_stratum` (stratified random N-per-stratum, keyed on `stratify_field`, default `expected_skill` — for classification suites like activation). Stratified sampling (whether the N-per-stratum count comes from the **CLI** `--sample-per-stratum` flag or **YAML** `dataset.sample_per_stratum`) is **nondeterministic** by default — it re-draws each run (so the nightly activation suite broadens coverage over time). Set `dataset.sample_seed` to pin a reproducible sample; an explicit seed always wins. (Only `--sample N` uses a fixed seed, since a smoke test wants the same N rows each run.) - **Per-criterion aggregation**: Each `BaseCriterion` subclass exposes `aggregate(criterion, per_row_results) -> CriterionAggregate | None`. Default emits `count / mean / median / std / min / max` so every criterion is suite-thresholdable for free. Classification-style criteria return `ClassificationCriterionResult` (subclass of `CriterionResult`) and layer accuracy / P/R/F1 / confusion via the shared `overlay_classification_metrics` utility. `BaseSuccessCriterion.suite_thresholds` gates the suite on those metrics; CLI exits non-zero on any gate failure. - **Sub-agent token accounting**: There is NO separate per-sub-agent field. Every sub-agent generation is captured as a `parent_tool_use_id`-tagged `AssistantMessage` in the turn transcript, so per-sub-agent usage is derived by grouping those messages on that id (the evalboard's `aggregateSubAgentUsage` does exactly this). Claude bubbles its sub-agent's intermediate generations into the parent stream natively, and the **terminal** generation (delivered as the Agent tool result, never streamed) is synthesized into one via `_synthesize_subagent_terminal_message` from `tool_use_result.usage`. Codex reconstructs all child generations from the child rollout (`_recover_subagent_tool_calls`). The turn total already includes sub-agent cost — Claude via the SDK's cumulative `model_usage`; Codex via `_fold_subagent_tokens`, which folds the child messages (their real per-generation tokens) into the parent total. `CommandTelemetry.result_summary` is stored **untruncated** (no 200-char cap) so sub-agent returns are preserved whole. Set `CODER_EVAL_RAW_SDK_LOG=1` to dump every raw SDK event to the task log for inspection. -- **Reconciliation message (stream self-reconciles to the turn total)**: The per-message stream consistently under-reports the authoritative turn total — a fixed prompt slice (~512 input tokens on Claude) is billed on no SDK-emitted message, and sub-agent input/cache only partially bubbles up. So `EventCollector.build_turn_record` appends one synthetic `ReconciliationMessage` (`role="reconciliation"`, in the `TranscriptMessage` union) per turn, carrying the per-bucket residual = `token_usage` − Σ(assistant message buckets). The invariant: **summing the four token buckets across `TurnRecord.messages` (assistant + reconciliation) equals `token_usage` exactly**, for both Claude and Codex (Codex's stream is already complete after `_recover_subagent_tool_calls`, so its residual is usually 0 and no entry is emitted). This is what lets the evalboard SUM the message stream as the source of truth instead of reading a separate aggregate ("agent tokens"): `selectTokenTotals` returns the stream sum whenever a reconciliation entry is present, and the timeline renders it as its own row. It is agent-agnostic (booked at the single `EventCollector` seam), carries no cost (cost stays on `token_usage`), and is excluded from generation/turn counts and the cost simulator. The Python `token_usage`/`total_token_usage` aggregate is unchanged and still authoritative for budget/judges/reports. +- **Reconciliation message (stream self-reconciles to the turn total)**: The per-message stream consistently under-reports the authoritative turn total — a fixed prompt slice (~512 input tokens on Claude) is billed on no SDK-emitted message, and sub-agent input/cache only partially bubbles up. So `EventCollector.build_turn_record` appends one synthetic `ReconciliationMessage` (`role="reconciliation"`, in the `TranscriptMessage` union) per turn, carrying the per-bucket residual = `token_usage` − Σ(assistant message buckets). The invariant: **summing the four token buckets across `TurnRecord.messages` (assistant + reconciliation) equals `token_usage` exactly**, for both Claude and Codex (Codex's stream is already complete after `_recover_subagent_tool_calls`, so its residual is usually 0 and no entry is emitted). This is what lets the evalboard SUM the message stream as the source of truth instead of reading a separate aggregate ("agent tokens"): `selectTokenTotals` returns the stream sum whenever a reconciliation entry is present, and the timeline renders it as its own row. It is agent-agnostic (booked at the single `EventCollector` seam), carries no cost (cost stays on `token_usage`), and is excluded from generation/turn counts and the cost simulator. The LiteLLM open-weight actual-cost join (`litellm_cost.apply_actual_cost`) deliberately writes cost at the TURN level only (`token_usage.total_cost_usd` = the real OpenRouter bill) plus the per-call `TurnRecord.provider_call_costs` audit record; it does NOT touch the message token buckets, so `EventCollector` stays the single writer and this invariant holds on every backend. The Python `token_usage`/`total_token_usage` aggregate is unchanged and still authoritative for budget/judges/reports. - **sandbox isolation**: Tasks that don't need MCP servers should set `setting_sources: []` in their `agent:` block to isolate the sandbox from the host project's CLAUDE.md and settings. Without this, the host project's CLAUDE.md (often 20 KB+) is injected into every API call, inflating cache-creation tokens and cost significantly. - **Run-time caps (non-criterion enforcement)**: `TaskDefinition.run_limits` (`RunLimits` model) is the single namespace for all run-time caps — `max_turns` / `task_timeout` / `turn_timeout` (structural) and `max_input_tokens` / `max_output_tokens` / `max_total_tokens` / `max_usd` (cumulative budget). Token/USD breaches abort with `FinalStatus.TOKEN_BUDGET_EXCEEDED` or `COST_BUDGET_EXCEEDED` (both `category == "failed"`). Structural caps are set from the CLI via `-D run_limits.max_turns=…` / `-D run_limits.task_timeout=…` / `-D run_limits.turn_timeout=…` (field-merged into `run_limits`); budget caps via `-D run_limits.max_usd=…` etc. or YAML. Layered config uses field-merge — a variant block overrides individual keys without replacing the task's block. - **Early stop on criterion (opt-in)**: `run_limits.stop_early` (default off) ends a single-shot Claude run early once the run's **armed** criteria are decided, so a raised `max_turns` isn't wasted on the smoke flavor. A criterion is armed by `stop_when: pass|fail|decided|auto`; only criteria that can decide from a partial trajectory may arm (non-empty `live_stop_polarities` ClassVar + `live_verdict` override — currently `skill_triggered`, `command_executed`; CE025 keeps the two consistent). `decided` arms **both** polarities; `auto` arms whichever polarities **this instance** can decide — the value for dataset-fanned criteria whose positive/distractor role flips per row. Stop rule: the pass-stop fires when every **pass-armed** criterion live-passes (fail-armed distractors are not required to pass; zero pass-armed ⇒ never pass-stops); the fail-stop fires on the first fail-armed live-fail but is **deferred while any pass-armed criterion is undecided** — a distractor misfire must not truncate a positive row's recall signal, so the latched misfire fires once the positives resolve (or the run continues to the cap). A fail-stop is therefore verdict-preserving; a pass-stop can miss a *later* distractor misfire, so authoritative P/R/F1 comes from a `stop_early: false` run. Driven by `orchestration/early_stop.py::EarlyStopWatcher` through the Claude agent's cooperative `should_stop` seam (tool-call granularity, no SIGKILL); live verdicts only *trigger* the stop — the standard `check_all_async` on the frozen trajectory is authoritative. An early-stopped run gates on the **armed subset** (`EvaluationResult.armed_criteria_passed`); a completed run gates on the full set. Every unsupported use is a hard error at resolution (plan *and* run), and a runtime verdict bug **fails open** to a full run. Surfaces: `EarlyStopInfo`, report notes/badges, `stopped_early` run.json rows, `EarlyStopped`/`EarlyStopReason` telemetry dims. Worked rationale: docs/TASK_DEFINITION_GUIDE.md § `stop_early`. Defaults off ⇒ behavior byte-for-behavior unchanged. diff --git a/docs/REPORT_SCHEMA.md b/docs/REPORT_SCHEMA.md index 76ec85df..3a261b94 100644 --- a/docs/REPORT_SCHEMA.md +++ b/docs/REPORT_SCHEMA.md @@ -168,15 +168,22 @@ fields so subclass keys round-trip. `iteration`, `user_input`, `agent_output`, `commands` (`list[CommandTelemetry]`), `timestamp`, `duration_seconds`, `token_usage`, `model_used`, `assistant_turn_count`, `messages` (`list[TranscriptMessage]`, discriminated on `role`: -`user`/`assistant`/`reconciliation`), `num_turns`, `max_turns_exhausted`, +`user`/`assistant`/`reconciliation`), `provider_call_costs` +(`list[ProviderCallCost]` — one row per real upstream call with its ACTUAL cost + +cache buckets + routed `provider`, captured proxy-side on the LiteLLM open-weight +backend and rendered by the evalboard as a per-call table; empty on every other +backend), `num_turns`, `max_turns_exhausted`, `result_summary` (`{is_error, subtype, stop_reason, result}`), `crashed`, `crash_reason`. > **Token invariant.** Summing the four token buckets across `messages` > (assistant + the synthetic `reconciliation` entry) equals `token_usage` exactly. > The `reconciliation` message carries the residual the per-message stream -> under-reports; it has no cost and is excluded from turn/generation counts. See the -> [Claude Code guide](agents/CLAUDE_CODE.md#telemetry). +> under-reports; it has no cost and is excluded from turn/generation counts. The +> LiteLLM actual-cost join writes cost at the TURN level only (`token_usage.total_cost_usd` +> = the real bill) plus the per-call `provider_call_costs` audit record — it does +> NOT touch the message token buckets, so this invariant holds on every backend. +> See the [Claude Code guide](agents/CLAUDE_CODE.md#telemetry). ### EarlyStopInfo diff --git a/evalboard/app/runs/[id]/[...task]/_sections.tsx b/evalboard/app/runs/[id]/[...task]/_sections.tsx index 13ef0fd9..71bbf2d6 100644 --- a/evalboard/app/runs/[id]/[...task]/_sections.tsx +++ b/evalboard/app/runs/[id]/[...task]/_sections.tsx @@ -9,6 +9,7 @@ import type { MessageEvent, MessageToolUse, SubAgentTotals, + TaskDetail, TokenTotals, ToolCall, } from "@/lib/runs"; @@ -533,6 +534,126 @@ export function MessageTimelineSection({ ); } +// Short-form a call id for the table (full ids are long provider hashes). Keeps +// the head/tail so a row is still identifiable at a glance. "—" when absent. +function shortCallId(id: string | null): string { + if (!id) return "—"; + return id.length > 14 ? `${id.slice(0, 6)}…${id.slice(-4)}` : id; +} + +function fmtCallTok(n: number | null): string { + return n != null ? fmtCompact(n) : "—"; +} + +// Per-call actual cost + cache table, sourced from each turn's +// provider_call_costs audit rows (LiteLLM/open-weight backend). One sub-table +// per turn with calls, plus a per-turn total. Renders nothing on Claude/Bedrock +// runs (empty providerCalls). +export function ProviderCallTableSection({ + providerCalls, +}: { + providerCalls: TaskDetail["providerCalls"]; +}) { + if (providerCalls.length === 0) return null; + const totalCalls = providerCalls.reduce((s, t) => s + t.calls.length, 0); + return ( +
+

+ Provider calls ({totalCalls}) +

+

+ Actual per-call cost + cache captured from the provider (LiteLLM + backend). The turn total is the real bill. +

+ {providerCalls.map(({ iteration, calls }) => { + const turnTotal = calls.reduce( + (s, c) => s + (c.costUsd ?? 0), + 0, + ); + return ( +
+
+ Turn {iteration + 1} +
+ + + + + + + + + + + + + + + {calls.map((c, i) => ( + + + + + + + + + + ))} + + + + + +
+ Call + + Provider + + Input + + Cache read + + Cache write + + Output + + Cost ($) +
+ {shortCallId(c.callId)} + + {c.provider ?? "—"} + + {fmtCallTok(c.inputTokens)} + + {fmtCallTok(c.cacheReadTokens)} + + {fmtCallTok(c.cacheWriteTokens)} + + {fmtCallTok(c.outputTokens)} + + {c.costUsd != null + ? fmtUsd(c.costUsd) + : "—"} +
+ Turn total + + {fmtUsd(turnTotal)} +
+ +
+ ); + })} +
+ ); +} + // Owns the cost-simulator lever state and renders BOTH the message timeline and // the simulator beneath it. Lifting the levers here lets the timeline show each // message's projected token Δ inline (the "message view" of the diff) driven by diff --git a/evalboard/app/runs/[id]/[...task]/page.tsx b/evalboard/app/runs/[id]/[...task]/page.tsx index 4676d04d..4f5589fa 100644 --- a/evalboard/app/runs/[id]/[...task]/page.tsx +++ b/evalboard/app/runs/[id]/[...task]/page.tsx @@ -23,6 +23,7 @@ import { CriteriaSection, FlowDebugSection, LogTailSection, + ProviderCallTableSection, ToolTimelineSection, } from "./_sections"; @@ -301,6 +302,7 @@ export default async function TaskPage({ recordedCostUsd={task.totalCostUsd} /> )} + {(task.toolCalls.length > 0 || task.finalAssistantText) && ( { + test("prices a message from the rate card (no cost_usd field present)", () => { + const turns: TurnEntry[] = [ + { + model_used: "claude-sonnet-4-6", + messages: [ + { + role: "assistant", + message_id: "m1", + model: "claude-sonnet-4-6", + started_at: "2026-01-01T00:00:00.000Z", + completed_at: "2026-01-01T00:00:01.000Z", + generation_duration_ms: 1000, + content_blocks: [{ block_type: "text", text: "hi" }], + input_tokens: 1000, + output_tokens: 2000, + }, + ], + }, + ]; + const [asst] = parseMessages(turns); + // claude-sonnet-4-6: input 3 / output 15 per MTok → (1000*3 + 2000*15)/1e6 = 0.033 + expect(asst.costUsd).toBeCloseTo(0.033, 9); + }); + + test("leaves an unpriced (OpenRouter) model at null — no inline actual cost", () => { + const turns: TurnEntry[] = [ + { + model_used: "deepseek/deepseek-v4-pro", + messages: [ + { + role: "assistant", + message_id: "m1", + model: "deepseek/deepseek-v4-pro", + started_at: "2026-01-01T00:00:00.000Z", + completed_at: "2026-01-01T00:00:01.000Z", + generation_duration_ms: 1000, + content_blocks: [{ block_type: "text", text: "hi" }], + input_tokens: 715, + cache_read_tokens: 4096, + output_tokens: 8, + }, + ], + }, + ]; + const [asst] = parseMessages(turns); + // OpenRouter models are intentionally unpriced in the rate card → null. + expect(asst.costUsd).toBeNull(); + }); +}); diff --git a/evalboard/lib/__tests__/parseMessages.test.ts b/evalboard/lib/__tests__/parseMessages.test.ts index e3bd183a..ba21f4f7 100644 --- a/evalboard/lib/__tests__/parseMessages.test.ts +++ b/evalboard/lib/__tests__/parseMessages.test.ts @@ -770,19 +770,17 @@ describe("parseMessages — reconciliation entry", () => { expect(recon.toolUses).toEqual([]); }); - test("prices the reconciliation row from the turn model so the Cost column sums to the turn total", () => { - // Mirrors the OpenRouter/LiteLLM sparse-stream case that motivated this: - // the per-message stream carries only output, and the bulk of the input + - // cache-read tokens land in the single reconciliation row. Before the fix - // that row read "—" for cost, so the Cost column summed to far less than - // the real total. + test("prices the reconciliation row from the turn model (Bedrock open-weight: static rates are authoritative)", () => { + // Bedrock open-weight runs on fixed Bedrock rates (like Claude) with no + // OpenRouter actual-cost capture, so static pricing IS correct here and the + // reconciliation row must carry it (else the Cost column under-sums). const turns: TurnEntry[] = [ { - model_used: "deepseek/deepseek-v4-pro", + model_used: "zai.glm-5", messages: [ { role: "assistant", - model: "deepseek/deepseek-v4-pro", + model: "zai.glm-5", started_at: "2026-01-01T00:00:00.000Z", completed_at: "2026-01-01T00:00:01.000Z", generation_duration_ms: 1000, @@ -804,13 +802,45 @@ describe("parseMessages — reconciliation entry", () => { ]; const events = parseMessages(turns); const [asst, recon] = events; - // deepseek-v4-pro: in 0.435 / out 0.87 / cacheRead 0.003625 per MTok. - // reconciliation = 100000*0.435 + 300000*0.003625 = 44587.5 tok-$ / 1e6. - expect(recon.model).toBe("deepseek/deepseek-v4-pro"); - expect(recon.costUsd).toBeCloseTo(0.0445875, 9); - // The whole point: assistant + reconciliation sum to the turn total - // (1000*0.87/1e6 + 0.0445875), instead of the reconciliation row reading "—". - expect((asst.costUsd ?? 0) + (recon.costUsd ?? 0)).toBeCloseTo(0.0454575, 9); + // zai.glm-5: in 1.2 / out 3.84 / cacheRead 0 per MTok. + // reconciliation = 100000*1.2 + 300000*0 = 120000 tok-$ / 1e6 = 0.12. + expect(recon.model).toBe("zai.glm-5"); + expect(recon.costUsd).toBeCloseTo(0.12, 9); + expect((asst.costUsd ?? 0) + (recon.costUsd ?? 0)).toBeCloseTo(0.12384, 9); + }); + + test("does NOT statically price OpenRouter models (their cost is the captured actual)", () => { + // OpenRouter routes per-request, so a static headline rate is wrong; the + // evalboard deliberately leaves these unpriced (→ "—") and shows the real + // per-call cost in the per-call table (ProviderCallTableSection) instead. + const turns: TurnEntry[] = [ + { + model_used: "deepseek/deepseek-v4-pro", + messages: [ + { + role: "assistant", + model: "deepseek/deepseek-v4-pro", + started_at: "2026-01-01T00:00:00.000Z", + completed_at: "2026-01-01T00:00:01.000Z", + generation_duration_ms: 1000, + content_blocks: [{ block_type: "text", text: "hi" }], + input_tokens: 0, + output_tokens: 1000, + cache_read_tokens: 0, + }, + { + role: "reconciliation", + input_tokens: 100000, + output_tokens: 0, + cache_read_tokens: 300000, + note: "billed but not streamed", + }, + ], + }, + ]; + const [asst, recon] = parseMessages(turns); + expect(asst.costUsd).toBeNull(); // no static price for OpenRouter + expect(recon.costUsd).toBeNull(); }); test("reconciliation cost stays null when the turn model is absent (legacy runs)", () => { diff --git a/evalboard/lib/__tests__/pricing-parity.test.ts b/evalboard/lib/__tests__/pricing-parity.test.ts index 648f49f6..7e5e1a67 100644 --- a/evalboard/lib/__tests__/pricing-parity.test.ts +++ b/evalboard/lib/__tests__/pricing-parity.test.ts @@ -85,6 +85,13 @@ describe("pricing.ts ↔ pricing.py parity", () => { "gpt-5.6-sol", "gpt-5.6-terra", "gpt-5.6-luna", + // OpenRouter open-weight models: priced in pricing.py only for the Python + // max_usd static fallback. The evalboard deliberately does NOT statically + // price them — OpenRouter routes per-request, so it shows the captured + // ACTUAL per-call cost instead (see the per-call table, provider_call_costs). + "moonshotai/kimi-k3", + "z-ai/glm-5.2", + "deepseek/deepseek-v4-pro", ]); test("every pricing.py model is mirrored in pricing.ts or explicitly unmirrored", () => { diff --git a/evalboard/lib/__tests__/providerCalls.test.ts b/evalboard/lib/__tests__/providerCalls.test.ts new file mode 100644 index 00000000..9755fd0f --- /dev/null +++ b/evalboard/lib/__tests__/providerCalls.test.ts @@ -0,0 +1,149 @@ +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; +import { parseProviderCalls, type TurnEntry } from "../runs"; + +// parseProviderCalls maps each turn's snake-case provider_call_costs audit rows +// into the camelCase per-call table shape, keyed by iteration, skipping turns +// with no rows. +describe("parseProviderCalls", () => { + test("maps rows and skips empty turns", () => { + const turns: TurnEntry[] = [ + { provider_call_costs: [] }, // skipped + { + provider_call_costs: [ + { + call_id: "call-abc", + provider: "openrouter", + cost_usd: 0.0021, + input_tokens: 715, + cache_read_tokens: 4096, + cache_write_tokens: 0, + output_tokens: 8, + }, + ], + }, + ]; + const out = parseProviderCalls(turns); + expect(out).toHaveLength(1); + expect(out[0].iteration).toBe(1); + expect(out[0].calls[0]).toEqual({ + callId: "call-abc", + provider: "openrouter", + costUsd: 0.0021, + inputTokens: 715, + cacheReadTokens: 4096, + cacheWriteTokens: 0, + outputTokens: 8, + }); + }); + + test("is empty when no turn carries provider_call_costs", () => { + expect(parseProviderCalls([{ messages: [] }])).toEqual([]); + }); + + test("nulls absent fields", () => { + const out = parseProviderCalls([ + { provider_call_costs: [{ cost_usd: 0.5 }] }, + ]); + expect(out[0].calls[0]).toEqual({ + callId: null, + provider: null, + costUsd: 0.5, + inputTokens: null, + cacheReadTokens: null, + cacheWriteTokens: null, + outputTokens: null, + }); + }); +}); + +// End-to-end: a TaskDetail parsed off disk exposes providerCalls from the turn's +// provider_call_costs. Mirrors collect.test.ts's env-stub + fresh-import pattern +// so RUNS_DIR resolves to a throwaway dir. +const RUN = "2026-01-01_00-00-00"; +const TASK = "demo-task"; +let tmp: string; + +async function write(rel: string, body: string): Promise { + const abs = path.join(tmp, rel); + await fs.mkdir(path.dirname(abs), { recursive: true }); + await fs.writeFile(abs, body); +} + +async function loadRuns() { + vi.resetModules(); + vi.stubEnv("EVALBOARD_LOCAL_RUNS_DIR", tmp); + return import("../runs"); +} + +beforeEach(async () => { + tmp = await fs.mkdtemp(path.join(os.tmpdir(), "evalboard-provcalls-")); + await write( + `${RUN}/run.json`, + JSON.stringify({ + run_id: RUN, + task_results: [{ task_id: TASK, status: "success" }], + }), + ); + await write( + `${RUN}/default/${TASK}/00/task.json`, + JSON.stringify({ + final_status: "success", + iterations: [ + { + model_used: "deepseek/deepseek-v4-pro", + provider_call_costs: [ + { + call_id: "call-1", + provider: "openrouter", + cost_usd: 0.0012, + input_tokens: 100, + cache_read_tokens: 200, + cache_write_tokens: 50, + output_tokens: 25, + }, + { + call_id: "call-2", + provider: "openrouter", + cost_usd: 0.0034, + input_tokens: 300, + cache_read_tokens: 0, + cache_write_tokens: 0, + output_tokens: 40, + }, + ], + }, + ], + }), + ); +}); + +afterEach(async () => { + vi.unstubAllEnvs(); + await fs.rm(tmp, { recursive: true, force: true }); +}); + +describe("readTaskDetail: providerCalls", () => { + test("exposes the per-call rows from provider_call_costs", async () => { + const { readTaskDetail } = await loadRuns(); + const detail = await readTaskDetail(RUN, TASK); + expect(detail).not.toBeNull(); + expect(detail?.providerCalls).toHaveLength(1); + const turn = detail!.providerCalls[0]; + expect(turn.iteration).toBe(0); + expect(turn.calls).toHaveLength(2); + expect(turn.calls[0]).toEqual({ + callId: "call-1", + provider: "openrouter", + costUsd: 0.0012, + inputTokens: 100, + cacheReadTokens: 200, + cacheWriteTokens: 50, + outputTokens: 25, + }); + expect(turn.calls[1].callId).toBe("call-2"); + expect(turn.calls[1].costUsd).toBe(0.0034); + }); +}); diff --git a/evalboard/lib/pricing.ts b/evalboard/lib/pricing.ts index 93ac23c8..ad6d0293 100644 --- a/evalboard/lib/pricing.ts +++ b/evalboard/lib/pricing.ts @@ -50,15 +50,16 @@ export const PRICING: Record = { "gemini-3.1-pro-preview-customtools": p(2, 12, 2, 0.2), "gemini-3.5-flash": p(1.5, 9, 1.5, 0.15), "gemini-3-flash-preview": p(1.5, 9, 1.5, 0.15), - // OpenRouter open-weight models (litellm backend). Mirror of pricing.py; - // these providers cache implicitly (cache_write == input, unused) so only - // cache_read carries a discounted rate. NOTE: OpenRouter routes per-request, - // so these rates are only accurate when the litellm config pins the provider - // (sort: price) — otherwise the billed rate can differ from the headline. - "moonshotai/kimi-k3": p(3, 15, 3, 0.3), - "z-ai/glm-5.2": p(0.826, 2.596, 0.826, 0.1534), - "deepseek/deepseek-v4-pro": p(0.435, 0.87, 0.435, 0.003625), + // OpenRouter open-weight models (litellm backend) are DELIBERATELY NOT priced + // here. OpenRouter routes per-request, so a static headline rate is wrong (the + // billed rate depends on the provider it landed on), and there is no per-bucket + // rate to show. Instead the harness captures each call's ACTUAL cost proxy-side + // and the detail view renders it per call (TurnRecord.provider_call_costs → + // ProviderCallTableSection); a static estimate here would only reintroduce the + // wrong number. See pricing.py (kept for the Python-side max_usd fallback). // Bedrock open-weight models (litellm backend, eu-north-1). Mirror of pricing.py. + // These run on Bedrock (fixed rates, like Claude) with NO OpenRouter actual-cost + // capture, so static pricing is correct and required here. // The recorded model_used arrives prefixed (e.g. "converse/zai.glm-5"), so // resolvePricing strips the routing/region prefixes before lookup. "deepseek.v3.2": p(0.74, 2.22, 0.74, 0), diff --git a/evalboard/lib/runs.ts b/evalboard/lib/runs.ts index 45de6f98..fc141b0f 100644 --- a/evalboard/lib/runs.ts +++ b/evalboard/lib/runs.ts @@ -187,11 +187,13 @@ export interface MessageEvent { // cascade-aware thinking-cost simulator to price each message. null when // no raw in the group recorded it. model: string | null; - // True cost in USD for this message's API call, priced from the per-message - // token buckets and model against the shared rate table (lib/pricing.ts). - // The SDK only reports a cumulative per-turn cost, so this is the - // rate-accurate per-message attribution. null when the model is unpriced or - // no token figure was recorded (older runs). + // Rate-card ESTIMATE of this message's cost in USD, priced from the + // per-message token buckets and model against the shared rate table + // (lib/pricing.ts). The SDK only reports a cumulative per-turn cost, so this + // is the rate-accurate per-message attribution. Actual per-call cost (LiteLLM + // backend) is NOT distributed here — it lives in the separate providerCalls + // per-call table. null when the model is unpriced or no token figure was + // recorded (older runs). costUsd: number | null; // Only set on a `reconciliation` row: the human-readable explanation of why // these tokens are unattributed (e.g. fixed prompt overhead + sub-agent @@ -273,6 +275,26 @@ export interface TaskDetail extends TaskResultSummary { // the Agent row). The cost simulator consumes the values via Object.values(). // Empty for runs/turns with no spawned sub-agents. subAgentUsageByToolId: Record; + // Per-call ACTUAL cost + cache audit rows, grouped by turn iteration. Only + // turns whose `provider_call_costs` list is non-empty appear (LiteLLM/ + // open-weight backend; empty on Claude/Bedrock). Rendered as a standalone + // per-call table, NOT distributed onto individual transcript messages. + providerCalls: { iteration: number; calls: ProviderCallEntry[] }[]; +} + +// One per-call audit row from `TurnRecord.provider_call_costs` — the actual +// cost + cache split the LiteLLM proxy captured for a single upstream call +// (open-weight backend). The turn-level total_cost_usd remains the real bill; +// these rows are an itemized breakdown surfaced as a separate table. Empty list +// on non-LiteLLM backends. +export interface ProviderCallEntry { + callId: string | null; + provider: string | null; + costUsd: number | null; + inputTokens: number | null; + cacheReadTokens: number | null; + cacheWriteTokens: number | null; + outputTokens: number | null; } // Full per-sub-agent token breakdown (all components, cache-read included). @@ -1195,6 +1217,18 @@ interface MessageEntry { note?: string | null; } +// One raw per-call audit row on a turn's `provider_call_costs` list. Snake-case +// as serialized in task.json; mapped to ProviderCallEntry (camelCase). +interface ProviderCallEntryRaw { + call_id?: string | null; + provider?: string | null; + cost_usd?: number | null; + input_tokens?: number | null; + cache_read_tokens?: number | null; + cache_write_tokens?: number | null; + output_tokens?: number | null; +} + export interface TurnEntry { commands?: CommandEntry[]; messages?: MessageEntry[]; @@ -1202,6 +1236,9 @@ export interface TurnEntry { // reconciliation row, which carries no model of its own. model_used?: string | null; token_usage?: TokenUsageEntry | null; + // Per-call actual cost + cache audit rows (LiteLLM/open-weight backend); + // empty/absent on Claude/Bedrock. Surfaced as a standalone per-call table. + provider_call_costs?: ProviderCallEntryRaw[]; result_summary?: { result?: string | null; stop_reason?: string | null; @@ -1649,6 +1686,8 @@ export function parseMessages(turns: TurnEntry[]): MessageEvent[] { thinkingOutputTokens: haveThinkingOut ? thinkingOutSum : null, textOutputTokens, model, + // Rate-card estimate for this message (actual per-call cost lives + // in the separate providerCalls table, not distributed here). costUsd: messageCostUsd({ model, inputTokens: haveInputTok ? inputTokSum : null, @@ -1728,6 +1767,9 @@ export function parseMessages(turns: TurnEntry[]): MessageEvent[] { // task total_cost_usd is unaffected (it reads the backend aggregate, // not a sum of per-message costUsd), so there is no double-count. model: turn.model_used ?? null, + // Price the residual tokens from the rate card (Claude/Bedrock). + // Actual per-call cost (LiteLLM backend) is not distributed onto + // messages — it lives in the separate providerCalls table. costUsd: messageCostUsd({ model: turn.model_used ?? null, inputTokens: typeof msg.input_tokens === "number" ? msg.input_tokens : null, @@ -1743,6 +1785,34 @@ export function parseMessages(turns: TurnEntry[]): MessageEvent[] { return out; } +// Project each turn's `provider_call_costs` audit rows (LiteLLM/open-weight +// backend) into the per-call table shape, keyed by iteration index. Turns with +// no rows are skipped, so the result is empty on Claude/Bedrock runs and the +// table doesn't render. +export function parseProviderCalls( + turns: TurnEntry[], +): { iteration: number; calls: ProviderCallEntry[] }[] { + const out: { iteration: number; calls: ProviderCallEntry[] }[] = []; + turns.forEach((turn, iteration) => { + const raw = turn.provider_call_costs ?? []; + if (raw.length === 0) return; + const calls: ProviderCallEntry[] = raw.map((c) => ({ + callId: typeof c.call_id === "string" ? c.call_id : null, + provider: typeof c.provider === "string" ? c.provider : null, + costUsd: typeof c.cost_usd === "number" ? c.cost_usd : null, + inputTokens: typeof c.input_tokens === "number" ? c.input_tokens : null, + cacheReadTokens: + typeof c.cache_read_tokens === "number" ? c.cache_read_tokens : null, + cacheWriteTokens: + typeof c.cache_write_tokens === "number" ? c.cache_write_tokens : null, + outputTokens: + typeof c.output_tokens === "number" ? c.output_tokens : null, + })); + out.push({ iteration, calls }); + }); + return out; +} + function parseToolCalls(turns: TurnEntry[], max = 200): ToolCall[] { const out: ToolCall[] = []; let i = 0; @@ -1885,6 +1955,7 @@ export async function readTaskDetail( const flowDebug = parseFlowDebug(criteria); const toolCalls = parseToolCalls(task?.iterations ?? []); const messages = parseMessages(task?.iterations ?? []); + const providerCalls = parseProviderCalls(task?.iterations ?? []); // The per-message stream is the authoritative cumulative bill and is // preferred whenever it carries token data. Iteration token_usage comes // from the SDK ResultMessage snapshot, which is NOT cumulative for @@ -1937,6 +2008,7 @@ export async function readTaskDetail( messages, tokens, subAgentUsageByToolId, + providerCalls, }; } diff --git a/litellm/cost_logger.py b/litellm/cost_logger.py new file mode 100644 index 00000000..d0396776 --- /dev/null +++ b/litellm/cost_logger.py @@ -0,0 +1,235 @@ +"""LiteLLM proxy success-logging callback: capture per-call ACTUAL cost + cache. + +Why this exists +--------------- +For coder_eval's open-weight (``litellm``) backend the Claude Code binary calls +this proxy in **Anthropic Messages** format. OpenRouter's real per-call +``usage.cost`` and cache-read tokens are dropped in the OpenAI->Anthropic +translation, so the Python harness only ever sees an aggregate turn total priced +at Claude list rates. This callback runs **inside the proxy**, where each call's +OpenAI-shaped response still carries the real numbers, and appends one JSONL +record per call to ``$LITELLM_COST_LOG``. The harness later joins those records +back to a run/task/turn by the ``x-ce-*`` correlation headers the agent stamped +via ``ANTHROPIC_CUSTOM_HEADERS``. + +IMPORTANT: read ``response_obj.usage.cost`` (OpenRouter's real, routed-provider +cost), **never** ``standard_logging_object.response_cost`` (LiteLLM's own map, +which is ``0.0`` for models it doesn't price -- e.g. +``openrouter/deepseek/deepseek-v4-pro``). Requires ``usage: {include: true}`` on +each ``openrouter/*`` model in ``litellm-config.yaml`` so OpenRouter populates +``usage.cost``. + +Register in ``litellm-config.yaml``:: + + litellm_settings: + callbacks: cost_logger.proxy_handler_instance + +This module is intentionally **self-contained** (no ``coder_eval`` import): the +proxy may run in its own ephemeral environment +(``uvx --from 'litellm[proxy]' litellm``). +""" + +from __future__ import annotations + +import json +import math +import os +from typing import Any + + +# LiteLLM is present in the proxy's environment but NOT in coder_eval's test env. +# Guard the import so the pure helpers below stay unit-testable standalone. +try: + from litellm.integrations.custom_logger import CustomLogger +except Exception: # pragma: no cover - only when litellm is absent (i.e. under test) + + class CustomLogger: # type: ignore[no-redef] + """Fallback base so this module imports without litellm installed.""" + + +# Correlation-header prefix, matching claude_code_agent._build_sdk_env's tags. +_TAG_PREFIX = "x-ce-" + + +def _num(value: Any) -> float | int | None: + """Keep only real, FINITE numbers (reject bool — an int subclass — strings, and + NaN/Inf). A non-finite cost would serialize as a bare ``NaN`` token (invalid JSON + that breaks parsing the whole ``task.json``) and make the ``max_usd`` gate's + ``cost > limit`` silently never fire, so it is filtered at this boundary.""" + if isinstance(value, int | float) and not isinstance(value, bool) and math.isfinite(value): + return value + return None + + +def _to_dict(obj: Any) -> dict[str, Any]: + """Coerce a dict / pydantic model / litellm object to a plain dict.""" + if obj is None: + return {} + if isinstance(obj, dict): + return obj + for attr in ("model_dump", "dict"): + fn = getattr(obj, attr, None) + if callable(fn): + # Best-effort coercion: if one dumper raises on a quirky litellm/pydantic + # object, fall through to the next attr (or to {} below) rather than + # propagate — a logging callback must never break the proxy response path. + try: + result = fn() + if isinstance(result, dict): + return result + except Exception: + # This dumper failed on a quirky object; try the next attr / {}. + pass + return {} + + +def build_cost_record( + usage: dict[str, Any] | None, + tags: dict[str, str], + *, + model: str | None, + provider: str | None = None, + call_id: str | None, +) -> dict[str, Any] | None: + """Build one per-call JSONL record from an OpenAI-shaped ``usage`` dict plus + the forwarded ``x-ce-*`` correlation tags. + + Returns ``None`` when there is no usable signal (no correlation tag AND no + cost), so the harness falls back to static pricing instead of recording a + bogus ``$0`` row. + """ + usage = usage or {} + cost = _num(usage.get("cost")) + input_total, cache_read, cache_write, output = _extract_token_buckets(usage) + record: dict[str, Any] = { + "run_id": tags.get(f"{_TAG_PREFIX}run-id"), + "task_id": tags.get(f"{_TAG_PREFIX}task-id"), + "iteration": tags.get(f"{_TAG_PREFIX}iteration"), + "attempt": tags.get(f"{_TAG_PREFIX}attempt"), + "call_id": call_id, + "model": model, + # OpenRouter's actually-routed upstream (e.g. 'fireworks'); best-effort — may + # be null if the Anthropic-translated response doesn't surface it. + "provider": provider, + # OpenRouter's real routed-provider price. NOT LiteLLM's response_cost. + "cost": cost, + "input": input_total, + "cache_read": cache_read, + "cache_write": cache_write, + "output": output, + } + if record["run_id"] is None and cost is None: + return None + return record + + +def _extract_token_buckets( + usage: dict[str, Any], +) -> tuple[int | float | None, int | float | None, int | float | None, int | float | None]: + """Return ``(input_total, cache_read, cache_write, output)`` from a usage dict + of EITHER shape. + + On the Anthropic-inbound ``/v1/messages`` path a call's usage can come back + OpenAI-shaped (``prompt_tokens`` / ``completion_tokens`` / + ``prompt_tokens_details.cached_tokens``) OR Anthropic-shaped (``input_tokens`` + / ``output_tokens`` / ``cache_read_input_tokens`` / + ``cache_creation_input_tokens``). Reading only the OpenAI keys left tokens + ``None`` on the Anthropic-shaped calls (while cost survived). + + ``input_total`` is normalized to the FULL prompt (incl. cached) so the + harness's ``uncached = input - cache_read - cache_write`` holds either way: + OpenAI ``prompt_tokens`` is already the total; Anthropic ``input_tokens`` is + only the uncached slice, so the cache buckets are added back. + """ + details = usage.get("prompt_tokens_details") or {} + cache_read = _num(details.get("cached_tokens")) + if cache_read is None: + cache_read = _num(usage.get("cache_read_input_tokens")) + cache_write = _num(details.get("cache_write_tokens")) + if cache_write is None: + cache_write = _num(usage.get("cache_creation_input_tokens")) + output = _num(usage.get("completion_tokens")) + if output is None: + output = _num(usage.get("output_tokens")) + input_total = _num(usage.get("prompt_tokens")) + if input_total is None: + uncached = _num(usage.get("input_tokens")) + if uncached is not None: + input_total = uncached + (cache_read or 0) + (cache_write or 0) + return input_total, cache_read, cache_write, output + + +def extract_tags(kwargs: dict[str, Any]) -> dict[str, str]: + """Pull the ``x-ce-*`` correlation headers out of the LiteLLM call kwargs. + + Headers surface in different places across LiteLLM versions/paths; merge the + known locations and keep only our ``x-ce-*`` tags (keys lower-cased). + """ + headers: dict[str, Any] = {} + lp = kwargs.get("litellm_params") or {} + for source in ( + (lp.get("metadata") or {}).get("headers"), + (lp.get("proxy_server_request") or {}).get("headers"), + (kwargs.get("proxy_server_request") or {}).get("headers"), + (kwargs.get("metadata") or {}).get("headers"), + ): + if isinstance(source, dict): + headers.update(source) + return {str(k).lower(): str(v) for k, v in headers.items() if str(k).lower().startswith(_TAG_PREFIX)} + + +def append_record(record: dict[str, Any] | None) -> None: + """Append one record as a JSONL line to ``$LITELLM_COST_LOG`` (no-op if unset). + + One ``write()`` of a small (<4 KiB) line in append mode is atomic on POSIX, + so parallel tasks funneled through one proxy interleave lines safely. + """ + path = os.environ.get("LITELLM_COST_LOG") + if not path or record is None: + return + line = json.dumps(record, separators=(",", ":")) + "\n" + with open(path, "a", encoding="utf-8") as fh: + fh.write(line) + + +class CostLogger(CustomLogger): + """LiteLLM success callback → one per-call JSONL cost record.""" + + def _emit(self, kwargs: dict[str, Any], response_obj: Any) -> None: + try: + response = _to_dict(response_obj) + slo = kwargs.get("standard_logging_object") or {} + lp = kwargs.get("litellm_params") or {} + # model surfaces in different places across paths (top-level response, + # kwargs, litellm_params, or the standard logging object). + model = ( + response.get("model") + or kwargs.get("model") + or lp.get("model") + or (slo.get("model") if isinstance(slo, dict) else None) + ) + usage = _to_dict(response.get("usage")) + # OpenRouter surfaces the routed upstream in a few spots depending on the + # path; best-effort (null-tolerant — the harness/table just show blank). + provider = response.get("provider") or usage.get("provider") or response.get("provider_name") + record = build_cost_record( + usage, + extract_tags(kwargs), + model=model, + provider=provider, + call_id=response.get("id"), + ) + append_record(record) + except Exception: + # A logging callback must NEVER break the proxy's response path. + pass + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + self._emit(kwargs, response_obj) + + def log_success_event(self, kwargs, response_obj, start_time, end_time): + self._emit(kwargs, response_obj) + + +# The instance litellm-config.yaml references: `callbacks: cost_logger.proxy_handler_instance`. +proxy_handler_instance = CostLogger() diff --git a/litellm/litellm-config.yaml b/litellm/litellm-config.yaml index a9189740..211cf800 100644 --- a/litellm/litellm-config.yaml +++ b/litellm/litellm-config.yaml @@ -40,41 +40,67 @@ model_list: # OpenRouter models — cost-optimization path. Not on Bedrock; OpenRouter is # OpenAI-format, so LiteLLM does the Anthropic<->OpenAI translation. # - # provider routing (extra_body.provider): OpenRouter is a multi-provider router, - # and its per-request routing is nondeterministic — a request can land on a - # provider far pricier than the model page's headline rate (observed: DeepSeek - # V4 Pro routed to Novita at ~$1.47/M input vs the $0.435/M headline, 3.4x). - # `sort: price` pins the cheapest upstream and `allow_fallbacks: false` forbids - # silently upgrading to a pricier one, so the billed rate is deterministic and - # matches the pricing.py table (and is the cost floor we benchmark for). + # provider routing (extra_body.provider): OpenRouter is a multi-provider router. + # `provider.only` restricts each model to a HARD allowlist of providers we've + # observed to be reliable, `sort: price` tries them cheapest-first, and + # `allow_fallbacks: true` lets routing fall over to the NEXT provider in that set + # when the cheapest is saturated — so a contended provider no longer 429s with + # nowhere to go (the original single-pin failure). Fallback is bounded to the + # vetted `only` set, and cost_logger.py records the ACTUAL routed provider + + # usage.cost per call, so a fallover is both billed correctly and visible (the + # per-call table shows which upstream served each call). Cost is exact regardless + # of which listed provider serves the call. + # + # extra_body.usage.include (set on every model below): ask OpenRouter to return + # the ACTUAL per-call usage.cost + cache tokens for the provider it routed to. The + # Anthropic response the SDK receives has no slot for these, so cost_logger.py (the + # callback in litellm_settings) reads them proxy-side and writes one JSONL record + # per call for the harness to join back by the x-ce-* correlation headers. - model_name: moonshotai/kimi-k3 litellm_params: model: openrouter/moonshotai/kimi-k3 api_key: os.environ/OPENROUTER_API_KEY extra_body: + # usage.include + provider pin — see the OpenRouter note above. + usage: + include: true provider: sort: price - allow_fallbacks: false + allow_fallbacks: true + only: ["baseten", "together", "fireworks"] - model_name: z-ai/glm-5.2 litellm_params: model: openrouter/z-ai/glm-5.2 api_key: os.environ/OPENROUTER_API_KEY extra_body: + # usage.include + provider pin — see the OpenRouter note above. + usage: + include: true provider: sort: price - allow_fallbacks: false + allow_fallbacks: true + only: ["novita", "streamlake", "gmicloud", "alibaba"] - model_name: deepseek/deepseek-v4-pro litellm_params: model: openrouter/deepseek/deepseek-v4-pro api_key: os.environ/OPENROUTER_API_KEY extra_body: + # usage.include + provider pin — see the OpenRouter note above. + usage: + include: true provider: sort: price - allow_fallbacks: false + allow_fallbacks: true + only: ["streamlake", "gmicloud", "novita", "alibaba"] litellm_settings: # Silently drop provider-unsupported OpenAI/Anthropic params rather than 400. drop_params: true + # Per-call ACTUAL-cost + cache capture for the open-weight backend. Appends one + # JSONL record per call to $LITELLM_COST_LOG (set in start-litellm.sh). Reads + # OpenRouter's real usage.cost, NOT LiteLLM's response_cost (0.0 for models it + # doesn't price). See cost_logger.py for the full rationale. + callbacks: cost_logger.proxy_handler_instance general_settings: # Read the virtual key from the environment — never hardcode a secret here. diff --git a/litellm/start-litellm.sh b/litellm/start-litellm.sh index da3192db..169ece47 100755 --- a/litellm/start-litellm.sh +++ b/litellm/start-litellm.sh @@ -53,6 +53,15 @@ export OPENROUTER_API_KEY="${OPENROUTER_API_KEY:-$(read_env OPENROUTER_API_KEY)} export LITELLM_MASTER_KEY="${LITELLM_MASTER_KEY:-$(read_env LITELLM_AUTH_TOKEN)}" export LITELLM_MASTER_KEY="${LITELLM_MASTER_KEY:-sk-spike-local}" +# --- per-call cost/cache capture (cost_logger.py) --- +# The success callback appends one JSONL record per call to LITELLM_COST_LOG; +# coder_eval reads the SAME path to join actual OpenRouter cost + cache buckets +# back to each run/task. Make cost_logger importable by the proxy (it lives next +# to this script). PYTHONPATH is honored even inside the uvx-isolated env. +export LITELLM_COST_LOG="${LITELLM_COST_LOG:-$REPO_ROOT/tmp/litellm-costs.jsonl}" +mkdir -p "$(dirname "$LITELLM_COST_LOG")" +export PYTHONPATH="$SCRIPT_DIR${PYTHONPATH:+:$PYTHONPATH}" + # --- preflight: fail loud instead of a runtime 401 --- [ -f "$CONFIG" ] || { echo "ERROR: config not found: $CONFIG" >&2; exit 1; } if [ -z "$AWS_BEARER_TOKEN_BEDROCK" ]; then @@ -64,6 +73,7 @@ echo "config : $CONFIG" echo "region : $AWS_REGION" echo "bedrock tok: set (${#AWS_BEARER_TOKEN_BEDROCK} chars)" echo "master key : $LITELLM_MASTER_KEY" +echo "cost log : $LITELLM_COST_LOG" # --- stop any stale proxy on the port (the classic 'creds-less running proxy') --- existing=$(lsof -tiTCP:"$PORT" -sTCP:LISTEN 2>/dev/null || true) @@ -81,6 +91,7 @@ Set these in coder_eval's .env (or shell) to use it: API_BACKEND=litellm LITELLM_BASE_URL=http://localhost:$PORT LITELLM_AUTH_TOKEN=$LITELLM_MASTER_KEY + LITELLM_COST_LOG=$LITELLM_COST_LOG # SAME path as above, so the run joins ACTUAL per-call cost (else static pricing) # then run: coder-eval run --model zai.glm-5 (or deepseek.v3.2 / moonshotai.kimi-k2.5) EOF diff --git a/src/coder_eval/agent.py b/src/coder_eval/agent.py index 4b4510a2..e5ebbf9a 100644 --- a/src/coder_eval/agent.py +++ b/src/coder_eval/agent.py @@ -79,6 +79,13 @@ def __init__(self, config: ClaudeCodeAgentConfig, ...): # between messages override it to True. supports_cooperative_stop: ClassVar[bool] = False + # Capability flag: whether this agent's constructor accepts the ``cost_log_tags`` + # kwarg (proxy-side actual-cost correlation headers for the LiteLLM backend). + # Default False — the agent-agnostic ``create_agent`` factory must only forward + # ``cost_log_tags`` to agents that set this True, else a route-driven kwarg would + # crash every agent (NoOp/Codex/Antigravity/plugins) whose ``__init__`` lacks it. + supports_cost_log_tags: ClassVar[bool] = False + def _begin_turn(self) -> None: """Mark the start of a ``communicate()`` turn: reset the pending slot and bump the iteration counter so a mid-turn failure can be rolled back. diff --git a/src/coder_eval/agents/claude_code_agent.py b/src/coder_eval/agents/claude_code_agent.py index 01ce89ca..71cb2267 100644 --- a/src/coder_eval/agents/claude_code_agent.py +++ b/src/coder_eval/agents/claude_code_agent.py @@ -656,6 +656,10 @@ class ClaudeCodeAgent(Agent[ClaudeCodeAgentConfig]): # ``should_stop`` check runs, so this agent supports early-stop-on-criterion. supports_cooperative_stop: ClassVar[bool] = True + # This agent's __init__ accepts cost_log_tags and stamps them into + # ANTHROPIC_CUSTOM_HEADERS for the proxy-side actual-cost join (LiteLLM backend). + supports_cost_log_tags: ClassVar[bool] = True + def __init__( self, config: ClaudeCodeAgentConfig, @@ -663,6 +667,7 @@ def __init__( *, instance_name: str = "coder", extra_mcp_servers: dict[str, Any] | None = None, + cost_log_tags: dict[str, str] | None = None, ): """Initialize the Claude Code agent. @@ -680,10 +685,21 @@ def __init__( ``_FRAMEWORK_OWNED_SDK_FIELDS`` and explicitly denied via ``sdk_options`` for security. The judge criterion is the only caller today. + cost_log_tags: LiteLLM-only correlation headers (``x-ce-run-id`` / + ``x-ce-task-id`` / ``x-ce-attempt``, with ``x-ce-iteration`` appended + per turn) stamped into ``ANTHROPIC_CUSTOM_HEADERS`` so a proxy-side + cost callback can attribute each call's real cost back to this run. + None on Direct/Bedrock. """ self.config = config self.route = route or DirectRoute() self._extra_mcp_servers = extra_mcp_servers or {} + # Correlation headers stamped on every SDK->proxy request (LiteLLM route + # only), so a proxy-side cost-logging callback can join each call's real + # usage.cost + cache buckets back to this run/task. None => no header + # (Direct/Bedrock, or when the orchestrator supplies none). This turn's + # iteration is appended per-communicate() in _build_claude_query. + self._cost_log_tags = cost_log_tags self.client: ClaudeSDKClient | None = None self.working_directory: Path | None = None # _state / _iteration / _iteration_was_incremented / pending_turn lifecycle @@ -729,6 +745,7 @@ def _build_sdk_env( route: ApiRoute, path_prepend: list[str] | None = None, plugin_tools_dir: str | None = None, + cost_log_tags: dict[str, str] | None = None, ) -> tuple[dict[str, str], str | None]: """Build SDK environment variables and resolve effective model for the given route. @@ -741,6 +758,9 @@ def _build_sdk_env( plugin_tools_dir: Fallback canonical ``node_modules/@uipath`` to export as ``PLUGIN_TOOLS_DIR`` when the process environment doesn't already provide one. An external ``PLUGIN_TOOLS_DIR`` always wins. + cost_log_tags: LiteLLM-only correlation headers stamped (newline-separated + ``Name: Value``) into ``ANTHROPIC_CUSTOM_HEADERS``. Values must be + single-line ASCII (validated here) — the header block is CR/LF-delimited. Returns: Tuple of (env_vars_dict, model_override_or_None). @@ -809,6 +829,23 @@ def _build_sdk_env( env["ANTHROPIC_MODEL"] = cr.model if cr.small_model: env["ANTHROPIC_SMALL_FAST_MODEL"] = cr.small_model + if cost_log_tags: + # Stamp every SDK->proxy request with correlation headers so a + # LiteLLM logging callback can attribute each call's real + # usage.cost + cache buckets back to this run/task/turn. Claude + # Code forwards ANTHROPIC_CUSTOM_HEADERS (newline-separated + # `Name: Value`) verbatim, incl. to a non-anthropic base URL. + # + # Sanitize at the seam: x-ce-task-id carries the author-defined + # task_id/variant_id, so a value with a CR/LF would inject extra + # headers into every SDK->proxy request (forged cost attribution, + # or an auth/routing header override). Non-ASCII also breaks the + # latin-1 header encoding. Reject both loudly rather than emit them. + for name, value in cost_log_tags.items(): + joined = f"{name}{value}" + if "\r" in joined or "\n" in joined or not joined.isascii(): + raise ValueError(f"cost_log_tags {name!r} must be single-line ASCII (got {value!r})") + env["ANTHROPIC_CUSTOM_HEADERS"] = "\n".join(f"{k}: {v}" for k, v in cost_log_tags.items()) return {**base_env, **env}, cr.model raise AssertionError(f"Unhandled route type: {type(route).__name__}") @@ -1117,10 +1154,17 @@ def _build_claude_query( # Build env overrides and resolve model for the configured API route. # Precedence: task/CLI agent.model > route default (e.g. BEDROCK_MODEL). + # Per-turn cost-correlation headers (LiteLLM route only): the run/task tag + # from the orchestrator plus this turn's iteration, so the proxy-side cost + # log can be joined back to the exact turn. + cost_log_tags: dict[str, str] | None = None + if self._cost_log_tags is not None: + cost_log_tags = {**self._cost_log_tags, "x-ce-iteration": str(self._iteration)} env, route_model = self._build_sdk_env( self.route, path_prepend=self._env_path_prepend, plugin_tools_dir=self._plugin_tools_dir, + cost_log_tags=cost_log_tags, ) effective_model = self._resolve_effective_model(self.config.model, env, route_model) diff --git a/src/coder_eval/config.py b/src/coder_eval/config.py index 860b7ea1..c65bb1d6 100644 --- a/src/coder_eval/config.py +++ b/src/coder_eval/config.py @@ -114,6 +114,12 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: litellm_auth_token: str | None = None litellm_model: str | None = None litellm_small_model: str | None = None + # Path to the per-call cost/cache JSONL the LiteLLM proxy's cost_logger writes + # (LITELLM_COST_LOG). Must point at the SAME file the proxy uses (see + # litellm/start-litellm.sh). When set and the file exists, the harness joins each + # call's ACTUAL OpenRouter cost + cache onto the turn (litellm_cost.apply_actual_cost), + # overriding the static rate-card estimate; unset/missing => static pricing (fallback). + litellm_cost_log: str | None = None # Codex settings (CodexAgent). CODEX_MODEL is the fallback model/deployment # used when a task doesn't pin agent.model; CODEX_BASE_URL routes to a custom diff --git a/src/coder_eval/litellm_cost.py b/src/coder_eval/litellm_cost.py new file mode 100644 index 00000000..c1b0dcdc --- /dev/null +++ b/src/coder_eval/litellm_cost.py @@ -0,0 +1,230 @@ +"""Join proxy-captured ACTUAL per-call cost/cache onto a run's turns. + +For the open-weight (LiteLLM) backend the Claude binary's Anthropic transport +drops OpenRouter's real ``usage.cost`` + per-call cache before Python can see it, +so a proxy-side callback (``litellm/cost_logger.py``) writes one JSONL record per +call to ``LITELLM_COST_LOG``. This module reads those records back and joins them +onto the matching turns at the TURN level: + +* the turn's ``token_usage.total_cost_usd`` is overridden with the SUM of its + calls' real cost (the bill), replacing the static rate-card estimate; +* the per-call breakdown is attached as ``TurnRecord.provider_call_costs`` — a + deterministic audit record (one row per real proxy call, with its real cost + + cache buckets + provider) that the evalboard renders as a per-call table. + +Token buckets are LEFT UNTOUCHED (SDK-authoritative): the join only writes cost, +so the ``EventCollector`` remains the single writer of the message token-bucket +invariant. There is deliberately NO per-generation distribution — matching a proxy +call to a transcript generation has no deterministic key (only positional / +output-token heuristics), so that view lives in the per-call table off +``provider_call_costs`` instead of being guessed onto the message stream. + +Coverage. A turn's cost is overridden only when every call that reported usage is +priced. Degenerate calls that report NO usage at all (no cost AND no tokens — seen +occasionally on some providers) are ignored, so one of them can't revert a whole +turn to the static estimate. A call that reports usage but no cost is a genuine +gap: the turn keeps its static estimate (overriding would bill it at $0), no +breakdown is attached, and a warning names the unpriced ids. A turn with no +matching record keeps its static estimate too. + +Retry safety: multiple ``TurnRecord``s can share an ``iteration`` (a crashed +attempt + its retry), and both attempts' proxy calls carry that iteration tag. An +iteration's calls are credited to a single survivor turn (the last with that +iteration THAT HAS GENERATIONS); earlier siblings are zeroed. Crucially the +credit/zero decision is made TOGETHER: a sibling is only zeroed when the survivor +is actually credited with real cost — if the survivor falls back to static, the +sibling keeps its static estimate too, so the iteration's spend is never dropped. + +Transactional: the full plan is computed before any turn is mutated, so a +malformed record (which raises while building the per-call breakdown) aborts the +whole join with the run left untouched — matching the caller's "keeping static +pricing" contract. +""" + +from __future__ import annotations + +import json +import logging +from collections import defaultdict +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from coder_eval.models import AssistantMessage, ProviderCallCost, TokenUsage + + +if TYPE_CHECKING: + from coder_eval.models import EvaluationResult, TurnRecord + + +logger = logging.getLogger(__name__) + + +def load_cost_records(path: str | Path) -> list[dict[str, Any]]: + """Read the proxy's per-call JSONL cost log; tolerant of a missing file, + blank lines, and partial/garbled lines (returns only well-formed dict rows). + + Streamed line-by-line rather than slurped whole, so a long-lived shared log is + not loaded into memory as one giant string. NOTE the log is append-only and, by + default (see ``start-litellm.sh``), shared across runs — it is not rotated here, + so each task's join re-reads the file; scope ``LITELLM_COST_LOG`` to a per-run + path (or rotate it) if it grows large. ``apply_actual_cost`` filters to the + current run/task/attempt, so stale rows are ignored, only re-read. + """ + p = Path(path) + if not p.is_file(): + return [] + records: list[dict[str, Any]] = [] + with p.open(encoding="utf-8") as fh: + for raw in fh: + line = raw.strip() + if not line: + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(obj, dict): + records.append(obj) + return records + + +def _to_call(record: dict[str, Any]) -> ProviderCallCost: + return ProviderCallCost( + call_id=record.get("call_id"), + provider=record.get("provider"), + cost_usd=record.get("cost"), + input_tokens=record.get("input"), + cache_read_tokens=record.get("cache_read"), + cache_write_tokens=record.get("cache_write"), + output_tokens=record.get("output"), + ) + + +def _reports_no_usage(call: ProviderCallCost) -> bool: + """A degenerate call: no cost AND no tokens at all (a null/empty record some + providers occasionally emit). Ignored so it can't block an otherwise-priced turn.""" + return call.cost_usd is None and not any( + (call.input_tokens, call.cache_read_tokens, call.cache_write_tokens, call.output_tokens) + ) + + +def apply_actual_cost( + result: EvaluationResult, + *, + run_id: str, + task_id: str, + attempt: str | None = None, + records: list[dict[str, Any]], +) -> int: + """Override each turn's cost with the proxy's real per-call cost and attach the + per-call breakdown. Mutates ``result`` in place; call BEFORE the run-level token + aggregation so the run total re-derives from the corrected per-turn costs. + + Args: + result: the run's ``EvaluationResult`` (its ``iterations`` are the turns). + run_id / task_id: the correlation tags the agent stamped (``x-ce-run-id`` / + ``x-ce-task-id``); only records matching BOTH are joined. + attempt: the per-attempt nonce (``x-ce-attempt``). When given, records must + also match it — so a re-run into the same ``--run-dir`` (same run_id) does + not re-match, and double-count, a prior attempt's rows. ``None`` disables + the check (records without the tag still join). + records: rows from :func:`load_cost_records`. + + Returns: + The number of turns that received real cost (0 => everything kept its + static estimate, e.g. an empty/mismatched log). + """ + mine = [ + r + for r in records + if r.get("run_id") == run_id + and r.get("task_id") == task_id + and (attempt is None or r.get("attempt") == attempt) + ] + if not mine: + return 0 + + by_iteration: dict[str, list[dict[str, Any]]] = defaultdict(list) + for record in mine: + by_iteration[str(record.get("iteration"))].append(record) + + turns = result.iterations + survivor_index = _survivor_index(turns) + + # Phase 1 — COMPUTE the plan without mutating any turn. Building the per-call + # breakdown (_to_call) validates each record, so a malformed row raises HERE, + # before any mutation, and the caller keeps static pricing for the whole run + # rather than persisting a half-joined mix. + matched_iterations: set[str] = set() + creditable: dict[str, list[ProviderCallCost]] = {} # iteration -> usable calls (only when fully priced) + for iteration, recs in by_iteration.items(): + matched_iterations.add(iteration) + calls = [_to_call(record) for record in recs] + usable = [c for c in calls if not _reports_no_usage(c)] # drop degenerate no-usage calls + unpriced = [c.call_id for c in usable if c.cost_usd is None] + if usable and not unpriced: + creditable[iteration] = usable + else: + reason = ( + f"{len(unpriced)}/{len(usable)} call(s) report usage but no cost {unpriced}" + if usable + else "no usable call" + ) + logger.warning( + "LiteLLM actual-cost: iteration=%s not fully priced (%s); keeping the static estimate", + iteration, + reason, + ) + + # Phase 2 — MUTATE. Only iterations that are creditable touch anything; a + # sibling is zeroed ONLY when its survivor is credited (so an un-creditable + # iteration leaves BOTH the survivor and its crashed sibling on static — the + # spend is never dropped). + applied = 0 + for i, turn in enumerate(turns): + iteration = str(turn.iteration) + calls = creditable.get(iteration) + if calls is None: + continue + if survivor_index[iteration] != i: + if turn.token_usage is not None: + turn.token_usage.total_cost_usd = 0.0 # credited to the survivor instead + continue + turn.provider_call_costs = calls + total = sum(c.cost_usd for c in calls if c.cost_usd is not None) + if turn.token_usage is None: + turn.token_usage = TokenUsage(total_cost_usd=total) + else: + turn.token_usage.total_cost_usd = total + applied += 1 + + orphans = sorted(set(by_iteration) - matched_iterations) + if orphans: + logger.warning( + "LiteLLM actual-cost: %d cost-record iteration(s) %s matched no turn (run=%s task=%s); spend unbooked", + len(orphans), + orphans, + run_id, + task_id, + ) + return applied + + +def _survivor_index(turns: list[TurnRecord]) -> dict[str, int]: + """Map each iteration to the turn its calls are credited to. Multiple + ``TurnRecord``s can share an iteration — a crash+retry, or (multi-turn runs) a + trailing empty/model=None turn — so pick the LAST turn with that iteration THAT + HAS GENERATIONS; the calls belong to whichever turn actually generated, and + crediting an empty turn would strand them off their generations. Fall back to + the last turn if none have any.""" + survivor: dict[str, int] = {} + for i, turn in enumerate(turns): + key = str(turn.iteration) + has_generations = any(isinstance(m, AssistantMessage) for m in turn.messages) + if key not in survivor: + survivor[key] = i + else: + prev_has_generations = any(isinstance(m, AssistantMessage) for m in turns[survivor[key]].messages) + if has_generations or not prev_has_generations: + survivor[key] = i + return survivor diff --git a/src/coder_eval/models/__init__.py b/src/coder_eval/models/__init__.py index 084b0c2a..1d155604 100644 --- a/src/coder_eval/models/__init__.py +++ b/src/coder_eval/models/__init__.py @@ -177,6 +177,7 @@ CommandStatistics, CommandTelemetry, ContentBlock, + ProviderCallCost, ReconciliationMessage, SlowestCommandInfo, TokenUsage, @@ -267,6 +268,7 @@ "CommandTelemetry", "CommandStatistics", "ContentBlock", + "ProviderCallCost", "ReconciliationMessage", "SlowestCommandInfo", "TokenUsage", diff --git a/src/coder_eval/models/results.py b/src/coder_eval/models/results.py index feecdbe5..68d7a179 100644 --- a/src/coder_eval/models/results.py +++ b/src/coder_eval/models/results.py @@ -24,6 +24,7 @@ from coder_eval.models.telemetry import ( CommandStatistics, CommandTelemetry, + ProviderCallCost, TokenUsage, TranscriptMessage, ) @@ -350,6 +351,15 @@ class TurnRecord(BaseModel): default=None, description="SDK ResultMessage summary, when one was emitted (clean turns or partials that got one).", ) + provider_call_costs: list[ProviderCallCost] = Field( + default_factory=list, + description=( + "Per-call ACTUAL cost + cache captured proxy-side for the open-weight (LiteLLM) backend, " + "joined onto this turn by litellm_cost.apply_actual_cost. Empty on every other backend " + "(the SDK reports cost/cache natively there). When present, total_cost_usd is the sum of " + "these calls' cost_usd (the real bill), not the static rate-card estimate." + ), + ) crashed: bool = Field( default=False, description=( diff --git a/src/coder_eval/models/sandbox.py b/src/coder_eval/models/sandbox.py index 906add84..eb7b6dc1 100644 --- a/src/coder_eval/models/sandbox.py +++ b/src/coder_eval/models/sandbox.py @@ -236,6 +236,12 @@ class DockerDriverConfig(BaseModel): "LITELLM_AUTH_TOKEN", "LITELLM_MODEL", "LITELLM_SMALL_MODEL", + # Path to the proxy's per-call cost log for the actual-cost join. NOTE: + # forwarding the var is necessary but not sufficient under --driver docker + # — the log file itself must also be bind-mounted into the container for + # the join to see it (follow-up); without the mount, docker runs keep + # static pricing while local runs get real cost. + "LITELLM_COST_LOG", # Codex agent auth/routing — without these the in-container codex # binary falls back to a ChatGPT login that doesn't exist in the # container and auth fails. CODEX_API_KEY drives login_api_key; diff --git a/src/coder_eval/models/telemetry.py b/src/coder_eval/models/telemetry.py index c77cddec..775ae7ab 100644 --- a/src/coder_eval/models/telemetry.py +++ b/src/coder_eval/models/telemetry.py @@ -116,6 +116,27 @@ def __add__(self, other: TokenUsage) -> TokenUsage: ) +class ProviderCallCost(BaseModel): + """One upstream API call's ACTUAL cost + cache buckets, captured proxy-side. + + For the open-weight (LiteLLM) backend the Claude binary's Anthropic transport + drops OpenRouter's real ``usage.cost`` + per-call cache before Python can see + it, so a proxy-side callback (``litellm/cost_logger.py``) records each call and + the harness joins them onto the turn (``litellm_cost.apply_actual_cost``). Empty + on every other backend — the SDK reports cost/cache natively there. + """ + + call_id: str | None = Field(default=None, description="Upstream generation id (e.g. OpenRouter gen-...).") + provider: str | None = Field( + default=None, description="Upstream provider OpenRouter routed to (e.g. 'fireworks'), if reported." + ) + cost_usd: float | None = Field(default=None, description="Real per-call cost (OpenRouter usage.cost), if reported.") + input_tokens: int | None = Field(default=None, description="Prompt tokens for this call (incl. cached).") + cache_read_tokens: int | None = Field(default=None, description="Cached prompt tokens served on this call.") + cache_write_tokens: int | None = Field(default=None, description="Prompt tokens written to cache on this call.") + output_tokens: int | None = Field(default=None, description="Completion tokens for this call.") + + class ContentBlock(BaseModel): """One content block within a message, in emission order. diff --git a/src/coder_eval/orchestrator.py b/src/coder_eval/orchestrator.py index 796520c6..6d73840d 100644 --- a/src/coder_eval/orchestrator.py +++ b/src/coder_eval/orchestrator.py @@ -4,6 +4,7 @@ import logging import re import time +import uuid from collections.abc import Callable from contextlib import suppress from dataclasses import dataclass @@ -27,6 +28,7 @@ from .errors.executor import execute_with_retry from .errors.retry import create_error_context from .evaluation.checker import SuccessChecker, _short_failure_reason +from .litellm_cost import apply_actual_cost, load_cost_records from .models import ( ROUTE_NAMES, AgentKind, @@ -344,6 +346,12 @@ def __init__( """ self.task = task self.run_dir = run_dir + # Per-attempt nonce for the LiteLLM cost-log join. The proxy log is + # append-only and the run_id is a deterministic hash of run_dir, so a + # re-run into the same --run-dir would otherwise re-match (and double-count) + # a prior attempt's rows. A fresh nonce per Orchestrator (one per process + # invocation) scopes the join to THIS attempt's records. + self._cost_attempt_nonce = uuid.uuid4().hex self.preservation_mode = preservation_mode self.workspace_dir = workspace_dir self.task_file = task_file @@ -700,6 +708,11 @@ def _finalize_result(self, start_time: float) -> None: if not self.result.model_used and self.task.agent is not None and self.task.agent.model: self.result.model_used = self.task.agent.model + # Open-weight (LiteLLM) backend: replace per-turn cost with the ACTUAL + # per-call OpenRouter cost captured proxy-side. Runs BEFORE aggregation so + # the run total re-derives from the corrected per-turn costs. + self._join_litellm_actual_cost() + # Aggregate token usage self._aggregate_token_usage() @@ -892,6 +905,50 @@ def _check_expected_turns(self, *, iteration: int) -> None: ) self._expected_turns_warning_emitted = True + @property + def _cost_correlation_run_id(self) -> str: + """The LiteLLM cost-log correlation run id — a stable hash of the run dir. + + Single accessor used by BOTH the stamp site (``_create_agent``, into + ``x-ce-run-id``) and the join site (``_join_litellm_actual_cost``); keeping + the derivation in one place means the two can't drift and silently revert + every turn to static pricing. + """ + return hash_identifier(self.run_dir.as_posix()) + + def _join_litellm_actual_cost(self) -> None: + """Override per-turn cost with the proxy-captured ACTUAL per-call OpenRouter + cost (and attach the per-call cache breakdown) for the open-weight backend. + + No-op unless the agent ran on a ``LiteLLMRoute`` AND ``LITELLM_COST_LOG`` is + configured. Never fatal: a failure, or an empty/mismatched log, leaves each + turn's static rate-card estimate in place (the whole-turn fallback). + """ + if not (isinstance(self.route, LiteLLMRoute) and settings.litellm_cost_log and self.result is not None): + return + try: + applied = apply_actual_cost( + self.result, + run_id=self._cost_correlation_run_id, + task_id=self._log_task_id, + attempt=self._cost_attempt_nonce, + records=load_cost_records(settings.litellm_cost_log), + ) + if applied: + logger.info("LiteLLM actual-cost join: real per-call cost applied to %d turn(s)", applied) + else: + # Tags were stamped but nothing matched (file absent, proxy never + # wrote, wrong path, or a run/task/attempt mismatch). The run stays + # on the static rate card — warn so it isn't mistaken for the real bill. + logger.warning( + "LiteLLM actual-cost join found no matching records in %s (run=%s task=%s); cost stays static", + settings.litellm_cost_log, + self._cost_correlation_run_id, + self._log_task_id, + ) + except Exception: + logger.warning("LiteLLM actual-cost join failed; keeping static pricing", exc_info=True) + def _aggregate_token_usage(self) -> None: """Aggregate token usage from turns, storing on self.result. @@ -1206,7 +1263,7 @@ async def _create_agent(self) -> Agent[Any]: ValueError: If agent type is not supported TypeError: If config doesn't match agent's expected type """ - from coder_eval.agents import create_agent + from coder_eval.agents import AgentRegistry, create_agent from coder_eval.plugins import ensure_plugins_loaded # Safety net for the production agent-construction path: create_agent no @@ -1215,7 +1272,30 @@ async def _create_agent(self) -> Agent[Any]: ensure_plugins_loaded() assert self.task.agent is not None assert self.task.agent.type is not None - return create_agent(self.task.agent.type, self.task.agent, route=self.route) + # LiteLLM (open-weight) route only: give the agent correlation headers so a + # proxy-side cost-logging callback can attribute each call's real cost + + # cache buckets back to this task-run. x-ce-run-id is a stable per-task-run + # key (the join, in _finalize_result, recomputes it identically); x-ce-task-id + # is the human-readable canonical id. + # + # Gate on AGENT CAPABILITY, not the route: the route is settings-derived and + # independent of agent type, but only agents whose __init__ accepts the kwarg + # (supports_cost_log_tags) may receive it — otherwise the agent-agnostic + # factory would forward it into NoOp/Codex/Antigravity/plugin constructors + # that don't declare it and crash with TypeError under API_BACKEND=litellm. + kwargs: dict[str, Any] = {} + registration = AgentRegistry.get(self.task.agent.type) + if ( + isinstance(self.route, LiteLLMRoute) + and registration is not None + and registration.agent_class.supports_cost_log_tags + ): + kwargs["cost_log_tags"] = { + "x-ce-run-id": self._cost_correlation_run_id, + "x-ce-task-id": self._log_task_id, + "x-ce-attempt": self._cost_attempt_nonce, + } + return create_agent(self.task.agent.type, self.task.agent, route=self.route, **kwargs) async def _communicate_with_retry( self, diff --git a/tests/_fixtures/golden_streams/_scrub.py b/tests/_fixtures/golden_streams/_scrub.py index 03f7afde..fa368828 100644 --- a/tests/_fixtures/golden_streams/_scrub.py +++ b/tests/_fixtures/golden_streams/_scrub.py @@ -32,9 +32,17 @@ } ) +# Fields DROPPED (not masked) from the snapshot because they are NOT produced by +# the agent turn-loop this golden captures — they are populated later by the +# orchestrator (e.g. the LiteLLM actual-cost join sets ``provider_call_costs``). +# Always empty here, and agent-agnostic, so dropping keeps the golden stable +# across backends (Claude + Codex) without a per-field regen. +DROP_KEYS = frozenset({"provider_call_costs"}) + def scrub(obj: Any) -> Any: - """Recursively replace scrub-listed field values with a stable placeholder. + """Recursively replace scrub-listed field values with a stable placeholder, + and drop ``DROP_KEYS`` fields entirely. A ``None`` value is preserved (so the meaningful, deterministic present-vs-absent distinction survives — e.g. ``duration_ms=None`` on an @@ -45,6 +53,7 @@ def scrub(obj: Any) -> Any: return { key: (SCRUB_PLACEHOLDER if (key in SCRUB_KEYS and value is not None) else scrub(value)) for key, value in obj.items() + if key not in DROP_KEYS } if isinstance(obj, list): return [scrub(item) for item in obj] diff --git a/tests/test_event_collector.py b/tests/test_event_collector.py index 6231a97d..76e49e49 100644 --- a/tests/test_event_collector.py +++ b/tests/test_event_collector.py @@ -196,10 +196,13 @@ class TestFullFieldParity: """ # TurnRecord fields NOT carried verbatim from AgentEndEvent: - # commands -> reduced from the ToolEnd stream - # token_usage -> derived from end.usage.tokens - # timestamp -> record's own creation stamp, not an event field - _DERIVED: ClassVar[set[str]] = {"commands", "token_usage", "timestamp"} + # commands -> reduced from the ToolEnd stream + # token_usage -> derived from end.usage.tokens + # timestamp -> record's own creation stamp, not an event field + # provider_call_costs -> joined in post-run by the orchestrator from the + # LiteLLM proxy cost log (litellm_cost.apply_actual_cost), + # not emitted by the agent/EventCollector. + _DERIVED: ClassVar[set[str]] = {"commands", "token_usage", "timestamp", "provider_call_costs"} def _full_agent_end(self) -> AgentEndEvent: """An AgentEndEvent with every verbatim field set to a non-default sentinel.""" diff --git a/tests/test_litellm_config.py b/tests/test_litellm_config.py new file mode 100644 index 00000000..97b10d17 --- /dev/null +++ b/tests/test_litellm_config.py @@ -0,0 +1,57 @@ +"""Shape guards for the LiteLLM proxy config (litellm/litellm-config.yaml). + +A typo in this YAML (a mis-spelled key, a wrong callback path) fails silently at +proxy startup rather than in CI, so pin the structure the open-weight cost feature +relies on: `usage.include` + the vetted provider pins per model, and the callback +registration matching the real symbol in cost_logger.py. +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import yaml + + +_REPO_ROOT = Path(__file__).resolve().parent.parent +_CONFIG = _REPO_ROOT / "litellm" / "litellm-config.yaml" +_COST_LOGGER = _REPO_ROOT / "litellm" / "cost_logger.py" + +_CALLBACK = "cost_logger.proxy_handler_instance" + + +def _load() -> dict: + return yaml.safe_load(_CONFIG.read_text(encoding="utf-8")) + + +class TestLitellmConfigShape: + def test_cost_callback_registered(self): + assert _load()["litellm_settings"]["callbacks"] == _CALLBACK + + def test_openrouter_models_capture_usage_and_pin_providers(self): + # Derive the OpenRouter set from the config (not a hardcoded literal) so a + # future openrouter/* entry missing usage.include is caught automatically — + # otherwise it silently regresses to static pricing. + models = [m for m in _load()["model_list"] if str(m["litellm_params"]["model"]).startswith("openrouter/")] + assert models, "no openrouter/* models in litellm-config model_list" + for model in models: + extra = model["litellm_params"]["extra_body"] + # usage.include drives the actual-cost capture (cost_logger reads usage.cost). + assert extra["usage"]["include"] is True, model["model_name"] + provider = extra["provider"] + assert provider["sort"] == "price" + # allow_fallbacks: true, but bounded to the vetted `only` set (fall over + # WITHIN the set on saturation; never outside it). + assert provider["allow_fallbacks"] is True + assert isinstance(provider["only"], list) and provider["only"] + + def test_callback_symbol_exists_in_cost_logger(self): + # The callbacks string above names `cost_logger.proxy_handler_instance`; + # prove that symbol actually exists so the config and module can't drift. + module_name, attr = _CALLBACK.split(".") + spec = importlib.util.spec_from_file_location(module_name, _COST_LOGGER) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + assert hasattr(module, attr) diff --git a/tests/test_litellm_cost.py b/tests/test_litellm_cost.py new file mode 100644 index 00000000..d0847d9f --- /dev/null +++ b/tests/test_litellm_cost.py @@ -0,0 +1,329 @@ +"""Tests for the actual-cost join (litellm_cost.py) + the orchestrator hook. + +Covers loading the proxy's per-call JSONL and stitching real OpenRouter cost + +per-call cache onto a run's turns, incl. the retry no-double-count rule and the +whole-turn fallback to static pricing. +""" + +from __future__ import annotations + +from datetime import datetime +from types import SimpleNamespace + +import pytest +from pydantic import ValidationError + +import coder_eval.orchestrator as orch_mod +from coder_eval.litellm_cost import apply_actual_cost, load_cost_records +from coder_eval.models import ( + AgentKind, + AssistantMessage, + DirectRoute, + EvaluationResult, + LiteLLMRoute, + ReconciliationMessage, + TokenUsage, + TurnRecord, +) +from coder_eval.telemetry import hash_identifier + + +def _turn(iteration: int, static_cost: float | None = None) -> TurnRecord: + usage = TokenUsage(uncached_input_tokens=100, output_tokens=10, total_cost_usd=static_cost) + return TurnRecord(iteration=iteration, user_input="u", agent_output="a", duration_seconds=1.0, token_usage=usage) + + +def _result(turns: list[TurnRecord]) -> EvaluationResult: + return EvaluationResult( + task_id="t", + task_description="d", + variant_id="v", + agent_type=AgentKind.CLAUDE_CODE, + started_at=datetime(2026, 7, 29), + final_status="SUCCESS", + iteration_count=len(turns), + environment_info={}, + iterations=turns, + ) + + +def _rec(iteration: int, cost: float | None, *, run_id="R", task_id="T", cache_read=0, call_id="gen"): + return { + "run_id": run_id, + "task_id": task_id, + "iteration": str(iteration), + "call_id": call_id, + "cost": cost, + "input": 100, + "cache_read": cache_read, + "cache_write": 0, + "output": 10, + } + + +class TestLoadCostRecords: + def test_missing_file_returns_empty(self, tmp_path): + assert load_cost_records(tmp_path / "nope.jsonl") == [] + + def test_skips_blank_and_garbled_and_non_dict_lines(self, tmp_path): + p = tmp_path / "c.jsonl" + p.write_text('{"run_id":"R","cost":0.01}\n\n \n{bad json\n[1,2,3]\n{"run_id":"R2"}\n', encoding="utf-8") + recs = load_cost_records(p) + assert len(recs) == 2 + assert recs[0]["run_id"] == "R" and recs[1]["run_id"] == "R2" + + +class TestApplyActualCost: + def test_overrides_per_turn_cost_and_attaches_calls(self): + result = _result([_turn(0, static_cost=0.5), _turn(1, static_cost=0.5)]) + records = [ + _rec(0, 0.01, call_id="g0a", cache_read=64), + _rec(0, 0.02, call_id="g0b"), + _rec(1, 0.05, call_id="g1"), + ] + applied = apply_actual_cost(result, run_id="R", task_id="T", records=records) + assert applied == 2 + assert result.iterations[0].token_usage.total_cost_usd == 0.03 # 0.01 + 0.02, not the 0.5 estimate + assert result.iterations[1].token_usage.total_cost_usd == 0.05 + # Per-call breakdown attached: 2 calls on turn 0, 1 on turn 1. + assert len(result.iterations[0].provider_call_costs) == 2 + assert result.iterations[0].provider_call_costs[0].cache_read_tokens == 64 + assert result.iterations[1].provider_call_costs[0].cost_usd == 0.05 + + def test_no_matching_records_is_noop_and_keeps_static(self): + result = _result([_turn(0, static_cost=0.5)]) + # run_id mismatch → nothing applied, static estimate intact. + applied = apply_actual_cost(result, run_id="R", task_id="T", records=[_rec(0, 0.01, run_id="OTHER")]) + assert applied == 0 + assert result.iterations[0].token_usage.total_cost_usd == 0.5 + assert result.iterations[0].provider_call_costs == [] + + def test_unmatched_iteration_keeps_static_for_that_turn_only(self): + result = _result([_turn(0, static_cost=0.5), _turn(1, static_cost=0.7)]) + applied = apply_actual_cost(result, run_id="R", task_id="T", records=[_rec(0, 0.01)]) + assert applied == 1 + assert result.iterations[0].token_usage.total_cost_usd == 0.01 # joined + assert result.iterations[1].token_usage.total_cost_usd == 0.7 # static fallback + + def test_retry_credits_survivor_and_zeroes_sibling_no_double_count(self): + # Two turns share iteration 0 (a crashed attempt + its retry). The + # iteration's calls must be credited ONCE (to the last/survivor turn). + result = _result([_turn(0, static_cost=0.5), _turn(0, static_cost=0.5)]) + records = [_rec(0, 0.01, call_id="a"), _rec(0, 0.02, call_id="b")] + apply_actual_cost(result, run_id="R", task_id="T", records=records) + assert result.iterations[1].token_usage.total_cost_usd == 0.03 # survivor gets the sum + assert result.iterations[0].token_usage.total_cost_usd == 0.0 # sibling zeroed + assert result.iterations[0].provider_call_costs == [] + assert len(result.iterations[1].provider_call_costs) == 2 + # Run aggregate re-derives to exactly the real total — no double count. + run_total = sum(t.token_usage.total_cost_usd for t in result.iterations) + assert run_total == 0.03 + + def test_creates_token_usage_when_absent(self): + turn = TurnRecord(iteration=0, user_input="u", agent_output="a", duration_seconds=1.0, token_usage=None) + result = _result([turn]) + apply_actual_cost(result, run_id="R", task_id="T", records=[_rec(0, 0.04)]) + assert result.iterations[0].token_usage is not None + assert result.iterations[0].token_usage.total_cost_usd == 0.04 + + def test_unpriced_call_keeps_static_and_attaches_nothing(self): + # A record with no cost (cost=None) must NOT override the static estimate + # (that would bill it at $0 and understate the turn) and must NOT attach a + # misleading breakdown — the turn falls back to static, loudly (warned). + result = _result([_turn(0, static_cost=0.5)]) + applied = apply_actual_cost(result, run_id="R", task_id="T", records=[_rec(0, None)]) + assert applied == 0 + assert result.iterations[0].token_usage.total_cost_usd == 0.5 + assert result.iterations[0].provider_call_costs == [] + + def test_partial_coverage_keeps_static_for_that_turn(self): + # One priced call + one unpriced call on the same turn: overriding would bill + # the unpriced call at $0, so the whole turn keeps its static estimate. + result = _result([_turn(0, static_cost=0.42)]) + records = [_rec(0, 0.01, call_id="a"), _rec(0, None, call_id="b")] + applied = apply_actual_cost(result, run_id="R", task_id="T", records=records) + assert applied == 0 + assert result.iterations[0].token_usage.total_cost_usd == 0.42 + assert result.iterations[0].provider_call_costs == [] + + def test_attempt_nonce_prevents_rerun_double_count(self): + # The append-only log holds a PRIOR attempt's rows AND this attempt's rows + # under the same (run_id, task_id, iteration). The attempt nonce scopes the + # join to THIS attempt, so cost is not summed across both (reproduced the + # $0.06-for-$0.03 double-count the review flagged). + result = _result([_turn(0, static_cost=0.5)]) + records = [ + {**_rec(0, 0.03, call_id="old"), "attempt": "prev"}, + {**_rec(0, 0.03, call_id="new"), "attempt": "curr"}, + ] + applied = apply_actual_cost(result, run_id="R", task_id="T", attempt="curr", records=records) + assert applied == 1 + assert result.iterations[0].token_usage.total_cost_usd == 0.03 # this attempt only, not 0.06 + assert [c.call_id for c in result.iterations[0].provider_call_costs] == ["new"] + + def test_transactional_no_mutation_when_a_record_is_malformed(self): + # A well-formed JSON row with a wrong-typed cost raises while building the + # per-call breakdown. The whole join must abort with NO turn mutated (the + # caller keeps static pricing), not leave a half-joined run. + result = _result([_turn(0, static_cost=0.5), _turn(1, static_cost=0.7)]) + good = _rec(0, 0.03) + bad = {**_rec(1, 0.0), "cost": {"not": "a number"}} + with pytest.raises(ValidationError): + apply_actual_cost(result, run_id="R", task_id="T", records=[good, bad]) + # Byte-identical to the pre-join state: neither turn was overridden. + assert result.iterations[0].token_usage.total_cost_usd == 0.5 + assert result.iterations[1].token_usage.total_cost_usd == 0.7 + assert result.iterations[0].provider_call_costs == [] + + def test_credits_gen_bearing_turn_not_trailing_empty_turn(self): + # Two TurnRecords share iteration=1 (seen on multi-turn runs): the real + # agent turn WITH generations, and a trailing empty model=None turn. The + # calls must be credited to the turn that generated, not the empty one. + real = _turn_msgs( + 1, + [_asst("m1", output=10), _asst("m2", output=20), ReconciliationMessage()], + uncached=1500, + cache_read=0, + output=30, + ) + empty = TurnRecord(iteration=1, user_input="u", agent_output="", duration_seconds=1.0, token_usage=TokenUsage()) + result = _result([real, empty]) + records = [_call_rec(1, 0.01, 600, 0, "g1", out=10), _call_rec(1, 0.02, 900, 0, "g2", out=20)] + apply_actual_cost(result, run_id="R", task_id="T", records=records) + # Credited to the REAL (generation-bearing) turn, not the trailing empty one. + assert len(real.provider_call_costs) == 2 + assert real.token_usage.total_cost_usd == 0.03 + assert empty.provider_call_costs == [] # empty turn stranded nothing + assert empty.token_usage.total_cost_usd == 0.0 + + def test_degenerate_no_usage_call_is_ignored(self): + # A degenerate call with NO cost AND no tokens (a null record some providers + # emit) must not revert an otherwise-priced turn to static; it is dropped. + result = _result([_turn(0, static_cost=0.5)]) + good = _rec(0, 0.03, call_id="real") + degenerate = {**_rec(0, None, call_id="null"), "input": 0, "output": 0} + applied = apply_actual_cost(result, run_id="R", task_id="T", records=[good, degenerate]) + assert applied == 1 + assert result.iterations[0].token_usage.total_cost_usd == 0.03 # priced, degenerate ignored + assert [c.call_id for c in result.iterations[0].provider_call_costs] == ["real"] + + def test_retry_unpriced_survivor_does_not_zero_the_sibling(self): + # Regression: if the survivor falls back to static (an unpriced usage call), + # the crashed sibling must NOT be zeroed first — otherwise the iteration's + # spend is silently dropped. Both keep their static estimate. + result = _result([_turn(0, static_cost=0.5), _turn(0, static_cost=0.6)]) + records = [_rec(0, 0.01, call_id="a"), _rec(0, None, call_id="b")] # b has usage but no cost + applied = apply_actual_cost(result, run_id="R", task_id="T", records=records) + assert applied == 0 + assert result.iterations[0].token_usage.total_cost_usd == 0.5 # sibling NOT zeroed + assert result.iterations[1].token_usage.total_cost_usd == 0.6 # survivor stays static + assert result.iterations[1].provider_call_costs == [] + + def test_provider_is_attached_to_the_breakdown(self): + result = _result([_turn(0, static_cost=0.5)]) + rec = {**_rec(0, 0.02), "provider": "fireworks"} + apply_actual_cost(result, run_id="R", task_id="T", records=[rec]) + assert result.iterations[0].provider_call_costs[0].provider == "fireworks" + + +class TestOrchestratorJoinHook: + """The orchestrator method is called as an unbound function on a minimal + fake self, so the branch is covered without standing up a full Orchestrator.""" + + def test_noop_off_litellm_route(self): + fake = SimpleNamespace(route=DirectRoute(), result=_result([_turn(0, 0.5)])) + orch_mod.Orchestrator._join_litellm_actual_cost(fake) # must not touch anything + assert fake.result.iterations[0].token_usage.total_cost_usd == 0.5 + + def test_joins_on_litellm_route(self, tmp_path, monkeypatch): + run_dir = tmp_path / "run" + run_dir.mkdir() + log = tmp_path / "costs.jsonl" + run_id = hash_identifier(run_dir.as_posix()) + log.write_text( + f'{{"run_id":"{run_id}","task_id":"calc","iteration":"0","attempt":"att1","cost":0.09,"cache_read":5}}\n' + ) + monkeypatch.setattr(orch_mod.settings, "litellm_cost_log", str(log)) + fake = SimpleNamespace( + route=LiteLLMRoute(base_url="http://x:4000", auth_token="k", model="deepseek/deepseek-v4-pro"), + result=_result([_turn(0, static_cost=0.5)]), + _cost_correlation_run_id=run_id, + _cost_attempt_nonce="att1", + _log_task_id="calc", + ) + orch_mod.Orchestrator._join_litellm_actual_cost(fake) + assert fake.result.iterations[0].token_usage.total_cost_usd == 0.09 # static 0.5 overridden + assert fake.result.iterations[0].provider_call_costs[0].cache_read_tokens == 5 + + def test_join_never_raises_on_bad_log(self, tmp_path, monkeypatch): + monkeypatch.setattr(orch_mod.settings, "litellm_cost_log", str(tmp_path / "does-not-exist.jsonl")) + fake = SimpleNamespace( + route=LiteLLMRoute(base_url="http://x:4000", auth_token="k"), + result=_result([_turn(0, static_cost=0.5)]), + _cost_correlation_run_id="R", + _cost_attempt_nonce="att1", + _log_task_id="calc", + ) + orch_mod.Orchestrator._join_litellm_actual_cost(fake) # missing file → no-op, no raise + assert fake.result.iterations[0].token_usage.total_cost_usd == 0.5 + + def test_run_total_rederives_from_actual_after_join(self, tmp_path, monkeypatch): + # The join must run BEFORE aggregation so the run-level total sums the + # corrected per-turn costs, not the static estimate. Two turns, static 0.5 + # each (Σ 1.0); the log books actuals 0.03 + 0.05 (Σ 0.08). If the ordering + # ever reversed, aggregation would sum the stale static costs and this fails. + run_dir = tmp_path / "run" + run_dir.mkdir() + log = tmp_path / "costs.jsonl" + run_id = hash_identifier(run_dir.as_posix()) + log.write_text( + f'{{"run_id":"{run_id}","task_id":"calc","iteration":"0","attempt":"att1","cost":0.03}}\n' + f'{{"run_id":"{run_id}","task_id":"calc","iteration":"1","attempt":"att1","cost":0.05}}\n' + ) + monkeypatch.setattr(orch_mod.settings, "litellm_cost_log", str(log)) + fake = SimpleNamespace( + route=LiteLLMRoute(base_url="http://x:4000", auth_token="k", model="deepseek/deepseek-v4-pro"), + result=_result([_turn(0, static_cost=0.5), _turn(1, static_cost=0.5)]), + _cost_correlation_run_id=run_id, + _cost_attempt_nonce="att1", + _log_task_id="calc", + ) + orch_mod.Orchestrator._join_litellm_actual_cost(fake) + orch_mod.Orchestrator._aggregate_token_usage(fake) + assert fake.result.total_token_usage.total_cost_usd == pytest.approx(0.08) # Σ actual, not Σ static + + +def _asst(message_id, output=5): + return AssistantMessage( + started_at=datetime(2026, 7, 29), + completed_at=datetime(2026, 7, 29), + generation_duration_ms=1.0, + message_id=message_id, + output_tokens=output, + ) + + +def _turn_msgs(iteration, messages, *, uncached, cache_read, output, static_cost=0.5): + tu = TokenUsage( + uncached_input_tokens=uncached, + cache_read_input_tokens=cache_read, + output_tokens=output, + total_cost_usd=static_cost, + ) + return TurnRecord( + iteration=iteration, user_input="u", agent_output="a", duration_seconds=1.0, token_usage=tu, messages=messages + ) + + +def _call_rec(iteration, cost, inp, cache_read, call_id, out=5): + return { + "run_id": "R", + "task_id": "T", + "iteration": str(iteration), + "call_id": call_id, + "cost": cost, + "input": inp, + "cache_read": cache_read, + "cache_write": 0, + "output": out, + } diff --git a/tests/test_litellm_cost_logger.py b/tests/test_litellm_cost_logger.py new file mode 100644 index 00000000..97b8ce87 --- /dev/null +++ b/tests/test_litellm_cost_logger.py @@ -0,0 +1,262 @@ +"""Tests for the proxy-side per-call cost/cache callback (litellm/cost_logger.py). + +The module lives outside the package (the proxy may run in its own env), so it's +loaded by file path. Its litellm import is guarded, so these run without litellm. +""" + +from __future__ import annotations + +import importlib.util +import inspect +import json +from datetime import datetime +from pathlib import Path + +import pytest + + +_PATH = Path(__file__).resolve().parent.parent / "litellm" / "cost_logger.py" + + +@pytest.fixture(scope="module") +def cl(): + spec = importlib.util.spec_from_file_location("cost_logger", _PATH) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +# Real OpenRouter warm-call usage observed in the 2026-07-29 probe (DeepSeek V4 Pro). +WARM_USAGE = { + "prompt_tokens": 4811, + "completion_tokens": 8, + "total_tokens": 4819, + "cost": 0.0006692671, + "prompt_tokens_details": {"cached_tokens": 4096, "cache_write_tokens": 0}, +} +TAGS = {"x-ce-run-id": "abc123", "x-ce-task-id": "calc/v1", "x-ce-iteration": "2", "x-ce-attempt": "att9"} + + +class TestBuildCostRecord: + def test_reads_real_usage_cost_and_cache(self, cl): + rec = cl.build_cost_record( + WARM_USAGE, TAGS, model="deepseek/deepseek-v4-pro", provider="fireworks", call_id="gen-1" + ) + assert rec is not None + assert rec["cost"] == 0.0006692671 # OpenRouter usage.cost, the REAL price + assert rec["provider"] == "fireworks" # routed upstream, for the per-call table + assert rec["input"] == 4811 + assert rec["cache_read"] == 4096 + assert rec["cache_write"] == 0 + assert rec["output"] == 8 + assert rec["run_id"] == "abc123" + assert rec["task_id"] == "calc/v1" + assert rec["iteration"] == "2" + assert rec["attempt"] == "att9" # per-attempt nonce for rerun de-dup + assert rec["model"] == "deepseek/deepseek-v4-pro" + + def test_reads_anthropic_shaped_usage(self, cl): + # On the Anthropic-inbound path a call's usage can come back Anthropic-shaped + # (input_tokens = UNCACHED slice, cache_read_input_tokens separate). The record's + # `input` must normalize to the FULL prompt so uncached = input - cache_read holds. + anthropic_usage = { + "input_tokens": 715, # uncached slice + "output_tokens": 8, + "cache_read_input_tokens": 4096, + "cache_creation_input_tokens": 0, + "cost": 0.0006692671, + } + rec = cl.build_cost_record(anthropic_usage, TAGS, model="deepseek/deepseek-v4-pro", call_id="g") + assert rec["input"] == 715 + 4096 + 0 # normalized to total prompt (4811) + assert rec["cache_read"] == 4096 + assert rec["cache_write"] == 0 + assert rec["output"] == 8 + assert rec["cost"] == 0.0006692671 + + def test_none_when_no_tag_and_no_cost(self, cl): + # Nothing joinable and no cost signal → skip (harness falls back to static). + assert cl.build_cost_record({}, {}, model=None, call_id=None) is None + assert cl.build_cost_record({"prompt_tokens": 10}, {}, model="m", call_id=None) is None + + def test_cost_alone_is_enough(self, cl): + # A cost with no correlation tag is still worth recording. + rec = cl.build_cost_record({"cost": 0.01}, {}, model="m", call_id=None) + assert rec is not None and rec["cost"] == 0.01 and rec["run_id"] is None + + def test_tag_alone_is_enough(self, cl): + # A correlation tag with no cost yet (e.g. cost absent) is still recorded. + rec = cl.build_cost_record({}, {"x-ce-run-id": "r"}, model="m", call_id=None) + assert rec is not None and rec["run_id"] == "r" and rec["cost"] is None + + def test_rejects_bool_and_nonnumeric_costs(self, cl): + rec = cl.build_cost_record( + {"cost": True, "prompt_tokens": "oops"}, {"x-ce-run-id": "r"}, model="m", call_id=None + ) + assert rec["cost"] is None # True is not a real number + assert rec["input"] is None + + def test_rejects_non_finite_cost(self, cl): + # A NaN/Inf cost would serialize as a bare NaN token (invalid JSON for the + # whole task.json) and make the max_usd gate's cost>limit silently never fire. + assert ( + cl.build_cost_record({"cost": float("nan")}, {"x-ce-run-id": "r"}, model="m", call_id=None)["cost"] is None + ) + assert ( + cl.build_cost_record({"cost": float("inf")}, {"x-ce-run-id": "r"}, model="m", call_id=None)["cost"] is None + ) + + +class TestExtractTags: + def test_pulls_and_lowercases_only_ce_tags(self, cl): + kwargs = { + "litellm_params": { + "metadata": { + "headers": { + "X-CE-Run-Id": "abc", + "x-ce-iteration": "3", + "authorization": "Bearer secret", + "content-type": "application/json", + } + } + } + } + assert cl.extract_tags(kwargs) == {"x-ce-run-id": "abc", "x-ce-iteration": "3"} + + def test_merges_alternate_header_locations(self, cl): + kwargs = {"proxy_server_request": {"headers": {"x-ce-task-id": "t1"}}} + assert cl.extract_tags(kwargs) == {"x-ce-task-id": "t1"} + + def test_empty_when_no_headers(self, cl): + assert cl.extract_tags({}) == {} + + +class TestAppendAndEmit: + def test_append_writes_jsonl(self, cl, tmp_path, monkeypatch): + path = tmp_path / "costs.jsonl" + monkeypatch.setenv("LITELLM_COST_LOG", str(path)) + cl.append_record({"run_id": "a", "cost": 0.01}) + cl.append_record({"run_id": "b", "cost": 0.02}) + cl.append_record(None) # no-op + lines = path.read_text().splitlines() + assert len(lines) == 2 + assert json.loads(lines[0])["run_id"] == "a" + assert json.loads(lines[1])["cost"] == 0.02 + + def test_append_noop_without_env(self, cl, monkeypatch): + monkeypatch.delenv("LITELLM_COST_LOG", raising=False) + cl.append_record({"run_id": "a"}) # must not raise + + def test_emit_end_to_end(self, cl, tmp_path, monkeypatch): + path = tmp_path / "costs.jsonl" + monkeypatch.setenv("LITELLM_COST_LOG", str(path)) + kwargs = {"litellm_params": {"metadata": {"headers": dict(TAGS)}}, "model": "deepseek/deepseek-v4-pro"} + response_obj = {"id": "gen-9", "provider": "Baidu", "model": "deepseek/deepseek-v4-pro", "usage": WARM_USAGE} + cl.proxy_handler_instance._emit(kwargs, response_obj) + rec = json.loads(path.read_text().splitlines()[0]) + assert rec["run_id"] == "abc123" and rec["cost"] == 0.0006692671 and rec["cache_read"] == 4096 + + def test_emit_never_raises_on_garbage(self, cl, monkeypatch): + monkeypatch.delenv("LITELLM_COST_LOG", raising=False) + cl.proxy_handler_instance._emit({}, None) + cl.proxy_handler_instance._emit({"litellm_params": None}, object()) + + +class TestHookInvocation: + """LiteLLM only ever calls log_success_event / async_log_success_event (the + config registers proxy_handler_instance); every other test calls _emit directly, + so cover the real entry points and the never-break-the-proxy guard here.""" + + def _kwargs(self): + return {"litellm_params": {"metadata": {"headers": dict(TAGS)}}, "model": "m"} + + def test_sync_hook_writes_a_record(self, cl, tmp_path, monkeypatch): + path = tmp_path / "costs.jsonl" + monkeypatch.setenv("LITELLM_COST_LOG", str(path)) + cl.proxy_handler_instance.log_success_event(self._kwargs(), {"id": "g", "usage": WARM_USAGE}, 0.0, 1.0) + assert json.loads(path.read_text().splitlines()[0])["cost"] == 0.0006692671 + + async def test_async_hook_writes_a_record(self, cl, tmp_path, monkeypatch): + path = tmp_path / "costs.jsonl" + monkeypatch.setenv("LITELLM_COST_LOG", str(path)) + await cl.proxy_handler_instance.async_log_success_event( + self._kwargs(), {"id": "g", "usage": WARM_USAGE}, 0.0, 1.0 + ) + assert json.loads(path.read_text().splitlines()[0])["cost"] == 0.0006692671 + + def test_a_failing_writer_never_propagates(self, cl, tmp_path, monkeypatch): + # The never-break-the-proxy guard (the bare except in _emit): even if + # append_record raises (read-only path, ENOSPC, …) the hook must not propagate. + monkeypatch.setenv("LITELLM_COST_LOG", str(tmp_path / "c.jsonl")) + + def _boom(*_a, **_k): + raise OSError("disk full") + + monkeypatch.setattr(cl, "append_record", _boom) + cl.proxy_handler_instance.log_success_event(self._kwargs(), {"id": "g", "usage": WARM_USAGE}, 0.0, 1.0) + + def test_conforms_to_real_litellm_customlogger_signature(self): + # Under test the repo's own litellm/ dir shadows the pip package, so + # cost_logger falls back to a stub CustomLogger. When the real package IS + # installed, assert the hook names/signatures the proxy will call exist on the + # ABC — so an upstream rename fails loudly instead of silently capturing nothing. + mod = pytest.importorskip("litellm.integrations.custom_logger") + for name in ("log_success_event", "async_log_success_event"): + assert hasattr(mod.CustomLogger, name) + params = list(inspect.signature(getattr(mod.CustomLogger, name)).parameters) + assert params[:5] == ["self", "kwargs", "response_obj", "start_time", "end_time"] + + +class TestWireContractRoundTrip: + """Producer (build_cost_record) → JSONL → consumer (load_cost_records → + apply_actual_cost). The record keys are hand-declared on both sides, so a rename + would silently make cost fail to land; this crosses the boundary to pin them.""" + + def test_cost_survives_the_json_round_trip(self, cl, tmp_path): + from coder_eval.litellm_cost import apply_actual_cost, load_cost_records + from coder_eval.models import AgentKind, EvaluationResult, TokenUsage, TurnRecord + + tags = {"x-ce-run-id": "R", "x-ce-task-id": "T", "x-ce-iteration": "0", "x-ce-attempt": "att1"} + rec = cl.build_cost_record(WARM_USAGE, tags, model="m", call_id="gen-1") + path = tmp_path / "c.jsonl" + path.write_text(json.dumps(rec) + "\n", encoding="utf-8") + + turn = TurnRecord( + iteration=0, + user_input="u", + agent_output="a", + duration_seconds=1.0, + token_usage=TokenUsage(total_cost_usd=0.99), + ) + result = EvaluationResult( + task_id="T", + task_description="d", + variant_id="v", + agent_type=AgentKind.CLAUDE_CODE, + started_at=datetime(2026, 7, 30), + final_status="SUCCESS", + iteration_count=1, + environment_info={}, + iterations=[turn], + ) + applied = apply_actual_cost(result, run_id="R", task_id="T", attempt="att1", records=load_cost_records(path)) + assert applied == 1 + assert turn.token_usage.total_cost_usd == WARM_USAGE["cost"] # the real cost round-tripped and landed + + +class TestToDict: + """`_to_dict` coerces a dict / pydantic model / litellm object to a plain dict.""" + + def test_uses_model_dump_and_survives_a_raising_dumper(self, cl): + class Ok: + def model_dump(self): + return {"k": "v"} + + class Boom: + def model_dump(self): + raise RuntimeError("nope") + + assert cl._to_dict(Ok()) == {"k": "v"} # happy path + # A raising dumper falls through to {} — never propagates (the callback must + # not break the proxy). This is the branch the bare except guards. + assert cl._to_dict(Boom()) == {} diff --git a/tests/test_litellm_route.py b/tests/test_litellm_route.py index 8714cebb..3d7a7a8c 100644 --- a/tests/test_litellm_route.py +++ b/tests/test_litellm_route.py @@ -295,6 +295,62 @@ def test_custom_route_neutralizes_inherited_anthropic_api_key(self, monkeypatch) env, _ = ClaudeCodeAgent._build_sdk_env(LiteLLMRoute(base_url="http://x:4000", auth_token="sk-1")) assert env["ANTHROPIC_API_KEY"] == "" + def test_cost_log_tags_become_custom_headers(self): + """cost_log_tags → ANTHROPIC_CUSTOM_HEADERS as newline-separated + `Name: Value` pairs (the format Claude Code forwards verbatim), so the + proxy-side cost log can join each call back to the run/task/turn.""" + route = LiteLLMRoute(base_url="http://x:4000", auth_token="sk-1", model="deepseek/deepseek-v4-pro") + tags = {"x-ce-run-id": "abc123", "x-ce-task-id": "calc/v1", "x-ce-iteration": "2"} + env, _ = ClaudeCodeAgent._build_sdk_env(route, cost_log_tags=tags) + assert env["ANTHROPIC_CUSTOM_HEADERS"] == "x-ce-run-id: abc123\nx-ce-task-id: calc/v1\nx-ce-iteration: 2" + + def test_no_cost_log_tags_omits_custom_headers(self): + route = LiteLLMRoute(base_url="http://x:4000", auth_token="sk-1") + env, _ = ClaudeCodeAgent._build_sdk_env(route) + assert "ANTHROPIC_CUSTOM_HEADERS" not in env + env2, _ = ClaudeCodeAgent._build_sdk_env(route, cost_log_tags={}) + assert "ANTHROPIC_CUSTOM_HEADERS" not in env2 # empty dict is a no-op + + def test_cost_log_tags_ignored_on_non_litellm_routes(self): + """The tag is a LiteLLM-only concern; Bedrock/Direct must not emit it.""" + tags = {"x-ce-run-id": "abc123"} + bedrock = BedrockRoute(bearer_token="t", region="eu-north-1", model="x") + env_b, _ = ClaudeCodeAgent._build_sdk_env(bedrock, cost_log_tags=tags) + assert "ANTHROPIC_CUSTOM_HEADERS" not in env_b + env_d, _ = ClaudeCodeAgent._build_sdk_env(DirectRoute(), cost_log_tags=tags) + assert "ANTHROPIC_CUSTOM_HEADERS" not in env_d + + def test_cost_log_tags_gated_on_agent_capability_not_route(self): + """Regression: cost_log_tags is a Claude-only constructor kwarg, but the + route that triggers it (LiteLLM) is agent-independent. The agent-agnostic + create_agent factory must forward it ONLY to agents that declare + supports_cost_log_tags — otherwise a none/codex/antigravity task crashes + with TypeError under API_BACKEND=litellm.""" + from coder_eval.agents import AgentRegistry, create_agent + from coder_eval.models import NoneAgentConfig + from coder_eval.plugins import ensure_plugins_loaded + + ensure_plugins_loaded() + # Capability contract the orchestrator gate reads. + assert AgentRegistry.get(AgentKind.CLAUDE_CODE).agent_class.supports_cost_log_tags is True + assert AgentRegistry.get(AgentKind.NONE).agent_class.supports_cost_log_tags is False + + route = LiteLLMRoute(base_url="http://x:4000", auth_token="sk-1", model="deepseek/deepseek-v4-pro") + # A none-agent constructs fine on a LiteLLM route (the gate omits the kwarg)... + assert create_agent(AgentKind.NONE, NoneAgentConfig(type=AgentKind.NONE), route=route) is not None + # ...and it WOULD crash if the kwarg were forwarded — exactly what the gate prevents. + with pytest.raises(TypeError): + create_agent( + AgentKind.NONE, NoneAgentConfig(type=AgentKind.NONE), route=route, cost_log_tags={"x-ce-run-id": "r"} + ) + + def test_cost_log_tags_reject_header_injection(self): + # A task_id/variant_id carrying a CR/LF would inject extra headers into every + # SDK->proxy request; the seam must reject it (single-line ASCII only). + route = LiteLLMRoute(base_url="http://x:4000", auth_token="sk-1") + with pytest.raises(ValueError, match="single-line ASCII"): + ClaudeCodeAgent._build_sdk_env(route, cost_log_tags={"x-ce-task-id": "ok\nAuthorization: Bearer forged"}) + class TestResolveEffectiveModelCustom: """_resolve_effective_model() on the LiteLLM route — no prefixing."""