coulomb/clay-borg
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
clay-borg/tools/cb-cost.py
tegwick d382fd4555 T04: specs/SessionShape.md — compaction is the lever, not session length 
The task's original premise was wrong and is recorded rather than
deleted. It was written to prescribe one task per session; measurement
says the variable is context, not turn count.

Measured:
  - compaction cut context 27x (542,991 -> 19,974) and cost/turn 3.1x;
    the 202 turns after C1 cost less than half the 136 before it
  - a turn costs $0.010 at 20k context and $0.270 at 540k
  - break-even for a compaction is 2-11 turns, so: compact whenever
    context exceeds ~300k and work remains
  - a fresh session is NOT free -- cold start floors at ~51k and must
    then re-read the artifacts a compact summary already holds (~66k).
    Prefer compaction to continue work; prefer a fresh session when the
    task changes, because then prior context is pure overhead.

cb-cost now emits SH-1..SH-3 so the targets come from the instrument
rather than from analysis, per InnerLoop v1.1. All three are UNMET
(mean context 232,982 vs 200,000; p90 492,042 vs 300,000; batching 7.8%
vs 20%) and are reported unmet rather than retargeted -- retargeting in
the commit that first measures is precisely what T07 exists to prevent.

Eighth error instance found while writing this: CB-WP-0001's claim that
"0 of 330 tool calls were batched" is wrong. 330 was the count of
single-call responses, not the total; 31 responses batched, covering 76
calls. It was carried into this workplan unverified. Trusted-arithmetic
class -- the one the T01 audit flagged as having no executable defence,
confirming that finding within hours of making it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-07-31 09:19:32 +02:00

513 lines
19 KiB
Python
Raw Blame History

#!/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 price_of(prices, model, toks):
"""USD for one response. Returns None when the model is unpriced (CA-05)."""
pr = prices.get(model)
if not pr:
return None
cache = prices["cache"]
unit = pr["input"] / 1e6
return (
toks["input"] * unit
+ toks["output"] * pr["output"] / 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).
tool_calls = sum(
1
for r in rows
for c in (r["message"].get("content") or [])
if c.get("type") == "tool_use"
)
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,
"subagent": "/subagents/" in path,
}
)
return out
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")
for r in responses:
r["cost"] = price_of(prices, r["model"], r["toks"])
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"]
pr = prices[r["model"]]
unit = pr["input"] / 1e6
rates = {
"input": unit,
"output": pr["output"] / 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})"
)
sub = sum(r["cost"] or 0 for r in responses if r["subagent"])
return {
"session_shape": session_shape(responses),
"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."
)
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)
# 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 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("--self-test", action="store_true")
ap.add_argument("--json", action="store_true")
args = ap.parse_args()
if args.self_test:
return self_test()
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())
Reference in a new issue View git blame Copy permalink