test(budget): defend the third field of the stop event — observed (kø-y)

BudgetExceeded carries kind/limit/observed as ONE structured stop event, but only
`observed` was undefended. Measured against the whole suite before writing anything:
four of five raise sites (TokenMeter.charge, tick_round, and BOTH arms of exhausted())
could report any value at all without a single one of 621 tests noticing. Only
PortfolioMeter.check was covered.

What hid it: spikes/_harness.py carries its OWN copy of BudgetExceeded/TokenMeter, so
the spike suite's `observed` assert never touched the shipped module — the production
tick_round had no direct test whatsoever.

exhausted() is the only site that CHOOSES a ledger (the S3.4 pre-call guard), so a
refusal naming portfolio_tokens while reporting the run's own spend would misdirect
every reader of it. Both arms are pinned with observed != limit on purpose: at
exactly-exhausted the two coincide, and a test written there would pass on an
implementation that echoed the cap back as the spend.

No defect in the values themselves (unlike kø-x and kø-p) — the triple was coherent at
all five sites; the gap was purely coverage.

Load-bearing MEASURED: nine mutations, all red — five observed mutations (including the
control) and four echo mutations. 621 -> 623 tests.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LW749xcXQmVEgdipB6KNm4
This commit is contained in:
Kjell Tore Guttormsen 2026-08-03 20:46:36 +02:00
commit 3748011d35
2 changed files with 60 additions and 0 deletions

View file

@ -53,6 +53,21 @@ async def test_cap_crossed_raises_budget_exceeded() -> None:
await mw.process(_Ctx(_resp(60)), _noop) # type: ignore[arg-type]
assert exc.value.kind == "tokens"
assert exc.value.limit == 50
assert exc.value.observed == 60 # the spend that crossed it, not merely that something did
def test_round_cap_crossed_carries_the_round_ledger() -> None:
"""The rounds arm of the same stop event (kø-(y)).
The shipped ``tick_round`` had NO direct test at all only ``spikes/_harness.py``'s separate
copy of ``TokenMeter`` did, and that file is not the one that runs. Its entire error contract
was riding on a module the framework never imports."""
meter = TokenMeter(Budget(max_tokens=1000, max_rounds=2))
assert meter.tick_round() == 1
assert meter.tick_round() == 2 # exactly at the cap is still within it
with pytest.raises(BudgetExceeded) as exc:
meter.tick_round()
assert (exc.value.kind, exc.value.limit, exc.value.observed) == ("rounds", 2, 3)
def test_non_positive_cap_rejected() -> None:
@ -171,6 +186,37 @@ async def test_spend_is_ledgered_even_when_the_per_run_cap_raises() -> None:
assert portfolio.spent == 60 # ...and the global ledger still saw the spend
def test_exhausted_names_one_ledger_in_all_three_fields() -> None:
"""kø-(y): the pre-call guard's refusal must describe ONE ledger consistently.
``exhausted`` is where the S3.4 guard decides what to refuse a call with, and it is the only
raise site that CHOOSES between two ledgers. Naming the binding cap in ``kind`` is already
tested; this pins the other two fields to that same choice a refusal that says
``portfolio_tokens`` while reporting the run's own spend would misdirect every reader of it.
Both arms are built with ``observed != limit`` on purpose. At exactly-exhausted the two
coincide, so a test written at that point cannot tell them apart and would pass on an
implementation that echoed the cap back as the spend. Overrunning first is also the honest
case: a run that crossed its cap and had the error caught is precisely when the next call
must be refused."""
meter = TokenMeter(Budget(max_tokens=10, max_rounds=10))
with pytest.raises(BudgetExceeded):
meter.charge(15) # the overrun is real: the meter now stands at 15 against a cap of 10
own = meter.exhausted()
assert own is not None
assert (own.kind, own.limit, own.observed) == ("tokens", 10, 15)
# The global arm: this run has spent NOTHING — a sibling drained the pass.
portfolio = PortfolioMeter(
PortfolioBudget(max_total_tokens=100, max_tokens_per_run=100), spent=150
)
bound = TokenMeter(Budget(max_tokens=100, max_rounds=10), portfolio=portfolio)
assert bound.tokens == 0
shared = bound.exhausted()
assert shared is not None
assert (shared.kind, shared.limit, shared.observed) == ("portfolio_tokens", 100, 150)
def test_no_word_count_token_proxy_in_src() -> None:
# The meter is fed from real UsageDetails, never a len(text.split()) word-count proxy
# (research 03 Rec 3 — the Fase 1 _word_tokens anti-pattern is retired). NOTE: a bare