tegwick
7e21df378a
Some checks failed
ci / check (push) Failing after 4s
Review of where token-priced turns did work a deterministic tool could
do. Method: classify every turn in both transcripts by the tool calls it
made. The classifier is committed in tools/cb-cost.py and emitted by
`make cost-mix`, so the baseline is reproducible and the same command
can later falsify the predictions.
mech environment setup 84 turns $15.33
mech ad-hoc text patching 75 turns $13.86
git 37 turns $13.85
mech hub task status 25 turns $ 7.46
mech orientation / inspect 49 turns $ 6.87
hub other 32 turns $ 6.22
mech ad-hoc transcript 39 turns $ 4.56
mech workplan status edit 21 turns $ 4.06
MECHANICAL (dedup) 290 turns $51.26 = 38% of pass
Largest category is `cd` and `export PATH` -- pure friction, and
dep-weight.py already patched it at the leaf, which is evidence it was
noticed and fixed in the wrong place. Second is heredocs string-patching
markdown, which is also the mechanism behind duplicated-fact drift, the
error class InnerLoop v1.2 names and cannot gate.
Explicitly NOT automated: git (37 turns, $13.85) is mostly commit
message authorship -- the highest-output turns in the corpus and the
project's reasoning record. Automating it would save money and destroy
what makes corrections cheap.
CB-WP-0004 implements five candidates and predicts $33-41 recovery
(25-30%), below the 38% measured share on purpose: some inspection and
patching is genuinely exploratory.
The control loop is the deliverable, not a formality. T05 tests three
things and must report all: did mechanical turns disappear, did they
RELOCATE into prose, and did quality hold. If mechanical turns fall and
prose rises by as much, the saving is zero and that is the result to
publish.
Also a self-indictment worth recording: every hub update_task_status in
this project carried hand-typed token estimates, in a repo whose central
finding is that estimated token counts are worthless. T02 fixes it by
reading measured values from cb-cost.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
685 lines
26 KiB
Python
685 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""M-D2-CST: USD cost of agentic work, attributed to workplan tasks.
|
|
|
|
Normative spec: specs/CostAccounting.md. Decision: decisions/ADR-0003.
|
|
|
|
Reads Claude Code session transcripts, deduplicates by requestId, prices
|
|
each response at its own model's rate and its own cache TTL, and attributes
|
|
it to the task named by the next commit at or after it within the same
|
|
session.
|
|
|
|
Positive control (InnerLoop v1.0 §Step 5, CostAccounting CA-02/CA-14):
|
|
this tool asserts it did the work it reports. `--self-test` exercises four
|
|
assertions against fixtures with known answers; every reporting run checks
|
|
the dedup invariant and reconciles attributed + unattributed + open +
|
|
unpriced against the raw total, aborting rather than printing a number that
|
|
does not add up.
|
|
|
|
Usage:
|
|
python3 tools/cb-cost.py # whole repo, no pin
|
|
python3 tools/cb-cost.py --pin fc76445 # pinned at a commit
|
|
python3 tools/cb-cost.py --by-task # per-task attribution
|
|
python3 tools/cb-cost.py --composition # cost split by component
|
|
python3 tools/cb-cost.py --self-test # positive control
|
|
python3 tools/cb-cost.py --json
|
|
"""
|
|
|
|
import argparse
|
|
import collections
|
|
import datetime
|
|
import glob
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
|
|
try:
|
|
import tomllib
|
|
except ModuleNotFoundError: # pragma: no cover - py<3.11
|
|
print("ERROR: needs Python 3.11+ for tomllib", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
PRICES = os.path.join(REPO, "benchmarks", "baselines", "model-prices.toml")
|
|
COMPONENTS = ("input", "output", "cache_read", "write_5m", "write_1h")
|
|
TASK_RE = re.compile(r"\bT\d\d\b")
|
|
UNATTRIBUTED = "UNATTRIBUTED"
|
|
OPEN_REMAINDER = "OPEN (uncommitted)"
|
|
|
|
|
|
class Abort(Exception):
|
|
"""A positive-control failure. Never degrades to a printed number."""
|
|
|
|
|
|
# ---------------------------------------------------------------- pricing
|
|
|
|
|
|
def load_prices(path=PRICES):
|
|
with open(path, "rb") as fh:
|
|
return tomllib.load(fh)
|
|
|
|
|
|
def components(usage):
|
|
"""Token counts per billable component (CA-04: writes split by TTL)."""
|
|
cc = usage.get("cache_creation") or {}
|
|
return {
|
|
"input": usage.get("input_tokens", 0),
|
|
"output": usage.get("output_tokens", 0),
|
|
"cache_read": usage.get("cache_read_input_tokens", 0),
|
|
"write_5m": cc.get("ephemeral_5m_input_tokens", 0),
|
|
"write_1h": cc.get("ephemeral_1h_input_tokens", 0),
|
|
}
|
|
|
|
|
|
def rates_at(prices, model, when=None):
|
|
"""Input/output rate in force for `model` at ISO instant `when` (CA-16).
|
|
|
|
Promotional rates are data, not comments. A response is priced at the
|
|
promo rate when its timestamp falls on or before `promo_until`.
|
|
"""
|
|
pr = prices.get(model)
|
|
if not pr:
|
|
return None
|
|
if "promo_until" in pr and when:
|
|
until = pr["promo_until"]
|
|
# tomllib returns a datetime.date for a bare TOML date.
|
|
if str(when)[:10] <= str(until)[:10]:
|
|
return pr["promo_input"], pr["promo_output"]
|
|
return pr["input"], pr["output"]
|
|
|
|
|
|
def check_price_sheet_age(prices, today=None):
|
|
"""CA-17: a stale sheet invalidates verdicts, so it fails a command."""
|
|
import datetime as _dt
|
|
|
|
recorded = prices.get("recorded")
|
|
if recorded is None:
|
|
return "price sheet has no `recorded` date"
|
|
max_age = prices.get("max_age_days", 90)
|
|
today = today or _dt.date.today()
|
|
if isinstance(recorded, _dt.datetime):
|
|
recorded = recorded.date()
|
|
age = (today - recorded).days
|
|
if age > max_age:
|
|
return (f"price sheet is {age} days old (max {max_age}); refresh "
|
|
f"benchmarks/baselines/model-prices.toml or new M-D2-CST "
|
|
f"`better` verdicts are invalid")
|
|
return None
|
|
|
|
|
|
def price_of(prices, model, toks, when=None):
|
|
"""USD for one response. Returns None when the model is unpriced (CA-05)."""
|
|
pr = prices.get(model)
|
|
if not pr:
|
|
return None
|
|
rin, rout = rates_at(prices, model, when)
|
|
cache = prices["cache"]
|
|
unit = rin / 1e6
|
|
return (
|
|
toks["input"] * unit
|
|
+ toks["output"] * rout / 1e6
|
|
+ toks["cache_read"] * unit * cache["read"]
|
|
+ toks["write_5m"] * unit * cache["write_5m"]
|
|
+ toks["write_1h"] * unit * cache["write_1h"]
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------- transcripts
|
|
|
|
|
|
def transcript_paths(slug):
|
|
"""Every transcript for the repo, including the subagent tree (CA-06)."""
|
|
base = os.path.expanduser(f"~/.claude/projects/{slug}")
|
|
return sorted(glob.glob(f"{base}/*.jsonl")) + sorted(
|
|
glob.glob(f"{base}/*/subagents/agent-*.jsonl")
|
|
)
|
|
|
|
|
|
def read_responses(path, pin=None):
|
|
"""Deduplicate one transcript by requestId, asserting CA-02."""
|
|
groups = collections.defaultdict(list)
|
|
for line in open(path):
|
|
try:
|
|
d = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if d.get("type") != "assistant":
|
|
continue
|
|
usage = (d.get("message") or {}).get("usage")
|
|
if usage is None:
|
|
continue
|
|
ts = d.get("timestamp") or ""
|
|
if pin and ts > pin:
|
|
continue
|
|
groups[d.get("requestId")].append(d)
|
|
|
|
out = []
|
|
for rid, rows in groups.items():
|
|
# CA-02 positive control. Input-side counters are charged once per
|
|
# response and MUST be identical across the group; a divergence means
|
|
# the format changed underneath us and dedup would mis-bill.
|
|
#
|
|
# output_tokens is different: in streamed transcripts (observed in the
|
|
# subagents/ tree) early lines carry a PARTIAL count and only the last
|
|
# line carries the final total — 5, 5, 195 for one response. Taking
|
|
# the first row silently under-reports output, which is why this is a
|
|
# max() and not a first-wins.
|
|
seen_model = {r["message"].get("model") for r in rows}
|
|
if len(seen_model) != 1:
|
|
raise Abort(
|
|
f"{os.path.basename(path)}: requestId {rid} spans {len(rows)} lines "
|
|
f"with {len(seen_model)} distinct models — CA-02 violated"
|
|
)
|
|
per_row = [components(r["message"]["usage"]) for r in rows]
|
|
toks = dict(per_row[0])
|
|
for field in ("input", "cache_read", "write_5m", "write_1h"):
|
|
distinct = {t[field] for t in per_row}
|
|
if len(distinct) != 1:
|
|
raise Abort(
|
|
f"{os.path.basename(path)}: requestId {rid} has {len(distinct)} "
|
|
f"distinct values for {field} across {len(rows)} lines "
|
|
f"({sorted(distinct)}) — CA-02 violated; input-side counters "
|
|
f"are charged once per response and must not vary"
|
|
)
|
|
toks["output"] = max(t["output"] for t in per_row)
|
|
head = rows[0]
|
|
# Tool calls are spread across the group's lines, so they are counted
|
|
# over the whole group — one response may carry several (SS-05).
|
|
blocks = [c for r in rows for c in (r["message"].get("content") or [])
|
|
if c.get("type") == "tool_use"]
|
|
tool_calls = len(blocks)
|
|
cats = sorted({
|
|
c for c in (classify_tool(b.get("name"), b.get("input") or {})
|
|
for b in blocks) if c
|
|
})
|
|
out.append(
|
|
{
|
|
"request_id": rid,
|
|
"model": head["message"].get("model"),
|
|
"timestamp": head.get("timestamp") or "",
|
|
"session": head.get("sessionId") or os.path.basename(path),
|
|
"toks": toks,
|
|
"tool_calls": tool_calls,
|
|
"categories": cats,
|
|
"subagent": "/subagents/" in path,
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
# CB-RES-0003: which turns are mechanical (a deterministic tool could do
|
|
# them) versus judgment (only an agent can). The baseline for measuring
|
|
# whether automation actually removes turns rather than relocating them.
|
|
def classify_tool(name, inp):
|
|
if name == "mcp__dev-hub__update_task_status":
|
|
return "hub task status"
|
|
if name.startswith("mcp__dev-hub__"):
|
|
return "hub other"
|
|
if name != "Bash":
|
|
return None
|
|
cmd = (inp.get("command") or "").strip()
|
|
if not cmd:
|
|
return None
|
|
if "python3 - <<" in cmd:
|
|
if "status: todo" in cmd or "status: done" in cmd:
|
|
return "workplan status edit"
|
|
if "jsonl" in cmd or "requestId" in cmd or "usage" in cmd:
|
|
return "ad-hoc transcript analysis"
|
|
return "ad-hoc text patching"
|
|
if cmd.startswith(("cd ", "export ")):
|
|
return "environment setup"
|
|
if cmd.split()[0] in ("grep", "ls", "wc", "sed", "head", "tail", "cat", "find"):
|
|
return "orientation / inspect"
|
|
if cmd.startswith("git "):
|
|
return "git"
|
|
if "make " in cmd:
|
|
return "make (gates)"
|
|
return None
|
|
|
|
|
|
# Categories a deterministic tool could plausibly own. Judgment-bearing
|
|
# categories (git commit messages, make gates) are excluded deliberately.
|
|
MECHANICAL = frozenset({
|
|
"environment setup", "ad-hoc text patching", "orientation / inspect",
|
|
"ad-hoc transcript analysis", "hub task status", "workplan status edit",
|
|
})
|
|
|
|
|
|
def session_shape(responses):
|
|
"""SH-1..SH-3 from specs/SessionShape.md."""
|
|
import statistics
|
|
|
|
ctx = sorted(r["toks"]["cache_read"] for r in responses)
|
|
with_tools = [r for r in responses if r["tool_calls"] > 0]
|
|
batched = [r for r in with_tools if r["tool_calls"] > 1]
|
|
calls = sum(r["tool_calls"] for r in with_tools)
|
|
pct = statistics.quantiles(ctx, n=100) if len(ctx) > 1 else [ctx[0]] * 99
|
|
return {
|
|
"SH-1_mean_context": statistics.mean(ctx) if ctx else 0,
|
|
"SH-2_p90_context": pct[89],
|
|
"SH-3_batching_rate": (len(batched) / len(with_tools)) if with_tools else 0.0,
|
|
"p50_context": pct[49],
|
|
"responses_with_tools": len(with_tools),
|
|
"tool_calls": calls,
|
|
"calls_in_batched_turns": calls - (len(with_tools) - len(batched)),
|
|
}
|
|
|
|
|
|
# ------------------------------------------------------------- attribution
|
|
|
|
|
|
def commit_index(pin=None):
|
|
"""(utc_time, subject) for each commit, oldest first (CA-09)."""
|
|
fmt = subprocess.run(
|
|
["git", "-C", REPO, "log", "--format=%cI|%s", "--reverse"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
).stdout.splitlines()
|
|
rows = []
|
|
for line in fmt:
|
|
iso, _, subject = line.partition("|")
|
|
# CA-09: convert, never subtract a fixed offset — this repo has two.
|
|
utc = (
|
|
datetime.datetime.fromisoformat(iso)
|
|
.astimezone(datetime.timezone.utc)
|
|
.isoformat()
|
|
.replace("+00:00", "Z")
|
|
)
|
|
rows.append((utc, subject))
|
|
if pin:
|
|
rows = [r for r in rows if r[0] <= pin]
|
|
return rows
|
|
|
|
|
|
def resolve_pin(ref):
|
|
"""A commit-ish pin becomes the UTC instant of that commit."""
|
|
if not ref:
|
|
return None
|
|
if ref.endswith("Z"):
|
|
return ref
|
|
iso = subprocess.run(
|
|
["git", "-C", REPO, "log", "-1", "--format=%cI", ref],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
).stdout.strip()
|
|
return (
|
|
datetime.datetime.fromisoformat(iso)
|
|
.astimezone(datetime.timezone.utc)
|
|
.isoformat()
|
|
.replace("+00:00", "Z")
|
|
)
|
|
|
|
|
|
def attribute(responses, commits):
|
|
"""Assign each response a bucket per CA-08/CA-10/CA-11.
|
|
|
|
Intervals are ending-at-commit: (prev_commit, this_commit]. Subagent
|
|
responses inherit the interval of their timestamp like any other.
|
|
"""
|
|
label_for = []
|
|
prev = ""
|
|
for ts, subject in commits:
|
|
m = TASK_RE.search(subject)
|
|
label_for.append((prev, ts, m.group(0) if m else UNATTRIBUTED))
|
|
prev = ts
|
|
last = commits[-1][0] if commits else ""
|
|
|
|
for r in responses:
|
|
ts = r["timestamp"]
|
|
if last and ts > last:
|
|
r["task"] = OPEN_REMAINDER # CA-11
|
|
continue
|
|
r["task"] = UNATTRIBUTED
|
|
for lo, hi, label in label_for:
|
|
if lo < ts <= hi:
|
|
r["task"] = label
|
|
break
|
|
return responses
|
|
|
|
|
|
# ----------------------------------------------------------------- report
|
|
|
|
|
|
def collect(slug, pin_ref=None):
|
|
prices = load_prices()
|
|
pin = resolve_pin(pin_ref)
|
|
paths = transcript_paths(slug)
|
|
if not paths:
|
|
raise Abort(f"no transcripts found for {slug}")
|
|
|
|
responses = []
|
|
for p in paths:
|
|
responses.extend(read_responses(p, pin))
|
|
if not responses:
|
|
# Positive control: a run that measured nothing must not report $0.00
|
|
# as though it were an answer.
|
|
raise Abort(f"no responses in {len(paths)} transcript(s) — refusing to report")
|
|
|
|
stale = check_price_sheet_age(prices)
|
|
if stale:
|
|
raise Abort(stale)
|
|
|
|
for r in responses:
|
|
r["cost"] = price_of(prices, r["model"], r["toks"], r["timestamp"])
|
|
|
|
attribute(responses, commit_index(pin))
|
|
|
|
by_task = collections.defaultdict(float)
|
|
by_component = collections.Counter()
|
|
by_component_cost = collections.defaultdict(float)
|
|
by_model = collections.defaultdict(float)
|
|
unpriced = []
|
|
cache = prices["cache"]
|
|
for r in responses:
|
|
if r["cost"] is None:
|
|
unpriced.append(r)
|
|
continue
|
|
by_task[r["task"]] += r["cost"]
|
|
by_model[r["model"]] += r["cost"]
|
|
rin, rout = rates_at(prices, r["model"], r["timestamp"])
|
|
unit = rin / 1e6
|
|
rates = {
|
|
"input": unit,
|
|
"output": rout / 1e6,
|
|
"cache_read": unit * cache["read"],
|
|
"write_5m": unit * cache["write_5m"],
|
|
"write_1h": unit * cache["write_1h"],
|
|
}
|
|
for k, n in r["toks"].items():
|
|
by_component[k] += n
|
|
by_component_cost[k] += n * rates[k]
|
|
|
|
total = sum(by_task.values())
|
|
# CA-14: reconciliation is asserted, not assumed.
|
|
residual = total - sum(by_component_cost.values())
|
|
if abs(residual) > 0.005:
|
|
raise Abort(
|
|
f"reconciliation failed: task total ${total:,.4f} vs component total "
|
|
f"${sum(by_component_cost.values()):,.4f} (residual ${residual:,.4f})"
|
|
)
|
|
|
|
# Tool mix: a turn's whole cost is charged to each category it touched,
|
|
# so columns may overlap and must not be summed as if disjoint.
|
|
mix_turns, mix_cost = collections.Counter(), collections.defaultdict(float)
|
|
for r in responses:
|
|
for c in r.get("categories") or []:
|
|
mix_turns[c] += 1
|
|
mix_cost[c] += r["cost"] or 0.0
|
|
mech = [r for r in responses
|
|
if set(r.get("categories") or []) & MECHANICAL]
|
|
|
|
sub = sum(r["cost"] or 0 for r in responses if r["subagent"])
|
|
return {
|
|
"session_shape": session_shape(responses),
|
|
"tool_mix": {
|
|
"turns": dict(mix_turns),
|
|
"cost": dict(mix_cost),
|
|
"mechanical_turns": len(mech),
|
|
"mechanical_cost": sum(r["cost"] or 0 for r in mech),
|
|
},
|
|
"slug": slug,
|
|
"pin": pin,
|
|
"responses": len(responses),
|
|
"transcripts": len(paths),
|
|
"total": total,
|
|
"subagent_total": sub,
|
|
"main_total": total - sub,
|
|
"by_task": dict(by_task),
|
|
"by_model": dict(by_model),
|
|
"tokens": dict(by_component),
|
|
"cost_by_component": dict(by_component_cost),
|
|
"unpriced": [
|
|
{"model": r["model"], "tokens": r["toks"]} for r in unpriced
|
|
],
|
|
"reconciled": True,
|
|
"residual": residual,
|
|
}
|
|
|
|
|
|
def render(rep, by_task=False, composition=False):
|
|
print(f"M-D2-CST cost report — {rep['slug']}")
|
|
print(f" pin {rep['pin'] or '(none — live file, not reproducible)'}")
|
|
print(f" transcripts {rep['transcripts']} responses {rep['responses']}")
|
|
print(f" main ${rep['main_total']:>10,.2f}")
|
|
print(f" subagent tree ${rep['subagent_total']:>10,.2f}")
|
|
print(f" TOTAL ${rep['total']:>10,.2f}")
|
|
|
|
if composition:
|
|
print("\n composition")
|
|
tot = rep["total"] or 1
|
|
for k in COMPONENTS:
|
|
c = rep["cost_by_component"].get(k, 0.0)
|
|
print(
|
|
f" {k:<12}{rep['tokens'].get(k,0):>14,} tok "
|
|
f"${c:>9,.2f} {100*c/tot:>5.1f}%"
|
|
)
|
|
|
|
if by_task:
|
|
print("\n by task")
|
|
rows = sorted(rep["by_task"].items(), key=lambda kv: -kv[1])
|
|
tot = rep["total"] or 1
|
|
for task, c in rows:
|
|
print(f" {task:<20}${c:>9,.2f} {100*c/tot:>5.1f}%")
|
|
# CA-10: the limit is reported with the number, every time.
|
|
un = rep["by_task"].get(UNATTRIBUTED, 0.0)
|
|
print(
|
|
f"\n NOTE: {100*un/tot:.0f}% of cost is UNATTRIBUTED — commits whose "
|
|
f"subject carries no T## tag.\n"
|
|
f" A per-task table is a view over {100*(1-un/tot):.0f}% of spend."
|
|
)
|
|
|
|
mix = rep["tool_mix"]
|
|
if mix["turns"]:
|
|
print("\n tool mix — a turn's cost is charged to every category it")
|
|
print(" touched, so columns overlap and must not be summed")
|
|
for k in sorted(mix["turns"], key=lambda x: -mix["cost"][x]):
|
|
tag = "mech" if k in MECHANICAL else " "
|
|
print(f" {tag} {k:<28}{mix['turns'][k]:>5} turns "
|
|
f"${mix['cost'][k]:>8,.2f}")
|
|
print(f" MECHANICAL (deduplicated) "
|
|
f"{mix['mechanical_turns']:>5} turns ${mix['mechanical_cost']:>8,.2f}"
|
|
f" = {100*mix['mechanical_cost']/(rep['total'] or 1):.0f}% of pass")
|
|
|
|
sh = rep["session_shape"]
|
|
print("\n session shape (specs/SessionShape.md)")
|
|
print(f" SH-1 mean context {sh['SH-1_mean_context']:>12,.0f} tok "
|
|
f"[{'ok ' if sh['SH-1_mean_context']<=200_000 else 'FAIL'} target 200,000]")
|
|
print(f" SH-2 p90 context {sh['SH-2_p90_context']:>12,.0f} tok "
|
|
f"[{'ok ' if sh['SH-2_p90_context']<=300_000 else 'FAIL'} target 300,000]")
|
|
print(f" SH-3 batching rate {100*sh['SH-3_batching_rate']:>11.1f}% "
|
|
f"[{'ok ' if sh['SH-3_batching_rate']>=0.20 else 'FAIL'} target 20.0%]")
|
|
print(f" {sh['tool_calls']} tool calls in {sh['responses_with_tools']} responses; "
|
|
f"{sh['calls_in_batched_turns']} in batched turns")
|
|
|
|
if rep["unpriced"]:
|
|
print(f"\n UNPRICED ({len(rep['unpriced'])} responses, model not in sheet):")
|
|
for u in rep["unpriced"]:
|
|
print(f" {u['model']} {u['tokens']}")
|
|
|
|
print(f"\n reconciled: ok (residual ${rep['residual']:.6f})")
|
|
|
|
|
|
# -------------------------------------------------------------- self-test
|
|
|
|
|
|
def self_test():
|
|
"""AC-5..AC-8. Each asserts a failure mode is actually detected."""
|
|
prices = load_prices()
|
|
checks = []
|
|
|
|
def check(name, ok, detail=""):
|
|
checks.append((name, ok, detail))
|
|
|
|
# AC-8: per-TTL cache pricing. 5m must not be priced at the 1h rate.
|
|
toks = {"input": 0, "output": 0, "cache_read": 0, "write_5m": 100_000, "write_1h": 0}
|
|
got = price_of(prices, "claude-fable-5", toks)
|
|
want = 100_000 * (10.0 / 1e6) * 1.25
|
|
wrong = 100_000 * (10.0 / 1e6) * 2.0
|
|
check("AC-8 5m cache priced at write_5m", abs(got - want) < 1e-9 and got != wrong,
|
|
f"${got:.4f} (1h would be ${wrong:.4f})")
|
|
|
|
# AC-5: dedup invariant is enforced.
|
|
import tempfile
|
|
with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as fh:
|
|
u1 = {"input_tokens": 1, "output_tokens": 2, "cache_read_input_tokens": 3,
|
|
"cache_creation": {"ephemeral_5m_input_tokens": 0,
|
|
"ephemeral_1h_input_tokens": 0}}
|
|
u2 = dict(u1)
|
|
# Diverging cache_read is a real violation (input-side, charged once).
|
|
u2["cache_read_input_tokens"] = 999
|
|
for u in (u1, u2):
|
|
fh.write(json.dumps({"type": "assistant", "requestId": "r1",
|
|
"timestamp": "2026-01-01T00:00:00Z",
|
|
"message": {"model": "claude-opus-5", "usage": u}}) + "\n")
|
|
bad = fh.name
|
|
try:
|
|
read_responses(bad)
|
|
check("AC-5 dedup violation aborts", False, "no Abort raised")
|
|
except Abort:
|
|
check("AC-5 dedup violation aborts", True)
|
|
finally:
|
|
os.unlink(bad)
|
|
|
|
# AC-9: streamed partial output_tokens must resolve to the final total,
|
|
# not the first line. Regression pin on a real defect: first-wins scored
|
|
# a measured subagent response at 5 output tokens instead of 195.
|
|
with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as fh:
|
|
for out_tok in (5, 5, 195):
|
|
u = {"input_tokens": 2, "output_tokens": out_tok,
|
|
"cache_read_input_tokens": 0,
|
|
"cache_creation": {"ephemeral_5m_input_tokens": 19008,
|
|
"ephemeral_1h_input_tokens": 0}}
|
|
fh.write(json.dumps({"type": "assistant", "requestId": "r2",
|
|
"timestamp": "2026-01-01T00:00:00Z",
|
|
"message": {"model": "claude-fable-5",
|
|
"usage": u}}) + "\n")
|
|
partial = fh.name
|
|
try:
|
|
rows = read_responses(partial)
|
|
got = rows[0]["toks"]["output"] if rows else None
|
|
check("AC-9 streamed partial output resolves to final", got == 195,
|
|
f"got {got}, first-wins would give 5")
|
|
finally:
|
|
os.unlink(partial)
|
|
|
|
# CA-16: a dated promo rate must apply before its expiry and lapse after.
|
|
pr = prices
|
|
before = rates_at(pr, "claude-sonnet-5", "2026-07-31T00:00:00Z")
|
|
after = rates_at(pr, "claude-sonnet-5", "2026-09-01T00:00:00Z")
|
|
check("CA-16 promo rate applies before expiry and lapses after",
|
|
before == (2.0, 10.0) and after == (3.0, 15.0),
|
|
f"{before} -> {after}")
|
|
|
|
# CA-17: staleness must actually trip, or the rule is decorative again.
|
|
import datetime as _dt
|
|
fresh = check_price_sheet_age(pr, _dt.date(2026, 8, 1))
|
|
stale = check_price_sheet_age(pr, _dt.date(2026, 11, 10))
|
|
check("CA-17 staleness detected past max_age_days",
|
|
fresh is None and stale is not None, "fresh ok, 102d trips")
|
|
|
|
# CB-02: thresholds must be ordered, or the budget silently never fires.
|
|
ap_defaults = {"soft": 10.00, "hard": 22.00}
|
|
check("CB-02 budget thresholds ordered and positive",
|
|
0 < ap_defaults["soft"] < ap_defaults["hard"],
|
|
f"soft ${ap_defaults['soft']:.2f} < hard ${ap_defaults['hard']:.2f}")
|
|
|
|
# AC-6: zero responses must not report $0.00 as an answer.
|
|
with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as fh:
|
|
fh.write(json.dumps({"type": "user", "message": {}}) + "\n")
|
|
empty = fh.name
|
|
try:
|
|
got = read_responses(empty)
|
|
check("AC-6 empty transcript yields no responses", got == [], f"{len(got)} rows")
|
|
finally:
|
|
os.unlink(empty)
|
|
|
|
# AC-7: the subagent tree is discovered by the path globs.
|
|
slug = "-home-worsch-clay-borg"
|
|
paths = transcript_paths(slug)
|
|
subs = [p for p in paths if "/subagents/" in p]
|
|
check("AC-7 subagent tree enumerated", len(subs) > 0,
|
|
f"{len(subs)} subagent transcript(s) of {len(paths)} total")
|
|
|
|
print("cb-cost self-test (positive control)")
|
|
ok = True
|
|
for name, passed, detail in checks:
|
|
print(f" [{'ok ' if passed else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
|
|
ok &= passed
|
|
return 0 if ok else 1
|
|
|
|
|
|
def budget(slug, soft, hard):
|
|
"""Live cost budget (specs/CostAccounting.md §7).
|
|
|
|
The per-task figure needs the commit that closes the task, so it can
|
|
only ever be retrospective. What IS observable mid-task is spend since
|
|
the LAST commit — the open remainder — because the transcript is an
|
|
append-live file. That is the number a budget can actually fire on.
|
|
"""
|
|
try:
|
|
rep = collect(slug, None)
|
|
except Abort as e:
|
|
print(f"ABORT — {e}", file=sys.stderr)
|
|
return 1
|
|
open_spend = rep["by_task"].get(OPEN_REMAINDER, 0.0)
|
|
head = subprocess.run(
|
|
["git", "-C", REPO, "log", "-1", "--format=%h %s"],
|
|
capture_output=True, text=True, check=True).stdout.strip()
|
|
|
|
print("cost budget — spend since the last commit")
|
|
print(f" last commit {head}")
|
|
print(f" open spend ${open_spend:,.2f}")
|
|
print(f" soft / hard ${soft:,.2f} / ${hard:,.2f}")
|
|
if open_spend > hard:
|
|
print(f"\n HARD BREACH — ${open_spend:,.2f} > ${hard:,.2f}. Commit what "
|
|
f"works, or stop and decompose. Uncommitted work is also "
|
|
f"unattributable.", file=sys.stderr)
|
|
return 1
|
|
if open_spend > soft:
|
|
print(f"\n soft breach — ${open_spend:,.2f} > ${soft:,.2f}. State progress "
|
|
f"as a percentage and decide: continue, or commit and decompose.")
|
|
return 0
|
|
print("\n within budget")
|
|
return 0
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--slug", default="-home-worsch-clay-borg")
|
|
ap.add_argument("--pin", help="commit-ish or ISO Z instant (CA-07)")
|
|
ap.add_argument("--by-task", action="store_true")
|
|
ap.add_argument("--composition", action="store_true")
|
|
ap.add_argument("--session-shape", action="store_true",
|
|
help="SH-1..SH-3 (always shown in the default report)")
|
|
ap.add_argument("--budget", action="store_true",
|
|
help="CB-01/CB-02: spend since the last commit, live")
|
|
ap.add_argument("--soft", type=float, default=10.00)
|
|
ap.add_argument("--hard", type=float, default=22.00)
|
|
ap.add_argument("--self-test", action="store_true")
|
|
ap.add_argument("--json", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
if args.self_test:
|
|
return self_test()
|
|
|
|
if args.budget:
|
|
return budget(args.slug, args.soft, args.hard)
|
|
|
|
try:
|
|
rep = collect(args.slug, args.pin)
|
|
except Abort as e:
|
|
print(f"ABORT — {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
if args.json:
|
|
print(json.dumps(rep, indent=2))
|
|
else:
|
|
render(rep, by_task=args.by_task, composition=args.composition)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|