From a30533c7e78cd6e3d16addc07feb9089c940701b Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Tue, 10 Nov 2026 22:41:38 +0530 Subject: [PATCH 1/5] fix(harness): record each Claude turn once, never the CLI's own estimate, and date the price table The CLI reports running totals within one connection, and the kit added each one, so a stage that sent a second message counted its earlier turns again. The simulated customer, one message per line, was billed (N+1)/2 times its true tokens. Each result now records its own counts: the difference from the connection's last totals, or, on a resumed connection's first result, the per-query usage, since SDK 0.2.176 on carries the old totals over a resume. A cost the kit can't price is now unpriced, never the CLI's estimate (about 7x Gemini's rate). The price table becomes dated rows, re-checked by 2027-03-31; the three Flash models double on 2027-01-01, per Google's Vertex pricing page. CI now runs these tests, so the check-by date turns it red from 2027-03-01. TH-9057 TH-9039 Signed-off-by: khushalsonawat --- .github/workflows/sdk-smoke.yml | 2 + src/fi/alk/harness/backends/claude.py | 74 ++- src/fi/alk/harness/backends/vertex_gemini.py | 101 ++- tests/harness/test_claude_cost.py | 632 +++++++++++++++++++ tests/harness/test_spend.py | 161 ++++- tests/harness/test_vertex_gemini_backend.py | 12 +- 6 files changed, 921 insertions(+), 61 deletions(-) create mode 100644 tests/harness/test_claude_cost.py diff --git a/.github/workflows/sdk-smoke.yml b/.github/workflows/sdk-smoke.yml index 2853b6220..10809e1d4 100644 --- a/.github/workflows/sdk-smoke.yml +++ b/.github/workflows/sdk-smoke.yml @@ -62,6 +62,8 @@ jobs: tests/harness/test_process_runtime.py tests/harness/test_simulation_mark_bootstrap.py tests/harness/test_process_preflight.py + tests/harness/test_spend.py + tests/harness/test_claude_cost.py tests/providers tests/harness/test_scenario_source_provision_agent.py tests/harness/test_hosted_entrypoint_provision_retry.py diff --git a/src/fi/alk/harness/backends/claude.py b/src/fi/alk/harness/backends/claude.py index 0759766f5..45d851166 100644 --- a/src/fi/alk/harness/backends/claude.py +++ b/src/fi/alk/harness/backends/claude.py @@ -12,6 +12,7 @@ from __future__ import annotations import dataclasses +import logging import os import tempfile from typing import Any, AsyncIterator @@ -49,6 +50,8 @@ tool_input_schema, ) +logger = logging.getLogger(__name__) + # The loop is Claude Code; the model it bills is not. DEFAULT_MODEL = "gemini-3.7-flash" @@ -184,8 +187,15 @@ def __init__( self._mirror_errors: list[str] = [] self._tool_names: dict[str, str] = {} self._terminal_save_succeeded = False + # The CLI reports running totals within one connection, so a result's own counts are its + # totals less the last ones seen. None until this connection's first report. + self._so_far: dict[str, int] | None = None + # A resumed connection's totals may carry the old process's turns on (SDK 0.2.176 on). + self._resumed = bool(getattr(options, "resume", None)) async def start(self) -> None: + # A new process starts its own totals, which the old baseline must not be taken from. + self._so_far = None self._client = ClaudeSDKClient(options=self._options) await self._client.connect() @@ -277,7 +287,8 @@ def _translate(self, received: Any) -> list[Any]: if isinstance(received, ResultMessage): # subtype alone is not the outcome. A call that failed upstream still arrives with # subtype "success", so the error facts ride along and Stage decides what failed. - counted = _tokens(getattr(received, "model_usage", None)) + model_usage = getattr(received, "model_usage", None) + own = self._own(received) errors = [ *list(getattr(received, "errors", None) or []), *self._mirror_errors, @@ -286,17 +297,23 @@ def _translate(self, received: Any) -> list[Any]: StageDone( outcome=received.subtype, turns=received.num_turns, - cost_usd=_cost( - getattr(received, "model_usage", None), - counted, - received.total_cost_usd, + # A turn that reported no usage cost nothing; None would read as unpriced. + cost_usd=( + 0.0 + if own is None + else _cost( + [self._reported_model] + if self._reported_model + else [str(name) for name in (model_usage or {})], + own, + ) ), - **counted, + **(own or _tokens(None)), session_id=received.session_id, models=( {self._reported_model} if self._reported_model - else set(getattr(received, "model_usage", None) or {}) + else set(model_usage or {}) ), is_error=bool( getattr(received, "is_error", False) or self._mirror_errors @@ -326,19 +343,50 @@ def _translate(self, received: Any) -> list[Any]: return returned return [] + def _own(self, received: Any) -> dict[str, int] | None: + """This result's own counts, or None when it reported no usage at all.""" + totals = _tokens(getattr(received, "model_usage", None)) + if not any(totals.values()): + # Nothing to take a difference from, so the baseline stands for the next result. + return None + if self._so_far is None and self._resumed: + # A resumed connection's first totals may hold the earlier process's turns, and + # `usage` reports this query alone on every SDK. It leaves out workers' tokens. + usage = getattr(received, "usage", None) + if not usage: + logger.warning( + "a resumed Claude session's first result reported no usage; " + "its own tokens are recorded as 0" + ) + own = _tokens({"usage": usage or {}}) + elif self._so_far is None: + own = totals + elif any(totals[count] < self._so_far[count] for count in totals) or ( + totals["tokens_in"] - totals["tokens_cached"] + < self._so_far["tokens_in"] - self._so_far["tokens_cached"] + ): + # The totals restarted under us, so this result is all new. + own = totals + else: + own = {count: totals[count] - self._so_far[count] for count in totals} + self._so_far = totals + return own -def _cost(model_usage: Any, counted: dict[str, int], reported: float | None) -> float | None: - """What the run cost, priced here where possible, else the CLI's own figure.""" + +def _cost(names: list[str], own: dict[str, int]) -> float | None: + """What this turn cost, priced here from its own tokens, or None where nothing is priced. + + Never the CLI's own estimate: it prices a non-Claude model at a guess, about 7x Gemini's rate. + """ from .vertex_gemini import priced - named = [str(name) for name in (model_usage or {})] ours = [ - priced(name, counted["tokens_in"], counted["tokens_out"], counted["tokens_cached"]) - for name in named + priced(name, own["tokens_in"], own["tokens_out"], own["tokens_cached"]) + for name in names ] known = [one for one in ours if one is not None] if not known: - return reported + return None # One stage runs on one model: summing would price the same tokens once per reported id. return max(known) diff --git a/src/fi/alk/harness/backends/vertex_gemini.py b/src/fi/alk/harness/backends/vertex_gemini.py index 44b92c54b..b8a4787eb 100644 --- a/src/fi/alk/harness/backends/vertex_gemini.py +++ b/src/fi/alk/harness/backends/vertex_gemini.py @@ -48,31 +48,46 @@ DEFAULT_MODEL = "gemini-3.7-flash" -# Vertex list pricing per 1M tokens: (input, output, the day this pair was last checked against -# the platform's litellm model table). An unknown or stale model reports no cost rather than a -# wrong one, and shows up in `unpriced_turns`. +# Vertex list pricing per 1M tokens, as rows of (input, output, first_day, last_day) in date order. +# Today's rates are the platform gateway's litellm model table (agentcc-gateway/internal/modeldb/ +# litellm.json, its vertex_ai/ entries, future-agi 9ec5f9fb1, 2026-11-02). The 2027 Flash rows are +# Google's Vertex pricing page (cloud.google.com/model-garden/pricing, Global tier, read +# 2026-11-10), which that table does not carry yet. A row ends either on a published price change +# (the 2026 Flash rows) or on CHECK_BY, the day every latest row must be re-checked by. A model +# with no row for today reports no cost rather than a wrong one, and shows up in `unpriced_turns`. +CHECK_BY = "2027-03-31" + # What Vertex charges for a cache read, as a share of the input rate. CACHE_READ_SHARE = 0.10 PRICES_PER_MILLION = { - "gemini-3.8-flash": (0.75, 3.75, "2026-12-31"), - "gemini-3.7-flash": (0.75, 3.75, "2026-12-31"), - "gemini-3.6-flash": (0.75, 3.75, "2026-12-31"), - "gemini-3.5-transcribe-preview": (2.5, 12, "2026-12-31"), - "gemini-3.5-transcribe-live-preview": (3.5, 21, "2026-12-31"), - "gemini-3.5-flash-lite": (0.3, 2.5, "2026-12-31"), - "gemini-3.5-flash": (1.5, 9, "2026-12-31"), - "gemini-3.1-pro-preview-customtools": (2, 12, "2026-12-31"), - "gemini-3.1-pro-preview": (2, 12, "2026-12-31"), - "gemini-3.1-flash-lite-preview": (0.25, 1.5, "2026-12-31"), - "gemini-3.1-flash-lite-image": (0.25, 1.5, "2026-12-31"), - "gemini-3.1-flash-lite": (0.25, 1.5, "2026-12-31"), - "gemini-3.1-flash-image-preview": (0.5, 3, "2026-12-31"), - "gemini-3.1-flash-image": (0.5, 3, "2026-12-31"), - "gemini-3-pro-preview": (2, 12, "2026-12-31"), - "gemini-3-pro-image-preview": (2, 12, "2026-12-31"), - "gemini-3-pro-image": (2, 12, "2026-12-31"), - "gemini-3-flash-preview": (0.5, 3, "2026-12-31"), + "gemini-3.8-flash": ( + (0.75, 3.75, "2026-11-02", "2026-12-31"), + (1.50, 7.50, "2027-01-01", CHECK_BY), + ), + "gemini-3.7-flash": ( + (0.75, 3.75, "2026-11-02", "2026-12-31"), + (1.50, 7.50, "2027-01-01", CHECK_BY), + ), + "gemini-3.6-flash": ( + (0.75, 3.75, "2026-11-02", "2026-12-31"), + (1.50, 7.50, "2027-01-01", CHECK_BY), + ), + "gemini-3.5-transcribe-preview": ((2.0, 12, "2026-11-02", CHECK_BY),), + "gemini-3.5-transcribe-live-preview": ((3.5, 21, "2026-11-02", CHECK_BY),), + "gemini-3.5-flash-lite": ((0.3, 2.5, "2026-11-02", CHECK_BY),), + "gemini-3.5-flash": ((1.5, 9, "2026-11-02", CHECK_BY),), + "gemini-3.1-pro-preview-customtools": ((2, 12, "2026-11-02", CHECK_BY),), + "gemini-3.1-pro-preview": ((2, 12, "2026-11-02", CHECK_BY),), + "gemini-3.1-flash-lite-preview": ((0.25, 1.5, "2026-11-02", CHECK_BY),), + "gemini-3.1-flash-lite-image": ((0.25, 1.5, "2026-11-02", CHECK_BY),), + "gemini-3.1-flash-lite": ((0.25, 1.5, "2026-11-02", CHECK_BY),), + "gemini-3.1-flash-image-preview": ((0.5, 3, "2026-11-02", CHECK_BY),), + "gemini-3.1-flash-image": ((0.5, 3, "2026-11-02", CHECK_BY),), + "gemini-3-pro-preview": ((2, 12, "2026-11-02", CHECK_BY),), + "gemini-3-pro-image-preview": ((2, 12, "2026-11-02", CHECK_BY),), + "gemini-3-pro-image": ((2, 12, "2026-11-02", CHECK_BY),), + "gemini-3-flash-preview": ((0.5, 3, "2026-11-02", CHECK_BY),), } def _json_schema(schema: Any) -> dict[str, Any]: @@ -680,26 +695,50 @@ def _cost( logger = logging.getLogger(__name__) +def _rows(name: str) -> tuple[tuple[Any, ...], ...] | None: + rows = PRICES_PER_MILLION.get(name) + if rows is None and "/" in name: + # A gateway prefixes the route, e.g. "vertex_ai/"; the price belongs to the model. + rows = PRICES_PER_MILLION.get(name.rsplit("/", 1)[-1]) + return rows + + +def current_rates(name: str, today: date | None = None) -> tuple[float, float] | None: + """The (input, output) rates per 1M tokens that hold on `today`, or None if no row does. + + Read the table through this rather than by position: a model's price is a schedule of rows. + """ + day = (today or date.today()).isoformat() + # Every row is checked whatever its place, so the table's order never picks the price; the + # check-by test keeps a model's rows from overlapping, so at most one row holds. + for rate_in, rate_out, first_day, last_day in _rows(name) or (): + if first_day <= day <= last_day: + return rate_in, rate_out + return None + + def priced( model: str, tokens_in: int, tokens_out: int, tokens_cached: int = 0 ) -> float | None: """What these tokens cost, or None where no price can be stood behind.""" - prices = PRICES_PER_MILLION.get(model) - if prices is None and "/" in model: - # A gateway prefixes the route, e.g. "vertex_ai/"; the price belongs to the model. - prices = PRICES_PER_MILLION.get(model.rsplit("/", 1)[-1]) - if prices is None: + rows = _rows(model) + if rows is None: return None - if len(prices) > 2 and date.today().isoformat() > str(prices[2]): + today = date.today() + rates = current_rates(model, today) + if rates is None: logger.warning( - "no current price for %s: the table's figures expired on %s", + "no current price for %s on %s: the table's rows run from %s to %s", model, - prices[2], + today.isoformat(), + min(row[2] for row in rows), + max(row[3] for row in rows), ) return None + rate_in, rate_out = rates cached = min(max(tokens_cached, 0), max(tokens_in, 0)) - billed_in = (tokens_in - cached) * prices[0] + cached * prices[0] * CACHE_READ_SHARE - return (billed_in + tokens_out * prices[1]) / 1_000_000 + billed_in = (tokens_in - cached) * rate_in + cached * rate_in * CACHE_READ_SHARE + return (billed_in + tokens_out * rate_out) / 1_000_000 class VertexGeminiBackend: diff --git a/tests/harness/test_claude_cost.py b/tests/harness/test_claude_cost.py new file mode 100644 index 000000000..1c96c4a07 --- /dev/null +++ b/tests/harness/test_claude_cost.py @@ -0,0 +1,632 @@ +"""What a Claude stage records: each turn's own tokens, priced here, never the CLI's estimate. + +The Claude Code CLI reports running totals within one connection, so a stage that sends a second +message would count its first turn again if results were added as reported. The simulated +customer's tokens are billed from these figures, so they must be each query's own. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import threading +from datetime import date +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from types import SimpleNamespace + +import pytest +from claude_agent_sdk import ResultMessage + +from fi.alk.harness import spend +from fi.alk.harness.backends import ConversationSession, SessionSpec +from fi.alk.harness.backends import vertex_gemini +from fi.alk.harness.backends.claude import ClaudeBackend, ClaudeSession +from fi.alk.harness.backends.vertex_gemini import priced +from fi.alk.harness.session import Stage, Turn + +M = "gemini-3.7-flash" + + +@pytest.fixture(autouse=True) +def _a_priced_day(monkeypatch): + """Prices are a dated schedule; these tests are about counting, so the day is fixed.""" + + class _Today(date): + @classmethod + def today(cls): + return date(2026, 12, 1) + + monkeypatch.setattr(vertex_gemini, "date", _Today) + # A journal the machine names must not receive these tests' turns. + monkeypatch.delenv(spend.JOURNAL_ALIAS, raising=False) + spend._stages.clear() + spend.journal_to(None) + yield + spend._stages.clear() + spend.journal_to(None) + + +def _usage(tokens_in, tokens_out=0, cached=0): + return { + "inputTokens": tokens_in, + "outputTokens": tokens_out, + "cacheReadInputTokens": cached, + } + + +def _result(model_usage, *, usage=None, cli_cost=9.99): + return ResultMessage( + subtype="success", + duration_ms=1, + duration_api_ms=1, + is_error=False, + num_turns=1, + session_id="s", + total_cost_usd=cli_cost, + usage=usage, + model_usage=model_usage, + ) + + +def _session(*, resume=None, reported=None): + return ClaudeSession(SimpleNamespace(resume=resume), reported_model=reported) + + +def _done(session, message): + [done] = session._translate(message) + return done + + +class _NoProcess: + """The SDK client, minus the CLI: only the session's own bookkeeping is under test.""" + + def __init__(self, options): + pass + + async def connect(self): + pass + + async def disconnect(self): + pass + + +def test_two_results_on_one_connection_record_their_own_turns(): + session = _session() + first = _done(session, _result({M: _usage(100_000, 1_000)})) + second = _done(session, _result({M: _usage(200_000, 2_000)})) + + assert (first.tokens_in, first.tokens_out) == (100_000, 1_000) + assert (second.tokens_in, second.tokens_out) == (100_000, 1_000) + assert first.cost_usd is not None + assert second.cost_usd == pytest.approx(priced(M, 100_000, 1_000)) + + +def test_a_new_connection_takes_its_first_result_whole(monkeypatch): + """Larger than the old baseline on purpose: a kept baseline would give 50k, not fail.""" + monkeypatch.setattr("fi.alk.harness.backends.claude.ClaudeSDKClient", _NoProcess) + session = _session() + + async def reconnect(): + await session.start() + _done(session, _result({M: _usage(100_000, 1_000)})) + await session.stop() + await session.start() + return _done(session, _result({M: _usage(150_000, 1_500)})) + + again = asyncio.run(reconnect()) + + assert (again.tokens_in, again.tokens_out) == (150_000, 1_500) + + +def test_cache_reads_are_priced_at_the_cache_rate(): + """Cached input costs a tenth of fresh input on Vertex; pricing it as fresh overstates the turn.""" + done = _done(_session(), _result({M: _usage(100_000, 0, cached=900_000)})) + + assert (done.tokens_in, done.tokens_cached) == (1_000_000, 900_000) + assert done.cost_usd == pytest.approx(priced(M, 1_000_000, 0, 900_000)) + assert done.cost_usd == pytest.approx(0.1425) + + +def test_a_turn_with_input_and_no_output_is_still_a_report(): + """Only a result with no counts at all is "no report"; input alone was still spent.""" + done = _done(_session(), _result({M: _usage(1_000, 0)})) + + assert (done.tokens_in, done.tokens_out) == (1_000, 0) + assert done.cost_usd == pytest.approx(priced(M, 1_000, 0)) + assert done.cost_usd > 0 + + +def test_a_lower_count_takes_all_three_counts_whole(): + """Totals that went down restarted where the kit couldn't see; none of them is a difference.""" + session = _session() + _done(session, _result({M: _usage(100_000, 1_000, cached=50_000)})) + after = _done(session, _result({M: _usage(120_000, 500, cached=60_000)})) + following = _done(session, _result({M: _usage(130_000, 600, cached=70_000)})) + + assert (after.tokens_in, after.tokens_out, after.tokens_cached) == ( + 180_000, + 500, + 60_000, + ) + # The restarted totals are the new baseline, so the next result is only its own turn. + assert (following.tokens_in, following.tokens_out, following.tokens_cached) == ( + 20_000, + 100, + 10_000, + ) + + +def test_fewer_fresh_input_tokens_take_all_three_counts_whole(): + """Fresh input never falls within one connection, so a fall is an unseen restart. A difference + there would record more cached than input tokens, which the platform's authoring bill refuses.""" + session = _session() + _done(session, _result({M: _usage(100_000, 1_000, cached=10_000)})) + after = _done(session, _result({M: _usage(5_000, 1_100, cached=200_000)})) + + assert (after.tokens_in, after.tokens_out, after.tokens_cached) == ( + 205_000, + 1_100, + 200_000, + ) + assert after.tokens_cached <= after.tokens_in + + +def test_a_resumed_result_with_no_usage_says_so(caplog): + """Its carried totals can't be told apart from the old process's, so it records 0, aloud.""" + caplog.set_level(logging.WARNING, logger="fi.alk.harness.backends.claude") + done = _done( + _session(resume="earlier"), _result({M: _usage(500_000, 5_000)}, usage=None) + ) + + assert (done.tokens_in, done.tokens_out, done.tokens_cached) == (0, 0, 0) + assert "reported no usage" in caplog.text + + +def test_an_unpriced_model_is_unpriced_even_with_the_cli_estimate(): + """The CLI's estimate for a non-Claude model is a guess, about 7x Gemini's rate.""" + session = _session() + done = _done( + session, _result({"a-model-nobody-priced": _usage(1_000, 10)}, cli_cost=0.5) + ) + + assert done.cost_usd is None + Stage(SessionSpec(system_prompt="x"), name="customer")._events(done, Turn()) + ledger = spend.snapshot() + assert ledger["unpriced_turns"] == 1 + assert ledger["total_usd"] == 0.0 + + +def test_an_alias_on_the_wire_is_priced_by_the_reported_model(): + session = _session(reported=M) + done = _done(session, _result({"alias-on-the-wire": _usage(1_000_000, 10)})) + + assert done.cost_usd == pytest.approx(priced(M, 1_000_000, 10)) + assert done.models == {M} + + +@pytest.mark.parametrize( + "nothing", [{}, None, {M: _usage(0)}], ids=["empty", "missing", "all-zero"] +) +def test_a_result_with_no_usage_costs_nothing_and_keeps_the_baseline(nothing): + """A first-call error arrives with no usage. It cost nothing, and None would mark a reading + incomplete; resetting the baseline would count every earlier turn again.""" + session = _session() + _done(session, _result({M: _usage(100_000, 1_000)})) + empty = _done(session, _result(nothing, cli_cost=0.0)) + after = _done(session, _result({M: _usage(130_000, 1_300)})) + + assert (empty.tokens_in, empty.tokens_out, empty.tokens_cached) == (0, 0, 0) + assert empty.cost_usd == 0.0 + assert (after.tokens_in, after.tokens_out) == (30_000, 300) + + +def test_a_resumed_first_result_takes_usage_not_the_carried_totals(): + """SDK 0.2.176 on carries the old process's totals over a resume; `usage` is this query's.""" + session = _session(resume="earlier") + first = _done( + session, + _result( + {M: _usage(500_000, 5_000)}, + usage={ + "input_tokens": 30_000, + "cache_read_input_tokens": 10_000, + "output_tokens": 70, + }, + ), + ) + second = _done(session, _result({M: _usage(520_000, 5_007)})) + + assert (first.tokens_in, first.tokens_out, first.tokens_cached) == ( + 40_000, + 70, + 10_000, + ) + assert (second.tokens_in, second.tokens_out) == (20_000, 7) + + +def test_two_priced_models_with_no_alias_give_the_higher_price(): + session = _session() + done = _done( + session, + _result( + {M: _usage(600_000, 6_000), "gemini-3.5-flash": _usage(400_000, 4_000)} + ), + ) + + cheaper, dearer = ( + priced(M, 1_000_000, 10_000), + priced("gemini-3.5-flash", 1_000_000, 10_000), + ) + assert dearer > cheaper + assert done.cost_usd == pytest.approx(dearer) + + +# -- the bundled CLI against a local stub, through ClaudeBackend, ClaudeSession and Stage --------- + + +class _MessagesStub(BaseHTTPRequestHandler): + """The Messages API on 127.0.0.1. Every request gets its own usage, and is logged as the truth + the ledger must equal.""" + + served = 0 + log: list[tuple[int, int]] = [] + + def log_message(self, *args) -> None: + pass + + def _json(self, body: dict) -> None: + raw = json.dumps(body).encode() + self.send_response(200) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def _event(self, name: str, data: dict) -> None: + self.wfile.write(f"event: {name}\ndata: {json.dumps(data)}\n\n".encode()) + self.wfile.flush() + + def do_HEAD(self) -> None: + self.send_response(200) + self.end_headers() + + def do_GET(self) -> None: + self._json({"data": [], "has_more": False}) + + def do_POST(self) -> None: + raw = self.rfile.read(int(self.headers.get("content-length") or 0)) + try: + request = json.loads(raw) + except ValueError: + request = {} + if "count_tokens" in self.path: + self._json({"input_tokens": 10}) + return + model = request.get("model") or M + _MessagesStub.served += 1 + n = _MessagesStub.served + fresh, cached, written = 1_000 * n, 100 * n, 10 * n + _MessagesStub.log.append((fresh + cached, written)) + usage = { + "input_tokens": fresh, + "output_tokens": written, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": cached, + } + if not request.get("stream"): + self._json( + { + "id": f"msg_{n}", + "type": "message", + "role": "assistant", + "model": model, + "content": [{"type": "text", "text": "ok"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": usage, + } + ) + return + self.send_response(200) + self.send_header("content-type", "text/event-stream") + self.end_headers() + self._event( + "message_start", + { + "type": "message_start", + "message": { + "id": f"msg_{n}", + "type": "message", + "role": "assistant", + "model": model, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {**usage, "output_tokens": 1}, + }, + }, + ) + self._event( + "content_block_start", + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "text", "text": ""}, + }, + ) + self._event( + "content_block_delta", + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": "Sure, go on."}, + }, + ) + self._event("content_block_stop", {"type": "content_block_stop", "index": 0}) + self._event( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": written}, + }, + ) + self._event("message_stop", {"type": "message_stop"}) + + +@pytest.fixture +def cli(tmp_path, monkeypatch): + """The real CLI, aimed at the stub. The key is a placeholder, and every other route it could + take points at a closed local port.""" + _MessagesStub.served = 0 + _MessagesStub.log = [] + server = ThreadingHTTPServer(("127.0.0.1", 0), _MessagesStub) + threading.Thread(target=server.serve_forever, daemon=True).start() + nowhere = "http://127.0.0.1:9" + for name, value in { + "ALK_CLAUDE_GATEWAY_URL": f"http://127.0.0.1:{server.server_address[1]}", + "ALK_CLAUDE_GATEWAY_API_KEY": "placeholder-not-a-key", + "ANTHROPIC_VERTEX_BASE_URL": nowhere, + "ANTHROPIC_BEDROCK_BASE_URL": nowhere, + "ANTHROPIC_FOUNDRY_BASE_URL": nowhere, + "OTEL_EXPORTER_OTLP_ENDPOINT": nowhere, + "CLAUDE_CODE_ENABLE_TELEMETRY": "0", + "DISABLE_TELEMETRY": "1", + "DISABLE_ERROR_REPORTING": "1", + "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1", + "DISABLE_AUTOUPDATER": "1", + "CLAUDE_CONFIG_DIR": str(tmp_path / "config"), + }.items(): + monkeypatch.setenv(name, value) + for name in ( + "AGENTCC_BASE_URL", + "AGENTCC_API_KEY", + "AGENTCC_CLAUDE_MODEL_ALIAS", + "ALK_HARNESS", + "ALK_HARNESS_MODEL", + spend.JOURNAL_ALIAS, + ): + monkeypatch.delenv(name, raising=False) + # No alias on the wire, so the ledger's model is the one the table prices. + monkeypatch.setattr("fi.agentcc.gateway", lambda: None, raising=False) + work = tmp_path / "work" + work.mkdir() + # The customer's stage names no folder, so the CLI runs where the test does. + monkeypatch.chdir(work) + try: + yield work + finally: + server.shutdown() + server.server_close() + + +def _truth() -> tuple[int, int]: + return ( + sum(tokens_in for tokens_in, _ in _MessagesStub.log), + sum(tokens_out for _, tokens_out in _MessagesStub.log), + ) + + +def _spec(work, conversation=None) -> SessionSpec: + return SessionSpec( + system_prompt="Be brief.", + max_turns=3, + model=M, + gated=False, + cwd=str(work), + conversation=conversation, + ) + + +def _ledger(stage: str) -> dict: + [entry] = [one for one in spend.snapshot()["stages"] if one["stage"] == stage] + return entry + + +def test_the_cli_two_messages_on_one_connection_record_each_turn(cli): + async def talk(): + async with Stage(_spec(cli), name="author", backend=ClaudeBackend()) as stage: + await stage.say("one") + await stage.say("two") + + asyncio.run(asyncio.wait_for(talk(), 150)) + + entry = _ledger("author") + assert len(_MessagesStub.log) >= 2 + assert (entry["tokens_in"], entry["tokens_out"]) == _truth() + assert entry["unpriced"] == 0 + + +def test_the_cli_a_resume_then_one_more_message_is_the_true_sum(cli): + """Newer SDKs carry the totals over a resume, older ones restart them; either way the ledger + is what the queries used.""" + + def conversation(resume=None): + return ConversationSession( + app_name="alk", + user_id="u", + session_id="conversation", + resume_session_id=resume, + ) + + async def talk(): + async with Stage( + _spec(cli, conversation()), name="chat", backend=ClaudeBackend() + ) as stage: + await stage.say("one") + await stage.say("two") + assert stage.session_id + async with Stage( + _spec(cli, conversation(stage.session_id)), + name="chat", + backend=ClaudeBackend(), + ) as again: + await again.say("three") + + asyncio.run(asyncio.wait_for(talk(), 150)) + + entry = _ledger("chat") + assert len(_MessagesStub.log) >= 3 + assert (entry["tokens_in"], entry["tokens_out"]) == _truth() + + +def test_the_cli_customer_lines_bill_their_true_tokens(cli): + """The simulated customer is billed from `simulator_input_tokens`, one query per line.""" + from fi.alk.harness.contract import AgentContract + from fi.alk.harness.run import conversation + from fi.alk.harness.scenario import Scenario + + async def agent_replies(text): + return "Your order ships tomorrow." + + async def nothing(*_args): + return None + + target = SimpleNamespace( + open=nothing, say=agent_replies, close=nothing, spent_usd=0.0 + ) + # The opening and two answers: three customer lines on one connection. + scenario = Scenario( + name="order", instruction="Ask where your order is.", max_turns=2 + ) + + transcript = asyncio.run( + asyncio.wait_for( + conversation.converse(target, scenario, AgentContract(), model=M), 150 + ) + ) + + assert [one.speaker for one in transcript.exchanges].count("customer") == 3 + assert len(_MessagesStub.log) >= 3 + assert ( + transcript.simulator_input_tokens, + transcript.simulator_output_tokens, + ) == _truth() + + +# -- #225's tripwire ------------------------------------------------------------------------------ + + +def test_a_claude_reading_adds_per_turn_figures_as_they_come(tmp_path, monkeypatch): + """Skipped until #225's reading lands. It subtracts Claude's running totals itself, which on + per-turn figures would record the first turn and nothing after it.""" + agent_reading = pytest.importorskip("fi.alk.harness.agent_reading") + + from fi.alk.harness import reading_entry, session + from fi.alk.harness.backends import StageDone + + class _LoudExit: + """A reading's hard limit ending this process is a failure, not a silent exit.""" + + def __getattr__(self, name): + return getattr(os, name) + + @staticmethod + def _exit(_code): + os.write(2, b"a reading's hard limit ended the test process\n") + os._exit(97) + + class _Live: + async def start(self): + pass + + async def stop(self): + pass + + async def send(self, message): + pass + + async def interrupt(self): + return True + + async def replies(self): + # What this backend now reports: each turn's own cost, not the running total. + yield StageDone( + outcome="success", turns=1, cost_usd=0.07875, tokens_in=100_000 + ) + + class _Claude: + name = "claude" + default_model = vertex_gemini.DEFAULT_MODEL + + def create(self, spec): + return _Live() + + def can_drive(self, model): + return True + + monkeypatch.setattr(reading_entry, "os", _LoudExit()) + monkeypatch.setattr(reading_entry, "_SECRET_FILES", (tmp_path / "no-secrets.json",)) + monkeypatch.setenv("ALK_HARNESS_MODEL", "gemini-2.5-flash") + monkeypatch.setenv("ALK_READING_MAX_DURATION_SECONDS", "20") + monkeypatch.setattr(session, "resolve", lambda *a, **k: _Claude()) + source = tmp_path / "source" / "agents" / "support" + source.mkdir(parents=True) + (source / "agent.py").write_text("x = 1\n", encoding="utf-8") + job = tmp_path / "job.json" + job.write_text( + json.dumps( + { + "job_id": "job", + "run_id": "run", + "execution": "hosted", + "source": { + "kind": "github", + "visibility": "public", + "repository": "acme/agent", + "commit_sha": "a" * 40, + }, + "runtime": {"isolation": "dedicated_vm"}, + "agent": {"connector": "auto"}, + } + ), + encoding="utf-8", + ) + output = tmp_path / "reading" + + # The opening and the reminder to submit: two turns, each with its own figures. + reading_entry.run_reading_mode( + agent_reading._read_agent(interrupt_wait=1.0), + job_path=job, + source=tmp_path / "source", + output=output, + ) + + cost = json.loads((output / "cost.json").read_text(encoding="utf-8")) + assert cost["total_usd"] == pytest.approx(0.1575) + + +def test_ci_runs_these_tests(): + """The check-by date is enforced by CI going red, which only happens if CI runs it.""" + workflow = ( + Path(__file__).resolve().parents[2] / ".github" / "workflows" / "sdk-smoke.yml" + ) + # A whole line of the pytest list, so a commented-out or renamed entry doesn't count. + listed = { + line.strip() for line in workflow.read_text(encoding="utf-8").splitlines() + } + assert "tests/harness/test_spend.py" in listed + assert "tests/harness/test_claude_cost.py" in listed diff --git a/tests/harness/test_spend.py b/tests/harness/test_spend.py index 8bc84238d..8e61212fb 100644 --- a/tests/harness/test_spend.py +++ b/tests/harness/test_spend.py @@ -106,30 +106,149 @@ def test_a_session_reports_its_tokens_too(tmp_path): assert (entry["tokens_in"], entry["tokens_out"]) == (9000, 2500) -def test_a_price_that_has_run_out_refuses_instead_of_billing_it(monkeypatch): - """gemini-3.7-flash is introductory pricing that doubles on 2027-01-01.""" +def _on(monkeypatch, day): + """Price as if today were `day`; the table is a schedule, so a price depends on the date.""" from datetime import date from fi.alk.harness.backends import vertex_gemini - class _Before(date): + class _Today(date): @classmethod def today(cls): - return date(2026, 12, 31) + return day - class _After(date): - @classmethod - def today(cls): - return date(2027, 1, 1) + monkeypatch.setattr(vertex_gemini, "date", _Today) - monkeypatch.setattr(vertex_gemini, "date", _Before) - assert vertex_gemini.priced("gemini-3.7-flash", 1_000_000, 1_000_000) == 4.5 - monkeypatch.setattr(vertex_gemini, "date", _After) - assert vertex_gemini.priced("gemini-3.7-flash", 1_000_000, 1_000_000) is None +@pytest.mark.parametrize("model", ["gemini-3.6-flash", "gemini-3.7-flash", "gemini-3.8-flash"]) +def test_a_price_that_has_run_out_refuses_instead_of_billing_it(monkeypatch, caplog, model): + """The Flash models double on 2027-01-01, as Google's Vertex pricing page says + (cloud.google.com/model-garden/pricing, Global tier, read 2026-11-10). Past the table's + re-check date they have no price at all, and say so rather than billing a stale one.""" + import logging + from datetime import date + + from fi.alk.harness.backends import vertex_gemini + + _on(monkeypatch, date(2026, 12, 31)) + assert vertex_gemini.priced(model, 1_000_000, 1_000_000) == 4.5 + assert vertex_gemini.priced(model, 1_000_000, 0, 1_000_000) == pytest.approx(0.075) + + _on(monkeypatch, date(2027, 1, 1)) + assert vertex_gemini.priced(model, 1_000_000, 1_000_000) == 9.0 + assert vertex_gemini.priced(model, 1_000_000, 0, 1_000_000) == pytest.approx(0.15) + + _on(monkeypatch, date(2027, 4, 1)) + caplog.set_level(logging.WARNING, logger=vertex_gemini.__name__) + assert vertex_gemini.priced(model, 1_000_000, 1_000_000) is None + assert f"no current price for {model}" in caplog.text assert vertex_gemini.priced("a-model-nobody-listed", 1_000, 1_000) is None +def test_every_price_is_re_checked_by_the_check_by_date(): + """Runs on the real date, so the table can't quietly outlive its prices: from 2027-03-01 this + is red on every PR until each model is re-checked against the gateway table and Google.""" + from datetime import date, timedelta + + from fi.alk.harness.backends.vertex_gemini import CHECK_BY, PRICES_PER_MILLION + + assert CHECK_BY == "2027-03-31" + for name, rows in PRICES_PER_MILLION.items(): + dated = sorted(rows, key=lambda row: row[2]) + assert dated[-1][3] == CHECK_BY, f"{name}'s latest row is not re-checked by {CHECK_BY}" + for row in dated: + assert row[2] <= row[3], f"{name} has a row that ends before it starts: {row}" + for earlier, later in zip(dated, dated[1:]): + # A gap is a day nothing is priced; an overlap is a day priced twice. + assert date.fromisoformat(later[2]) == date.fromisoformat(earlier[3]) + timedelta( + days=1 + ), f"{name}'s rows do not follow on: {earlier} then {later}" + assert (date.fromisoformat(CHECK_BY) - date.today()).days > 30, ( + f"the price table must be re-checked by {CHECK_BY}: compare every row with the gateway's " + "litellm.json and Google's Vertex pricing page, then move CHECK_BY on" + ) + + +def test_the_rate_is_the_row_whose_dates_hold_today(monkeypatch): + """The row is chosen by its dates, never by its place in the table.""" + from datetime import date + + from fi.alk.harness.backends import vertex_gemini + from fi.alk.harness.backends.vertex_gemini import current_rates + + rows = vertex_gemini.PRICES_PER_MILLION["gemini-3.7-flash"] + newest_first = tuple(sorted(rows, key=lambda row: row[2], reverse=True)) + monkeypatch.setitem(vertex_gemini.PRICES_PER_MILLION, "gemini-3.7-flash", newest_first) + + assert current_rates("gemini-3.7-flash", date(2026, 12, 31)) == (0.75, 3.75) + assert current_rates("gemini-3.7-flash", date(2027, 1, 1)) == (1.50, 7.50) + assert current_rates("gemini-3.7-flash", date(2026, 11, 1)) is None + assert current_rates("vertex_ai/gemini-3.7-flash", date(2027, 1, 1)) == (1.50, 7.50) + assert current_rates("a-model-nobody-listed", date(2027, 1, 1)) is None + + _on(monkeypatch, date(2026, 12, 31)) + assert vertex_gemini.priced("gemini-3.7-flash", 1_000_000, 1_000_000) == 4.5 + _on(monkeypatch, date(2027, 1, 1)) + assert vertex_gemini.priced("gemini-3.7-flash", 1_000_000, 1_000_000) == 9.0 + _on(monkeypatch, date(2026, 11, 1)) + assert vertex_gemini.priced("gemini-3.7-flash", 1_000_000, 1_000_000) is None + + +def test_a_two_call_gemini_stage_records_the_plain_sum(tmp_path, monkeypatch): + """Gemini reports each call on its own, so its ledger is the calls added up, unchanged by + the Claude backend's running-total handling.""" + from datetime import date + + from google.adk.models.google_llm import Gemini + from google.adk.models.llm_response import LlmResponse + from google.genai import types + + from fi.alk.harness.backends import vertex_gemini + from fi.alk.harness.backends.vertex_gemini import VertexGeminiBackend + + calls = iter([(1_000, 100, 200), (3_000, 300, 600)]) + + async def generate(self, llm_request, stream=False): + prompt, candidates, cached = next(calls) + yield LlmResponse( + content=types.Content(role="model", parts=[types.Part(text="ok")]), + usage_metadata=types.GenerateContentResponseUsageMetadata( + prompt_token_count=prompt, + candidates_token_count=candidates, + cached_content_token_count=cached, + ), + ) + + monkeypatch.setattr(Gemini, "generate_content_async", generate) + monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "test-project") + monkeypatch.setenv("GOOGLE_CLOUD_LOCATION", "global") + monkeypatch.setenv("GOOGLE_GENAI_USE_VERTEXAI", "TRUE") + _on(monkeypatch, date(2026, 12, 1)) + journal = _fresh(tmp_path) + + async def run(): + async with Stage( + SessionSpec(system_prompt="Reply briefly", max_turns=1, model="gemini-3.7-flash"), + name="build-environment", + backend=VertexGeminiBackend(), + overheard=False, + ) as stage: + await stage.say("one") + await stage.say("two") + + import asyncio + + asyncio.run(run()) + + entry = json.loads(journal.read_text(encoding="utf-8"))["stages"][0] + assert (entry["tokens_in"], entry["tokens_out"], entry["tokens_cached"]) == (4_000, 400, 800) + assert entry["usd"] == pytest.approx( + vertex_gemini.priced("gemini-3.7-flash", 1_000, 100, 200) + + vertex_gemini.priced("gemini-3.7-flash", 3_000, 300, 600) + ) + assert entry["unpriced"] == 0 + + def test_both_backends_report_the_same_units(): """One backend giving auditable tokens and the other not is a ledger you cannot reconcile.""" from fi.alk.harness.backends.claude import _tokens @@ -199,11 +318,14 @@ def test_every_harness_stage_feeds_the_one_ledger(): def test_the_price_table_agrees_with_the_platform_model_table(): - """Prices are copied from the platform's litellm table, so drift is a silent mis-bill.""" + """Prices are copied from the platform's litellm table, so drift is a silent mis-bill. + + From 2027-01-01 this is red for the three Flash models until the gateway's table carries + their 2027 rows too (TH-9508): both tables must change together.""" import json from pathlib import Path - from fi.alk.harness.backends.vertex_gemini import PRICES_PER_MILLION + from fi.alk.harness.backends.vertex_gemini import PRICES_PER_MILLION, current_rates # The platform checked out beside this repo, or nowhere: never an absolute path, which would # carry one machine's layout into a public repo and skip for everyone else. @@ -221,16 +343,23 @@ def test_the_price_table_agrees_with_the_platform_model_table(): pytest.skip("platform checkout not beside this repo") models = json.loads(table.read_text(encoding="utf-8")) - for name, (want_in, want_out, _good_until) in PRICES_PER_MILLION.items(): + # Today's rates only: the gateway's table holds no schedule (TH-9508 adds the 2027 change). + for name in PRICES_PER_MILLION: + rates = current_rates(name) + assert rates, f"{name} has no price for today" + want_in, want_out = rates entry = models.get(f"vertex_ai/{name}") or models.get(name) assert entry, f"{name} is priced here but absent from the platform table" assert round(entry["input_cost_per_token"] * 1_000_000, 6) == want_in, name assert round(entry["output_cost_per_token"] * 1_000_000, 6) == want_out, name -def test_a_gateway_route_in_front_of_the_model_does_not_lose_its_price(): +def test_a_gateway_route_in_front_of_the_model_does_not_lose_its_price(monkeypatch): + from datetime import date + from fi.alk.harness.backends import vertex_gemini + _on(monkeypatch, date(2026, 12, 1)) bare = vertex_gemini.priced("gemini-3.7-flash", 1_000_000, 1_000_000) assert bare is not None assert vertex_gemini.priced("vertex_ai/gemini-3.7-flash", 1_000_000, 1_000_000) == bare diff --git a/tests/harness/test_vertex_gemini_backend.py b/tests/harness/test_vertex_gemini_backend.py index ebf7890a2..d7bd318ae 100644 --- a/tests/harness/test_vertex_gemini_backend.py +++ b/tests/harness/test_vertex_gemini_backend.py @@ -277,9 +277,19 @@ def test_a_tool_call_carries_the_agent_that_made_it() -> None: assert Call(id="c1", name="x", by="scenario_writer").by == "scenario_writer" -def test_a_cache_read_is_not_charged_at_the_full_input_rate() -> None: +def test_a_cache_read_is_not_charged_at_the_full_input_rate(monkeypatch) -> None: + from datetime import date + + from fi.alk.harness.backends import vertex_gemini from fi.alk.harness.backends.vertex_gemini import CACHE_READ_SHARE, priced + class _Today(date): + @classmethod + def today(cls): + # The 0.75 rate holds through 2026-12-31: a price is a schedule of dated rows. + return date(2026, 12, 1) + + monkeypatch.setattr(vertex_gemini, "date", _Today) full = priced("gemini-3.7-flash", 1_000_000, 0, 0) all_cached = priced("gemini-3.7-flash", 1_000_000, 0, 1_000_000) assert full == 0.75 From a232700b39322b128b1eb3632f598b3829d07963 Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Wed, 11 Nov 2026 11:32:57 +0530 Subject: [PATCH 2/5] fix(harness): guard a turn below its own usage, test on the locked SDK, pin today's prices, price on UTC days Review on #274 (Dileep). A recorded turn below the query's own usage now takes the totals whole, so totals that restarted and came back above the baseline, or a CLI that reports each query on its own, can't under-record. CI reruns the Claude cost tests on the SDK version uv.lock pins, which the hosted image uses. Today's prices are checked in CI against a committed copy of the gateway's rows at 9ec5f9fb1, not only beside a future-agi checkout. Prices switch on the UTC day, as the platform's dated rows do. TH-9057 TH-9039 Signed-off-by: khushalsonawat --- .github/workflows/sdk-smoke.yml | 7 ++ src/fi/alk/harness/backends/claude.py | 6 ++ src/fi/alk/harness/backends/vertex_gemini.py | 17 ++-- .../fixtures/litellm_vertex_prices.json | 77 +++++++++++++++++++ tests/harness/test_claude_cost.py | 29 +++++-- tests/harness/test_spend.py | 46 +++++++++-- tests/harness/test_vertex_gemini_backend.py | 9 +-- 7 files changed, 167 insertions(+), 24 deletions(-) create mode 100644 tests/harness/fixtures/litellm_vertex_prices.json diff --git a/.github/workflows/sdk-smoke.yml b/.github/workflows/sdk-smoke.yml index 10809e1d4..51d7fc8a1 100644 --- a/.github/workflows/sdk-smoke.yml +++ b/.github/workflows/sdk-smoke.yml @@ -74,3 +74,10 @@ jobs: tests/test_release_workflow.py tests/test_release_pr_step.py tests/episode + - name: Rerun the Claude cost tests on the locked SDK + # The hosted image installs from uv.lock, while the step above gets pip's newest SDK. The + # two handle a resumed session differently (TH-9057), so the cost tests run on both. + run: | + version=$(python -c "import tomllib; lock = tomllib.load(open('uv.lock', 'rb')); print(next(p['version'] for p in lock['package'] if p['name'] == 'claude-agent-sdk'))") + python -m pip install "claude-agent-sdk==$version" + python -m pytest -q tests/harness/test_claude_cost.py diff --git a/src/fi/alk/harness/backends/claude.py b/src/fi/alk/harness/backends/claude.py index 45d851166..abd9e4c01 100644 --- a/src/fi/alk/harness/backends/claude.py +++ b/src/fi/alk/harness/backends/claude.py @@ -369,6 +369,12 @@ def _own(self, received: Any) -> dict[str, int] | None: own = totals else: own = {count: totals[count] - self._so_far[count] for count in totals} + # A difference below the query's own usage means the totals weren't running on: + # restarted and already back above the baseline, or a CLI that reports each query on + # its own. usage is this query's main-thread tokens, so the turn is never less. + per_query = _tokens({"usage": getattr(received, "usage", None) or {}}) + if any(own[count] < per_query[count] for count in own): + own = totals self._so_far = totals return own diff --git a/src/fi/alk/harness/backends/vertex_gemini.py b/src/fi/alk/harness/backends/vertex_gemini.py index b8a4787eb..768cd8362 100644 --- a/src/fi/alk/harness/backends/vertex_gemini.py +++ b/src/fi/alk/harness/backends/vertex_gemini.py @@ -24,7 +24,7 @@ import uuid from collections.abc import AsyncIterator from contextlib import aclosing -from datetime import date +from datetime import date, datetime, timezone from typing import Any from fi import agentcc @@ -53,8 +53,9 @@ # litellm.json, its vertex_ai/ entries, future-agi 9ec5f9fb1, 2026-11-02). The 2027 Flash rows are # Google's Vertex pricing page (cloud.google.com/model-garden/pricing, Global tier, read # 2026-11-10), which that table does not carry yet. A row ends either on a published price change -# (the 2026 Flash rows) or on CHECK_BY, the day every latest row must be re-checked by. A model -# with no row for today reports no cost rather than a wrong one, and shows up in `unpriced_turns`. +# (the 2026 Flash rows) or on CHECK_BY, the day every latest row must be re-checked by. Days are +# UTC days, as the platform's dated rows are. A model with no row for today reports no cost rather +# than a wrong one, and shows up in `unpriced_turns`. CHECK_BY = "2027-03-31" # What Vertex charges for a cache read, as a share of the input rate. @@ -703,12 +704,18 @@ def _rows(name: str) -> tuple[tuple[Any, ...], ...] | None: return rows +def _today() -> date: + """The price day, in UTC: the platform's dated price rows switch at midnight UTC, so the kit's + rows do too, wherever the process runs.""" + return datetime.now(timezone.utc).date() + + def current_rates(name: str, today: date | None = None) -> tuple[float, float] | None: """The (input, output) rates per 1M tokens that hold on `today`, or None if no row does. Read the table through this rather than by position: a model's price is a schedule of rows. """ - day = (today or date.today()).isoformat() + day = (today or _today()).isoformat() # Every row is checked whatever its place, so the table's order never picks the price; the # check-by test keeps a model's rows from overlapping, so at most one row holds. for rate_in, rate_out, first_day, last_day in _rows(name) or (): @@ -724,7 +731,7 @@ def priced( rows = _rows(model) if rows is None: return None - today = date.today() + today = _today() rates = current_rates(model, today) if rates is None: logger.warning( diff --git a/tests/harness/fixtures/litellm_vertex_prices.json b/tests/harness/fixtures/litellm_vertex_prices.json new file mode 100644 index 000000000..f7f63a3dc --- /dev/null +++ b/tests/harness/fixtures/litellm_vertex_prices.json @@ -0,0 +1,77 @@ +{ + "_source": "agentcc-gateway/internal/modeldb/litellm.json at future-agi 9ec5f9fb1 (2026-11-02): the input and output prices of the vertex_ai/ entries the kit prices.", + "rows": { + "vertex_ai/gemini-3-flash-preview": { + "input_cost_per_token": 5e-07, + "output_cost_per_token": 3e-06 + }, + "vertex_ai/gemini-3-pro-image": { + "input_cost_per_token": 2e-06, + "output_cost_per_token": 1.2e-05 + }, + "vertex_ai/gemini-3-pro-image-preview": { + "input_cost_per_token": 2e-06, + "output_cost_per_token": 1.2e-05 + }, + "vertex_ai/gemini-3-pro-preview": { + "input_cost_per_token": 2e-06, + "output_cost_per_token": 1.2e-05 + }, + "vertex_ai/gemini-3.1-flash-image": { + "input_cost_per_token": 5e-07, + "output_cost_per_token": 3e-06 + }, + "vertex_ai/gemini-3.1-flash-image-preview": { + "input_cost_per_token": 5e-07, + "output_cost_per_token": 3e-06 + }, + "vertex_ai/gemini-3.1-flash-lite": { + "input_cost_per_token": 2.5e-07, + "output_cost_per_token": 1.5e-06 + }, + "vertex_ai/gemini-3.1-flash-lite-image": { + "input_cost_per_token": 2.5e-07, + "output_cost_per_token": 1.5e-06 + }, + "vertex_ai/gemini-3.1-flash-lite-preview": { + "input_cost_per_token": 2.5e-07, + "output_cost_per_token": 1.5e-06 + }, + "vertex_ai/gemini-3.1-pro-preview": { + "input_cost_per_token": 2e-06, + "output_cost_per_token": 1.2e-05 + }, + "vertex_ai/gemini-3.1-pro-preview-customtools": { + "input_cost_per_token": 2e-06, + "output_cost_per_token": 1.2e-05 + }, + "vertex_ai/gemini-3.5-flash": { + "input_cost_per_token": 1.5e-06, + "output_cost_per_token": 9e-06 + }, + "vertex_ai/gemini-3.5-flash-lite": { + "input_cost_per_token": 3e-07, + "output_cost_per_token": 2.5e-06 + }, + "vertex_ai/gemini-3.5-transcribe-live-preview": { + "input_cost_per_token": 3.5e-06, + "output_cost_per_token": 2.1e-05 + }, + "vertex_ai/gemini-3.5-transcribe-preview": { + "input_cost_per_token": 2e-06, + "output_cost_per_token": 1.2e-05 + }, + "vertex_ai/gemini-3.6-flash": { + "input_cost_per_token": 7.5e-07, + "output_cost_per_token": 3.75e-06 + }, + "vertex_ai/gemini-3.7-flash": { + "input_cost_per_token": 7.5e-07, + "output_cost_per_token": 3.75e-06 + }, + "vertex_ai/gemini-3.8-flash": { + "input_cost_per_token": 7.5e-07, + "output_cost_per_token": 3.75e-06 + } + } +} diff --git a/tests/harness/test_claude_cost.py b/tests/harness/test_claude_cost.py index 1c96c4a07..eacedac21 100644 --- a/tests/harness/test_claude_cost.py +++ b/tests/harness/test_claude_cost.py @@ -34,12 +34,7 @@ def _a_priced_day(monkeypatch): """Prices are a dated schedule; these tests are about counting, so the day is fixed.""" - class _Today(date): - @classmethod - def today(cls): - return date(2026, 12, 1) - - monkeypatch.setattr(vertex_gemini, "date", _Today) + monkeypatch.setattr(vertex_gemini, "_today", lambda: date(2026, 12, 1)) # A journal the machine names must not receive these tests' turns. monkeypatch.delenv(spend.JOURNAL_ALIAS, raising=False) spend._stages.clear() @@ -139,6 +134,28 @@ def test_a_turn_with_input_and_no_output_is_still_a_report(): assert done.cost_usd > 0 +def test_a_difference_below_the_querys_own_usage_takes_the_totals_whole(): + """Totals that restarted and came back above the baseline, or a CLI that reports each query on + its own, would otherwise record less than the query's own usage (Dileep, #274).""" + session = _session() + _done( + session, + _result( + {M: _usage(100_000, 1_000)}, + usage={"input_tokens": 100_000, "output_tokens": 1_000}, + ), + ) + second = _done( + session, + _result( + {M: _usage(120_000, 1_200)}, + usage={"input_tokens": 120_000, "output_tokens": 1_200}, + ), + ) + + assert (second.tokens_in, second.tokens_out) == (120_000, 1_200) + + def test_a_lower_count_takes_all_three_counts_whole(): """Totals that went down restarted where the kit couldn't see; none of them is a difference.""" session = _session() diff --git a/tests/harness/test_spend.py b/tests/harness/test_spend.py index 8e61212fb..eb4f48a8b 100644 --- a/tests/harness/test_spend.py +++ b/tests/harness/test_spend.py @@ -108,16 +108,30 @@ def test_a_session_reports_its_tokens_too(tmp_path): def _on(monkeypatch, day): """Price as if today were `day`; the table is a schedule, so a price depends on the date.""" - from datetime import date + from fi.alk.harness.backends import vertex_gemini + + monkeypatch.setattr(vertex_gemini, "_today", lambda: day) + + +def test_the_price_day_is_the_utc_day(monkeypatch): + """The platform's dated rows switch at midnight UTC, so the kit's do too (Dileep, #274).""" + from datetime import datetime, timezone from fi.alk.harness.backends import vertex_gemini - class _Today(date): - @classmethod - def today(cls): - return day + def at(moment): + class _Clock(datetime): + @classmethod + def now(cls, tz=None): + assert tz is timezone.utc + return moment + + monkeypatch.setattr(vertex_gemini, "datetime", _Clock) - monkeypatch.setattr(vertex_gemini, "date", _Today) + at(datetime(2026, 12, 31, 23, 30, tzinfo=timezone.utc)) + assert vertex_gemini.current_rates("gemini-3.7-flash") == (0.75, 3.75) + at(datetime(2027, 1, 1, 0, 30, tzinfo=timezone.utc)) + assert vertex_gemini.current_rates("gemini-3.7-flash") == (1.50, 7.50) @pytest.mark.parametrize("model", ["gemini-3.6-flash", "gemini-3.7-flash", "gemini-3.8-flash"]) @@ -317,6 +331,26 @@ def test_every_harness_stage_feeds_the_one_ledger(): ) +def test_todays_prices_match_the_gateway_table_as_committed(): + """Runs in CI, unlike the live comparison below: the gateway's rows at 9ec5f9fb1, committed as + a fixture, so a mistyped price fails here (Dileep, #274).""" + import json + from datetime import date + from pathlib import Path + + from fi.alk.harness.backends.vertex_gemini import PRICES_PER_MILLION, current_rates + + fixture = Path(__file__).parent / "fixtures" / "litellm_vertex_prices.json" + rows = json.loads(fixture.read_text(encoding="utf-8"))["rows"] + assert sorted(rows) == sorted(f"vertex_ai/{name}" for name in PRICES_PER_MILLION) + for name in PRICES_PER_MILLION: + row = rows[f"vertex_ai/{name}"] + assert current_rates(name, date(2026, 12, 1)) == ( + round(row["input_cost_per_token"] * 1_000_000, 6), + round(row["output_cost_per_token"] * 1_000_000, 6), + ), name + + def test_the_price_table_agrees_with_the_platform_model_table(): """Prices are copied from the platform's litellm table, so drift is a silent mis-bill. diff --git a/tests/harness/test_vertex_gemini_backend.py b/tests/harness/test_vertex_gemini_backend.py index d7bd318ae..446b96d40 100644 --- a/tests/harness/test_vertex_gemini_backend.py +++ b/tests/harness/test_vertex_gemini_backend.py @@ -283,13 +283,8 @@ def test_a_cache_read_is_not_charged_at_the_full_input_rate(monkeypatch) -> None from fi.alk.harness.backends import vertex_gemini from fi.alk.harness.backends.vertex_gemini import CACHE_READ_SHARE, priced - class _Today(date): - @classmethod - def today(cls): - # The 0.75 rate holds through 2026-12-31: a price is a schedule of dated rows. - return date(2026, 12, 1) - - monkeypatch.setattr(vertex_gemini, "date", _Today) + # The 0.75 rate holds through 2026-12-31: a price is a schedule of dated rows. + monkeypatch.setattr(vertex_gemini, "_today", lambda: date(2026, 12, 1)) full = priced("gemini-3.7-flash", 1_000_000, 0, 0) all_cached = priced("gemini-3.7-flash", 1_000_000, 0, 1_000_000) assert full == 0.75 From ffa61b7800838f56de340c3f51b2294b9dc66a75 Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Wed, 11 Nov 2026 13:22:40 +0530 Subject: [PATCH 3/5] fix(harness): keep pricing past CHECK_BY at the latest row, and check the date weekly Past CHECK_BY a Gemini model kept no price, so every turn in a running release went unpriced. It now bills at its latest row's rate and warns once per model per UTC day. A model with no row, or a day before its first row, still reports no cost and counts in unpriced_turns. The re-check date no longer runs on PRs. A weekly scheduled workflow, price-check-by.yml, runs it from a month before CHECK_BY. It fails if the test is skipped, and a separate notify job opens or updates one issue on a scheduled failure. A PR test pins the workflow, its env var and the test's skip together. Refs: TH-9057 Signed-off-by: khushalsonawat --- .github/workflows/price-check-by.yml | 59 +++++++++ src/fi/alk/harness/backends/vertex_gemini.py | 40 ++++-- tests/harness/test_claude_cost.py | 3 +- tests/harness/test_spend.py | 123 +++++++++++++++++-- 4 files changed, 202 insertions(+), 23 deletions(-) create mode 100644 .github/workflows/price-check-by.yml diff --git a/.github/workflows/price-check-by.yml b/.github/workflows/price-check-by.yml new file mode 100644 index 000000000..323949160 --- /dev/null +++ b/.github/workflows/price-check-by.yml @@ -0,0 +1,59 @@ +name: Price check-by + +# The Gemini price table carries a re-check date (CHECK_BY in vertex_gemini.py). This goes red +# from a month before it, on its own schedule, so the reminder never blocks a PR. GitHub runs +# schedules on the default branch only, so this checks main's table. A red scheduled run opens or +# updates one issue, so the notice reaches the team and not only whoever last edited the cron. +permissions: + contents: read + +on: + schedule: + - cron: "0 6 * * 1" + workflow_dispatch: + +jobs: + check-by: + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + - name: Install package + run: python -m pip install . pytest + - name: Check the price table's re-check date + env: + ALK_PRICE_CHECK_BY: "1" + # A skip exits 0, so the job checks that the test ran and passed. + run: | + set -o pipefail + python -m pytest -q -rs "tests/harness/test_spend.py::test_the_price_table_is_re_checked_before_its_check_by_date" | tee pytest.out + grep -q "1 passed" pytest.out + + # Separate from the check, so the token that can write issues never reaches a job that installs + # packages. + notify: + needs: check-by + if: failure() && github.event_name == 'schedule' + runs-on: ubuntu-latest + timeout-minutes: 5 + permissions: + issues: write + steps: + - name: Open or update the re-check issue + env: + GH_TOKEN: ${{ github.token }} + run: | + title="Gemini price table: re-check before CHECK_BY" + body="The weekly price-check-by run failed: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}. Compare every row in vertex_gemini.py with the gateway's litellm.json and Google's Vertex pricing page, then move CHECK_BY on." + number=$(gh issue list --repo "$GITHUB_REPOSITORY" --state open --search "\"$title\" in:title" --json number --jq '.[0].number // empty') + if [ -n "$number" ]; then + gh issue comment "$number" --repo "$GITHUB_REPOSITORY" --body "$body" + else + gh issue create --repo "$GITHUB_REPOSITORY" --title "$title" --body "$body" + fi diff --git a/src/fi/alk/harness/backends/vertex_gemini.py b/src/fi/alk/harness/backends/vertex_gemini.py index 768cd8362..8878a5795 100644 --- a/src/fi/alk/harness/backends/vertex_gemini.py +++ b/src/fi/alk/harness/backends/vertex_gemini.py @@ -54,8 +54,10 @@ # Google's Vertex pricing page (cloud.google.com/model-garden/pricing, Global tier, read # 2026-11-10), which that table does not carry yet. A row ends either on a published price change # (the 2026 Flash rows) or on CHECK_BY, the day every latest row must be re-checked by. Days are -# UTC days, as the platform's dated rows are. A model with no row for today reports no cost rather -# than a wrong one, and shows up in `unpriced_turns`. +# UTC days, as the platform's dated rows are. Past CHECK_BY a model keeps its latest row's price, +# with a warning, so a running release keeps counting cost; a weekly scheduled +# workflow goes red a month before, so the table is re-checked without blocking PRs. A model with +# no row at all reports no cost rather than a wrong one, and shows up in `unpriced_turns`. CHECK_BY = "2027-03-31" # What Vertex charges for a cache read, as a share of the input rate. @@ -704,6 +706,10 @@ def _rows(name: str) -> tuple[tuple[Any, ...], ...] | None: return rows +# (model, day) pairs already warned about, so a stale price warns once a day, not once a turn. +_STALE_WARNED: set[tuple[str, date]] = set() + + def _today() -> date: """The price day, in UTC: the platform's dated price rows switch at midnight UTC, so the kit's rows do too, wherever the process runs.""" @@ -734,14 +740,28 @@ def priced( today = _today() rates = current_rates(model, today) if rates is None: - logger.warning( - "no current price for %s on %s: the table's rows run from %s to %s", - model, - today.isoformat(), - min(row[2] for row in rows), - max(row[3] for row in rows), - ) - return None + latest = max(rows, key=lambda row: row[3]) + if today.isoformat() <= latest[3]: + # Before the model's first row: there is no price to stand behind. + logger.warning( + "no current price for %s on %s: the table's rows run from %s to %s", + model, + today.isoformat(), + min(row[2] for row in rows), + latest[3], + ) + return None + # Past the re-check date the latest price is still the best one known. It is off by + # whatever changed since, either way; None would stop every turn's cost. + if (model, today) not in _STALE_WARNED: + _STALE_WARNED.add((model, today)) + logger.warning( + "price for %s is past its re-check date %s: billing at that row's rate; re-check " + "the table against the gateway's litellm.json and Google's Vertex pricing page", + model, + latest[3], + ) + rates = latest[0], latest[1] rate_in, rate_out = rates cached = min(max(tokens_cached, 0), max(tokens_in, 0)) billed_in = (tokens_in - cached) * rate_in + cached * rate_in * CACHE_READ_SHARE diff --git a/tests/harness/test_claude_cost.py b/tests/harness/test_claude_cost.py index eacedac21..d4da24071 100644 --- a/tests/harness/test_claude_cost.py +++ b/tests/harness/test_claude_cost.py @@ -637,7 +637,8 @@ def can_drive(self, model): def test_ci_runs_these_tests(): - """The check-by date is enforced by CI going red, which only happens if CI runs it.""" + """Every PR runs these tests, so the price table's shape and the cost rules are checked on each. + The check-by date itself goes red from the weekly price-check-by workflow, not here.""" workflow = ( Path(__file__).resolve().parents[2] / ".github" / "workflows" / "sdk-smoke.yml" ) diff --git a/tests/harness/test_spend.py b/tests/harness/test_spend.py index eb4f48a8b..c0a66a6f1 100644 --- a/tests/harness/test_spend.py +++ b/tests/harness/test_spend.py @@ -135,38 +135,69 @@ def now(cls, tz=None): @pytest.mark.parametrize("model", ["gemini-3.6-flash", "gemini-3.7-flash", "gemini-3.8-flash"]) -def test_a_price_that_has_run_out_refuses_instead_of_billing_it(monkeypatch, caplog, model): +def test_a_price_past_its_re_check_date_keeps_billing_the_latest_row(monkeypatch, caplog, model): """The Flash models double on 2027-01-01, as Google's Vertex pricing page says (cloud.google.com/model-garden/pricing, Global tier, read 2026-11-10). Past the table's - re-check date they have no price at all, and say so rather than billing a stale one.""" + re-check date a running release keeps the latest row's price and warns, rather than leaving + every turn unpriced.""" import logging from datetime import date from fi.alk.harness.backends import vertex_gemini + monkeypatch.setattr(vertex_gemini, "_STALE_WARNED", set()) + caplog.set_level(logging.WARNING, logger=vertex_gemini.__name__) + + _on(monkeypatch, date(2026, 11, 2)) + assert vertex_gemini.priced(model, 1_000_000, 1_000_000) == 4.5 + _on(monkeypatch, date(2026, 12, 31)) assert vertex_gemini.priced(model, 1_000_000, 1_000_000) == 4.5 assert vertex_gemini.priced(model, 1_000_000, 0, 1_000_000) == pytest.approx(0.075) - _on(monkeypatch, date(2027, 1, 1)) + _on(monkeypatch, date(2027, 3, 31)) assert vertex_gemini.priced(model, 1_000_000, 1_000_000) == 9.0 assert vertex_gemini.priced(model, 1_000_000, 0, 1_000_000) == pytest.approx(0.15) + assert caplog.records == [] _on(monkeypatch, date(2027, 4, 1)) + assert vertex_gemini.priced(model, 1_000_000, 1_000_000) == 9.0 + assert vertex_gemini.priced(model, 1_000_000, 0, 1_000_000) == pytest.approx(0.15) + stale = [r for r in caplog.records if "past its re-check date" in r.getMessage()] + # Once a day per model, not once a turn. + assert len(stale) == 1 and stale[0].levelno == logging.WARNING + assert model in stale[0].getMessage() and vertex_gemini.CHECK_BY in stale[0].getMessage() + + _on(monkeypatch, date(2027, 4, 2)) + vertex_gemini.priced(model, 1_000, 1_000) + assert len([r for r in caplog.records if "past its re-check date" in r.getMessage()]) == 2 + + +def test_a_model_with_no_row_stays_unpriced(monkeypatch, caplog): + """No price is guessed where the table has none: before a model's first row, or for a model + it doesn't list. Those turns are counted in `unpriced_turns`.""" + import logging + from datetime import date + + from fi.alk.harness.backends import vertex_gemini + + monkeypatch.setattr(vertex_gemini, "_STALE_WARNED", set()) caplog.set_level(logging.WARNING, logger=vertex_gemini.__name__) - assert vertex_gemini.priced(model, 1_000_000, 1_000_000) is None - assert f"no current price for {model}" in caplog.text + _on(monkeypatch, date(2026, 11, 1)) + assert vertex_gemini.priced("gemini-3.7-flash", 1_000_000, 1_000_000) is None + assert "no current price for gemini-3.7-flash" in caplog.text + assert "past its re-check date" not in caplog.text + _on(monkeypatch, date(2027, 4, 1)) assert vertex_gemini.priced("a-model-nobody-listed", 1_000, 1_000) is None + assert "past its re-check date" not in caplog.text -def test_every_price_is_re_checked_by_the_check_by_date(): - """Runs on the real date, so the table can't quietly outlive its prices: from 2027-03-01 this - is red on every PR until each model is re-checked against the gateway table and Google.""" +def test_every_model_s_rows_follow_on_and_end_on_the_check_by_date(): + """Runs on every PR: the table's shape, which no date can make red.""" from datetime import date, timedelta from fi.alk.harness.backends.vertex_gemini import CHECK_BY, PRICES_PER_MILLION - assert CHECK_BY == "2027-03-31" for name, rows in PRICES_PER_MILLION.items(): dated = sorted(rows, key=lambda row: row[2]) assert dated[-1][3] == CHECK_BY, f"{name}'s latest row is not re-checked by {CHECK_BY}" @@ -177,12 +208,80 @@ def test_every_price_is_re_checked_by_the_check_by_date(): assert date.fromisoformat(later[2]) == date.fromisoformat(earlier[3]) + timedelta( days=1 ), f"{name}'s rows do not follow on: {earlier} then {later}" - assert (date.fromisoformat(CHECK_BY) - date.today()).days > 30, ( - f"the price table must be re-checked by {CHECK_BY}: compare every row with the gateway's " - "litellm.json and Google's Vertex pricing page, then move CHECK_BY on" + + +# The weekly price-check-by workflow sets this; test_the_weekly_workflow_runs_the_check_by_test +# pins the two ends together. pytest evaluates the skipif string with `os` in scope. +CHECK_BY_ENV = "ALK_PRICE_CHECK_BY" + + +@pytest.mark.skipif( + "not os.environ.get(CHECK_BY_ENV)", + reason="runs from the weekly price-check-by workflow, so the date never blocks a PR", +) +def test_the_price_table_is_re_checked_before_its_check_by_date(): + """Red from a month before CHECK_BY (2027-03-01 for 2027-03-31), on the real UTC day, until + each model is re-checked against the gateway table and Google and CHECK_BY moves on.""" + from datetime import date + + from fi.alk.harness.backends import vertex_gemini + + assert (date.fromisoformat(vertex_gemini.CHECK_BY) - vertex_gemini._today()).days > 30, ( + f"the price table must be re-checked by {vertex_gemini.CHECK_BY}: compare every row with " + "the gateway's litellm.json and Google's Vertex pricing page, then move CHECK_BY on" ) +def test_the_check_by_date_test_goes_red_from_a_month_before(monkeypatch): + """The scheduled test's own boundary, checked on every PR with fixed days.""" + from datetime import date, timedelta + + from fi.alk.harness.backends import vertex_gemini + + check_by = date.fromisoformat(vertex_gemini.CHECK_BY) + _on(monkeypatch, check_by - timedelta(days=31)) + test_the_price_table_is_re_checked_before_its_check_by_date() + _on(monkeypatch, check_by - timedelta(days=30)) + with pytest.raises(AssertionError, match=f"must be re-checked by {check_by.isoformat()}"): + test_the_price_table_is_re_checked_before_its_check_by_date() + + +def test_the_weekly_workflow_runs_the_check_by_test(): + """The scheduled job and the test's skip share one env var and one node id, the job fails on a + skip, and a failure reaches the team: any slip here leaves the reminder green or turns every PR + red.""" + from pathlib import Path + + import yaml + + path = Path(__file__).resolve().parents[2] / ".github" / "workflows" / "price-check-by.yml" + workflow = yaml.safe_load(path.read_text(encoding="utf-8")) + # PyYAML reads the bare key `on` as the boolean True. + triggers = workflow.get("on", workflow.get(True)) + assert triggers["schedule"] == [{"cron": "0 6 * * 1"}] + steps = workflow["jobs"]["check-by"]["steps"] + check = next(step for step in steps if step.get("env", {}).get(CHECK_BY_ENV)) + node = "test_spend.py::" + test_the_price_table_is_re_checked_before_its_check_by_date.__name__ + assert node in check["run"] + # A skip there must fail the job, not pass it. + assert "1 passed" in check["run"] and "pipefail" in check["run"] + assert all("continue-on-error" not in step for step in steps) + assert "continue-on-error" not in workflow["jobs"]["check-by"] + # Only the notify job may write issues, and it installs nothing. + assert workflow["permissions"] == {"contents": "read"} + assert "permissions" not in workflow["jobs"]["check-by"] + notify = workflow["jobs"]["notify"] + assert notify["needs"] == "check-by" + assert notify["if"] == "failure() && github.event_name == 'schedule'" + assert notify["permissions"] == {"issues": "write"} + assert not any("uses" in step or "pip" in step.get("run", "") for step in notify["steps"]) + + marks = test_the_price_table_is_re_checked_before_its_check_by_date.pytestmark + skip = next(mark for mark in marks if mark.name == "skipif") + # The skip reads the same name the workflow sets, and nothing else. + assert skip.args == ("not os.environ.get(CHECK_BY_ENV)",) + + def test_the_rate_is_the_row_whose_dates_hold_today(monkeypatch): """The row is chosen by its dates, never by its place in the table.""" from datetime import date From be64c3529b5f1c8ef574fc237600b087ff0b06e0 Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Thu, 12 Nov 2026 11:04:25 +0530 Subject: [PATCH 4/5] fix(harness): the build's reading adds each turn as the backend reports it This PR now goes into #225's branch, where the reading took the difference of Claude's running totals itself (_own). The Claude backend now reports each turn's own figures, so the reading counted half: a two-turn reading recorded 0.07875 instead of 0.1575. The reading keeps each turn as reported; the two #225 tests that fed running totals feed per-turn figures; _gemini_priced reads current_rates, since the table is dated rows. Refs: TH-9057 Signed-off-by: khushalsonawat --- src/fi/alk/harness/agent_reading.py | 30 ++++------------------------- tests/harness/test_agent_reading.py | 16 ++++++++------- 2 files changed, 13 insertions(+), 33 deletions(-) diff --git a/src/fi/alk/harness/agent_reading.py b/src/fi/alk/harness/agent_reading.py index e8aac99b2..95055fa8e 100644 --- a/src/fi/alk/harness/agent_reading.py +++ b/src/fi/alk/harness/agent_reading.py @@ -13,7 +13,6 @@ import functools import logging import time -from dataclasses import replace from pathlib import Path from typing import Any @@ -66,7 +65,6 @@ # comes by soft + 90 s, which leaves the rest of the margin to the document. _CLOSE_SECONDS = 20.0 -_COUNTS = ("tokens_in", "tokens_out", "tokens_cached") def _soft_limit() -> float: @@ -105,43 +103,24 @@ def __init__( *, name: str, backend: Any, - running_totals: bool, ) -> None: - # Never replayed after a quiet stream: a replayed turn is paid for again, and a new - # session would restart Claude's running totals under the baseline the difference is - # taken from. + # Never replayed after a quiet stream: a replayed turn is paid for again. super().__init__(spec, name=name, backend=backend, idle_retries=0) - self.running_totals = running_totals self.done: list[StageDone] = [] self.in_flight = False self.opened = False - self._last_cost = 0.0 - self._last_counts = dict.fromkeys(_COUNTS, 0) async def __aenter__(self) -> "_ReadingStage": await super().__aenter__() self.opened = True return self - def _own(self, reported: StageDone) -> StageDone: - # Claude reports the session's running total on every result, so adding results as - # reported would count every earlier turn again. - if not self.running_totals: - return reported - cost = reported.cost_usd - if cost is not None: - cost, self._last_cost = max(0.0, cost - self._last_cost), cost - counts: dict[str, int] = {} - for key in _COUNTS: - total = int(getattr(reported, key) or 0) - counts[key] = max(0, total - self._last_counts[key]) - self._last_counts[key] = total - return replace(reported, cost_usd=cost, **counts) - def _events(self, received: Any, turn: Any) -> list[Any]: if not isinstance(received, StageDone): return super()._events(received, turn) - done = self._own(received) + # Each backend reports a turn's own figures (the Claude backend takes the difference of + # the CLI's running totals itself), so a turn is kept as reported. + done = received events = super()._events(done, turn) self.done.append(done) self.in_flight = False @@ -278,7 +257,6 @@ async def _reading( opened.spec, name=opened.name, backend=backend, - running_totals=getattr(backend, "name", "") == "claude", ) stopping = asyncio.Event() # The build's opening, the job's guidance included: a generic job's runtime advice changes diff --git a/tests/harness/test_agent_reading.py b/tests/harness/test_agent_reading.py index 63e9fd4e0..6adbc9132 100644 --- a/tests/harness/test_agent_reading.py +++ b/tests/harness/test_agent_reading.py @@ -1114,8 +1114,8 @@ def test_an_oversized_contract_is_refused(tmp_path, model, size, passes): def test_no_contract_after_the_reminder_is_incomplete(tmp_path, model): _repo(tmp_path) - # Claude's results carry the session's running total. - fake = model(_ends(cost_usd=0.07875), _ends(cost_usd=0.1575)) + # Each turn's own figures, as the Claude backend reports them. + fake = model(_ends(cost_usd=0.07875), _ends(cost_usd=0.07875)) output = tmp_path / "reading" assert _run(tmp_path) == 78 @@ -1243,7 +1243,7 @@ def test_an_error_flag_without_a_status_still_gets_the_reminder(tmp_path, model) _repo(tmp_path) fake = model( _ends(outcome="success", is_error=True, cost_usd=0.1), - _writes(CHAT, cost_usd=0.2), + _writes(CHAT, cost_usd=0.1), ) output = tmp_path / "reading" @@ -1318,9 +1318,9 @@ def _gemini_calls(name: str, args: dict, n: int): def _gemini_priced(calls: int) -> float: # The fake's 1,000 tokens in and 10 out per call, at the table's own rates, kept to the # six places the spend record keeps. - from fi.alk.harness.backends.vertex_gemini import PRICES_PER_MILLION + from fi.alk.harness.backends.vertex_gemini import current_rates - rate_in, rate_out = PRICES_PER_MILLION[_PRICED_GEMINI][:2] + rate_in, rate_out = current_rates(_PRICED_GEMINI) return round(calls * (1000 * rate_in + 10 * rate_out) / 1_000_000, 6) @@ -2041,11 +2041,13 @@ def _stage_entry() -> dict: return entry -def test_claude_running_totals_are_counted_once(tmp_path, model): +def test_claude_turns_are_added_as_the_backend_reports_them(tmp_path, model): + # The Claude backend reports each turn's own figures (it takes the difference of the CLI's + # running totals itself), so the reading adds them as reported and never subtracts again. _repo(tmp_path) model( _ends(cost_usd=0.07875, tokens_in=100_000), - _ends(cost_usd=0.1575, tokens_in=200_000), + _ends(cost_usd=0.07875, tokens_in=100_000), ) output = tmp_path / "reading" From 5e45ac810d94fd06ae014144ed21ea506d1858c2 Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Thu, 12 Nov 2026 11:19:34 +0530 Subject: [PATCH 5/5] test(harness): pin the locked-SDK rerun, the per-count guard, the UTC day and the weekly workflow Refs: TH-9057 Signed-off-by: khushalsonawat --- .github/workflows/price-check-by.yml | 8 +-- tests/harness/test_claude_cost.py | 75 +++++++++++++++++++++++++++- tests/harness/test_spend.py | 56 +++++++++++++++------ 3 files changed, 118 insertions(+), 21 deletions(-) diff --git a/.github/workflows/price-check-by.yml b/.github/workflows/price-check-by.yml index 323949160..22fa11c63 100644 --- a/.github/workflows/price-check-by.yml +++ b/.github/workflows/price-check-by.yml @@ -1,9 +1,6 @@ name: Price check-by -# The Gemini price table carries a re-check date (CHECK_BY in vertex_gemini.py). This goes red -# from a month before it, on its own schedule, so the reminder never blocks a PR. GitHub runs -# schedules on the default branch only, so this checks main's table. A red scheduled run opens or -# updates one issue, so the notice reaches the team and not only whoever last edited the cron. +# Schedules run on the default branch only, so this checks main's table. permissions: contents: read @@ -35,8 +32,7 @@ jobs: python -m pytest -q -rs "tests/harness/test_spend.py::test_the_price_table_is_re_checked_before_its_check_by_date" | tee pytest.out grep -q "1 passed" pytest.out - # Separate from the check, so the token that can write issues never reaches a job that installs - # packages. + # Separate, so the issues token never reaches the job that installs packages. notify: needs: check-by if: failure() && github.event_name == 'schedule' diff --git a/tests/harness/test_claude_cost.py b/tests/harness/test_claude_cost.py index d4da24071..ed7661583 100644 --- a/tests/harness/test_claude_cost.py +++ b/tests/harness/test_claude_cost.py @@ -136,7 +136,7 @@ def test_a_turn_with_input_and_no_output_is_still_a_report(): def test_a_difference_below_the_querys_own_usage_takes_the_totals_whole(): """Totals that restarted and came back above the baseline, or a CLI that reports each query on - its own, would otherwise record less than the query's own usage (Dileep, #274).""" + its own, would otherwise record less than the query's own usage.""" session = _session() _done( session, @@ -156,6 +156,47 @@ def test_a_difference_below_the_querys_own_usage_takes_the_totals_whole(): assert (second.tokens_in, second.tokens_out) == (120_000, 1_200) +def test_the_totals_taken_whole_include_a_workers_tokens(): + """`usage` leaves out a worker model's tokens; the totals taken whole don't, so a fallback to + the query's own usage would drop them.""" + worker = "gemini-3.6-flash" + session = _session() + _done(session, _result({M: _usage(100_000, 1_000)})) + second = _done( + session, + _result( + {M: _usage(120_000, 1_200), worker: _usage(30_000, 300)}, + usage={"input_tokens": 120_000, "output_tokens": 1_200}, + ), + ) + + assert (second.tokens_in, second.tokens_out) == (150_000, 1_500) + + +def test_cache_reads_below_the_querys_own_usage_take_the_totals_whole(): + """Tokens in and out can both clear the query's own usage while the cache reads fall short; the + counts are then compared one by one, not on the output alone.""" + session = _session() + _done(session, _result({M: _usage(100_000, 1_000, cached=50_000)})) + second = _done( + session, + _result( + {M: _usage(130_000, 1_200, cached=60_000)}, + usage={ + "input_tokens": 5_000, + "output_tokens": 100, + "cache_read_input_tokens": 20_000, + }, + ), + ) + + assert (second.tokens_in, second.tokens_out, second.tokens_cached) == ( + 190_000, + 1_200, + 60_000, + ) + + def test_a_lower_count_takes_all_three_counts_whole(): """Totals that went down restarted where the kit couldn't see; none of them is a difference.""" session = _session() @@ -648,3 +689,35 @@ def test_ci_runs_these_tests(): } assert "tests/harness/test_spend.py" in listed assert "tests/harness/test_claude_cost.py" in listed + + +def test_ci_reruns_these_tests_on_the_locked_sdk(): + """The hosted image installs from uv.lock, while the main pytest step gets pip's newest SDK. A + step after it must read the locked version, install exactly that, and then run these tests.""" + import yaml + + path = ( + Path(__file__).resolve().parents[2] / ".github" / "workflows" / "sdk-smoke.yml" + ) + steps = yaml.safe_load(path.read_text(encoding="utf-8"))["jobs"]["clean-install"][ + "steps" + ] + main = next( + index + for index, step in enumerate(steps) + if "pytest" in step.get("run", "") + and "tests/harness/test_spend.py" in step["run"] + ) + reruns = [ + step["run"] + for step in steps[main + 1 :] + if "uv.lock" in step.get("run", "") and "claude-agent-sdk==" in step["run"] + ] + assert len(reruns) == 1 + run = reruns[0] + read = run.index("uv.lock") + install = run.index("pip install") + assert run.index("claude-agent-sdk==") > install + test = run.index("pytest") + assert read < install < test + assert "tests/harness/test_claude_cost.py" in run[test:] diff --git a/tests/harness/test_spend.py b/tests/harness/test_spend.py index c0a66a6f1..f753fd7b4 100644 --- a/tests/harness/test_spend.py +++ b/tests/harness/test_spend.py @@ -114,24 +114,43 @@ def _on(monkeypatch, day): def test_the_price_day_is_the_utc_day(monkeypatch): - """The platform's dated rows switch at midnight UTC, so the kit's do too (Dileep, #274).""" + """The platform's dated rows switch at midnight UTC, so the kit's do too.""" + import os + import time from datetime import datetime, timezone from fi.alk.harness.backends import vertex_gemini - def at(moment): - class _Clock(datetime): - @classmethod - def now(cls, tz=None): - assert tz is timezone.utc - return moment + # In Kolkata 23:30 UTC is already the next day, so a local-time date can't pass for the UTC one. + # tzset() applies the zone to this process; the cleanup below restores it. + previous = os.environ.get("TZ") + monkeypatch.setenv("TZ", "Asia/Kolkata") + time.tzset() - monkeypatch.setattr(vertex_gemini, "datetime", _Clock) + def restore(): + if previous is None: + os.environ.pop("TZ", None) + else: + os.environ["TZ"] = previous + time.tzset() - at(datetime(2026, 12, 31, 23, 30, tzinfo=timezone.utc)) - assert vertex_gemini.current_rates("gemini-3.7-flash") == (0.75, 3.75) - at(datetime(2027, 1, 1, 0, 30, tzinfo=timezone.utc)) - assert vertex_gemini.current_rates("gemini-3.7-flash") == (1.50, 7.50) + try: + + def at(moment): + class _Clock(datetime): + @classmethod + def now(cls, tz=None): + assert tz is timezone.utc + return moment + + monkeypatch.setattr(vertex_gemini, "datetime", _Clock) + + at(datetime(2026, 12, 31, 23, 30, tzinfo=timezone.utc)) + assert vertex_gemini.current_rates("gemini-3.7-flash") == (0.75, 3.75) + at(datetime(2027, 1, 1, 0, 30, tzinfo=timezone.utc)) + assert vertex_gemini.current_rates("gemini-3.7-flash") == (1.50, 7.50) + finally: + restore() @pytest.mark.parametrize("model", ["gemini-3.6-flash", "gemini-3.7-flash", "gemini-3.8-flash"]) @@ -260,6 +279,11 @@ def test_the_weekly_workflow_runs_the_check_by_test(): triggers = workflow.get("on", workflow.get(True)) assert triggers["schedule"] == [{"cron": "0 6 * * 1"}] steps = workflow["jobs"]["check-by"]["steps"] + # The token is read-only, and the checkout leaves no copy of it in the repo's git config. + assert steps[0]["with"]["persist-credentials"] is False + # A condition anywhere here would switch the weekly check off with a green job. + assert "if" not in workflow["jobs"]["check-by"] + assert all("if" not in step for step in steps) check = next(step for step in steps if step.get("env", {}).get(CHECK_BY_ENV)) node = "test_spend.py::" + test_the_price_table_is_re_checked_before_its_check_by_date.__name__ assert node in check["run"] @@ -274,7 +298,11 @@ def test_the_weekly_workflow_runs_the_check_by_test(): assert notify["needs"] == "check-by" assert notify["if"] == "failure() && github.event_name == 'schedule'" assert notify["permissions"] == {"issues": "write"} - assert not any("uses" in step or "pip" in step.get("run", "") for step in notify["steps"]) + assert not any( + step.get("uses", "").startswith("actions/setup-python") + or any(word in step.get("run", "") for word in ("pip install", "uv ", "npm ")) + for step in notify["steps"] + ) marks = test_the_price_table_is_re_checked_before_its_check_by_date.pytestmark skip = next(mark for mark in marks if mark.name == "skipif") @@ -432,7 +460,7 @@ def test_every_harness_stage_feeds_the_one_ledger(): def test_todays_prices_match_the_gateway_table_as_committed(): """Runs in CI, unlike the live comparison below: the gateway's rows at 9ec5f9fb1, committed as - a fixture, so a mistyped price fails here (Dileep, #274).""" + a fixture, so a mistyped price fails here.""" import json from datetime import date from pathlib import Path