tegwick
3f1dbac164
CB-RES-0003 measured 84 turns and $15.33 — the largest mechanical category — spent prefixing commands with `cd` and `export PATH="$HOME/.cargo/bin:$PATH"`. Both causes are now fixed once instead of per-leaf. tools/repo.py resolves the repo root from __file__ and cargo from PATH then the standard rustup locations. Every tool imports ROOT from it, so the repo path is stated once rather than redefined in four files — single source of fact, the rule DFD earned in InnerLoop v1.2. rule-coverage and dep-weight now call enter_root(), which is why their relative paths did not need rewriting one by one. The Makefile derives REPO from MAKEFILE_LIST and resolves CARGO the same way, so `make -C <repo> <target>` works from any directory with no prefix. make env-test is the positive control, and is in `make all`: every tool runs from / with PATH=/usr/bin:/bin. Without it this fix could regress silently and invalidate T05's measurement — the whole point of the control loop. dep-weight's "cargo not on PATH" error is kept rather than deleted. It should now be unreachable, and --self-test asserts cargo_bin() resolves unaided; a control that never fires is cheaper than a regression. loop-lint failed on repo.py on its first run — a reporting tool with a positive control but no --self-test entry point. Second time the gate has caught work from its own pass within the hour. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
686 lines
26 KiB
Python
686 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)
|
|
|
|
from repo import ROOT as REPO # noqa: E402 (single source of fact, T01)
|
|
|
|
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())
|