fix(harness): record each Claude turn once, never the CLI's estimate, and date the price table (TH-9057, TH-9039)

.github/workflows/price-check-by.yml

@@ -0,0 +1,55 @@
+name: Price check-by
+
+# Schedules run on the default branch only, so this checks main's table.
+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, so the issues token never reaches the 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

.github/workflows/sdk-smoke.yml

@@ -88,6 +88,8 @@ jobs:
           tests/harness/test_bundle_author_v2.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_framework_detection.py
@@ -110,3 +112,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

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

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,56 @@ 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}
+            # 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
 
-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)