From 4b92b51bee6254da7424ae7c8bfc7f6f6780347c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabor=20K=C3=B6rber?= Date: Fri, 16 Oct 2026 17:35:21 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=A4=96=20forks:=20usage=20rendering=20and?= =?UTF-8?q?=20ledger=20reconciliation=20(block=202.3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugins/spire/skills/forks/forks.py | 258 +++++++++++- .../spire/skills/forks/tests/test_forks.py | 382 +++++++++++++++++- 2 files changed, 609 insertions(+), 31 deletions(-) diff --git a/plugins/spire/skills/forks/forks.py b/plugins/spire/skills/forks/forks.py index 64c9570..5836f6a 100644 --- a/plugins/spire/skills/forks/forks.py +++ b/plugins/spire/skills/forks/forks.py @@ -122,7 +122,7 @@ SCRUB_WATCH_PREFIXES = ("CLAUDE_", "ANTHROPIC_") # `where` lists unknown vars # C8: the user settings file. Module-level so tests can point it at a temp file. USER_SETTINGS_PATH: str | None = None -PHASE2_COMMANDS = ("usage",) +PHASE2_COMMANDS = () # commands that tolerate unknown arguments (none: all are implemented) # N3: a fork may only run where, list, usage and wait (`tick` launches pings). CHILD_REFUSED = ("train", "fork", "continue", "rebase", "kill", "expire", "init", "tick") @@ -1636,16 +1636,24 @@ def _rebase_succeeded(reg: dict, entry: dict, old: dict, ss, u, rt, base_commit, def finalize_job(state: str, job: str, *, ss: StreamState | None, hint: str, flags=(), reason=None, - detail=None, exit_code=None, wall_s=None, base_commit=None) -> dict | None: + detail=None, exit_code=None, wall_s=None, base_commit=None, + head_at_end=None) -> dict | None: """Write `results/.json`, the registry update and the ledger line, exactly once, under the lock (C3c). The registry status is written first: it is the once-only guard, so a finaliser killed after it leaves no second finaliser to run (M1); the ledger append also skips a job - that already has a line. Returns the result, or None when the job was already final.""" + that already has a line. Returns the result, or None when the job was already final. + `head_at_end` (HEAD of the job's cwd when the call ended) is recorded on the registry entry + for `usage`'s Commit column; a fork or continue finalised without one (killed, lost) reads + HEAD now.""" with RegistryLock(state): reg = read_registry(state) entry = (reg or {}).get("jobs", {}).get(job) if not entry or entry.get("status") not in RUNNING: return None + if head_at_end is None and entry.get("kind") in ("fork", "continue") and entry.get("cwd"): + head_at_end = _git_head(entry["cwd"]) + if head_at_end: + entry["head_at_end"] = head_at_end result, ledger = _conclude(state, reg, entry, ss, hint, flags, reason, detail, exit_code, wall_s, base_commit, _now()) write_json_atomic(result_file(state, job), result) @@ -1953,7 +1961,12 @@ def build_parser() -> _Parser: r.add_argument("--max-budget-usd", type=_pos_float) launch_overrides(r) - cmd("usage", "Aggregate the ledger (Phase 2).") # arguments arrive with 2.3; extra args tolerated + u = cmd("usage", "Aggregate the ledger: tokens and cost per kind, cold starts, cache-hit ratio.") + ug = u.add_mutually_exclusive_group() + ug.add_argument("--block", help="only this block's calls (a fork and its continues)") + ug.add_argument("--phase", help="only this phase's calls") + u.add_argument("--format", choices=("json", "md"), default="json", + help="json (default) or md: one markdown fragment on stdout instead of JSON") return p @@ -2164,17 +2177,6 @@ def cmd_init(args) -> tuple[dict, int]: "bases": sorted(reg.get("bases", {})), "config": cfg}, EXIT_OK -def _not_implemented(block=None, phase=None): - def f(args): - out = {"error": "not implemented", "command": args.command} - if block: - out["block"] = block - if phase: - out["phase"] = phase - return out, EXIT_USAGE - return f - - # --- launching: train, fork, continue ------------------------------------------------- @@ -3182,12 +3184,13 @@ class Supervisor: with contextlib.suppress(OSError): err.close() wall = _now() - t0 + head_end = _git_head(e["cwd"]) # HEAD when the call ended (`usage` Commit column) base_commit = None if e["kind"] in ("train", "rebase"): # the new base's HEAD at the end of the call - base_commit = _git_head(e["cwd"]) or e.get("head_at_launch") + base_commit = head_end or e.get("head_at_launch") res = finalize_job(self.state, self.job, ss=self.ss, hint=hint, flags=self.flags + flags_extra, reason=reason, detail=detail, exit_code=exit_code, wall_s=wall, - base_commit=base_commit) + base_commit=base_commit, head_at_end=head_end) final = res["status"] if res else (read_registry(self.state)["jobs"][self.job].get("status")) self.write_rt(status=final, ended_at=_iso(), exit_code=exit_code) return {"job": self.job, "status": final}, EXIT_OK @@ -3371,6 +3374,213 @@ def cmd_expire(args) -> tuple[dict, int]: "ok": all(k.get("ok", True) for k in killed)}, EXIT_OK +# --- usage (design 6, block 2.3) ------------------------------------------------------------ + +USAGE_KINDS = ("train", "rebase", "fork", "continue", "ping") +HIT_KINDS = ("fork", "continue", "rebase") # the calls whose cache verdict counts for the ratio +DASH = "—" +USAGE_BLOCK_HEADER = ("| Block | Owner (model id from modelUsage) | Session | Commit | in | out | " + "cache write | cache read | cost Δ |") +USAGE_PHASE_HEADER = "| Kind | Calls | in | out | cache write | cache read | cost |" +SHORT_SESSION = 8 +SHORT_COMMIT = 7 + + +def ledger_from_result(res: dict, entry: dict | None, mtime: float | None) -> dict: + """R1: a ledger line for a result that has none, from the result plus its registry entry + (what `_conclude` would have appended). Marked `reconciled`.""" + e = entry or {} + ended = e.get("ended_at") or (_iso(mtime) if mtime else None) + return { + "job": res["job"], "kind": res.get("kind"), "phase": res.get("phase"), + "block": res.get("block"), "base": res.get("base"), "session": res.get("session"), + "parent_session": res.get("parent_session"), + "started": e.get("started_at") or e.get("launched_at") or e.get("created_at"), + "ended": ended, "wall_s": res.get("wall_s"), "binary": e.get("binary_version"), + "model_requested": res.get("model_requested"), "models_used": res.get("models_used") or [], + "tokens": res.get("tokens") or {k: None for k in TOKEN_KEYS}, + "cost_delta_usd": res.get("cost_delta_usd"), "cost_cumulative_usd": e.get("cost_cum_usd"), + "turns": res.get("turns"), "cache": res.get("cache"), "base_state": res.get("base_state"), + "fingerprint": res.get("fingerprint"), "denials": res.get("denials"), + "status": res.get("status"), "flags": res.get("flags") or [], "reason": res.get("reason"), + "note": res.get("note"), "reconciled": True, + } + + +def reconcile_ledger(state: str) -> list[str]: + """R1: a kill between the registry write and the ledger append leaves a finished job with a + result and no ledger line. Append one for every result that has none, under the registry lock + and idempotently (the ledger dedupes on job id). Returns the job ids appended.""" + rdir = os.path.join(state, "results") + try: + names = sorted(n for n in os.listdir(rdir) if n.endswith(".json")) + except FileNotFoundError: + return [] + have, _ = _ledger_jobs(state) + if all(n[:-5] in have for n in names): + return [] # nothing to do: take no lock + added = [] + with RegistryLock(state): + reg = read_registry(state) or {} + have, _ = _ledger_jobs(state) # re-read under the lock: a finaliser may have run + for n in names: + job = n[:-5] + if job in have: + continue + path = os.path.join(rdir, n) + try: + res = read_json(path) + mtime = os.path.getmtime(path) + except (ForksError, OSError) as ex: + _diag("cannot reconcile %s: %s" % (job, ex)) + continue + if not isinstance(res, dict): + continue + res.setdefault("job", job) + if ledger_append(state, ledger_from_result(res, (reg.get("jobs") or {}).get(job), mtime)): + added.append(job) + if added: + _diag("ledger reconciled: appended %s" % ", ".join(added)) + return added + + +def read_ledger(state: str) -> list[dict]: + try: + data = read_bytes_retry(ledger_path(state)) + except FileNotFoundError: + return [] + out = [] + for ln in data.splitlines(): + try: + obj = json.loads(ln.decode("utf-8", "replace")) + except ValueError: + continue + if isinstance(obj, dict) and obj.get("job"): + out.append(obj) + return sorted(out, key=lambda x: _job_num(x["job"])) + + +def _usage_zero() -> dict: + return {"calls": 0, "input": 0, "output": 0, "cache_creation": 0, "cache_creation_1h": 0, + "cache_creation_5m": 0, "cache_read": 0, "cost_usd": 0.0} + + +def _usage_add(acc: dict, line: dict) -> None: + t = line.get("tokens") or {} + acc["calls"] += 1 + for k in TOKEN_KEYS: + acc[k] += int(t.get(k) or 0) + acc["cost_usd"] = round(acc["cost_usd"] + float(line.get("cost_delta_usd") or 0.0), 6) + + +def usage_aggregate(lines: list[dict]) -> dict: + """Totals per kind (every kind present, zero when unused) and overall, cold starts, and the + cache-hit ratio = hits / (fork + continue + rebase calls that have a verdict).""" + kinds = {k: _usage_zero() for k in USAGE_KINDS} + total = _usage_zero() + hits = verdicts = cold = 0 + for ln in lines: + _usage_add(kinds.setdefault(ln.get("kind") or "?", _usage_zero()), ln) + _usage_add(total, ln) + if "cold_start" in (ln.get("flags") or []): + cold += 1 + if ln.get("kind") in HIT_KINDS and ln.get("cache"): + verdicts += 1 + hits += ln["cache"] == "hit" + return {"kinds": kinds, "total": total, "cold_starts": cold, "cache_hits": hits, + "cache_verdicts": verdicts, + "cache_hit_ratio": round(hits / verdicts, 4) if verdicts else None} + + +def commits_between(cwd: str | None, a: str | None, b: str | None) -> list[str]: + """Short hashes of the commits in `a..b` (oldest first); empty when unknown or on a git failure.""" + if not (cwd and a and b) or a == b: + return [] + try: + r = git(["rev-list", "--reverse", "%s..%s" % (a, b)], cwd=cwd) + except (OSError, subprocess.SubprocessError): + return [] + if r.returncode != 0: + return [] + return [x[:SHORT_COMMIT] for x in r.stdout.decode("utf-8", "replace").split()] + + +def _n(v) -> str: + return DASH if v is None else "{:,}".format(int(v)) + + +def _cost(v) -> str: + return DASH if v is None else "${:.4f}".format(float(v)) + + +def usage_md_block(lines: list[dict], jobs: dict) -> str: + """The block table: one row per call; Commit = the commits from the job's `head_at_launch` to + its `head_at_end` for a sequential job, an em dash for a parallel one (the Athanor commits).""" + rows = [USAGE_BLOCK_HEADER, "|---|---|---|---|---|---|---|---|---|"] + for ln in lines: + e = jobs.get(ln["job"]) or {} + commit = DASH + if e.get("mode") != "parallel": + commit = ", ".join(commits_between(e.get("cwd"), e.get("head_at_launch"), + e.get("head_at_end"))) or DASH + t = ln.get("tokens") or {} + known = ln.get("tokens") is not None and any(t.get(k) is not None for k in TOKEN_KEYS) + sess = str(ln.get("session") or "")[:SHORT_SESSION] or DASH + cells = [str(ln.get("block") if ln.get("block") is not None else DASH), + ", ".join(ln.get("models_used") or []) or DASH, sess, commit] + cells += [_n(t.get(k) if known else None) for k in ("input", "output", "cache_creation", + "cache_read")] + cells.append(_cost(ln.get("cost_delta_usd"))) + rows.append("| " + " | ".join(cells) + " |") + return "\n".join(rows) + "\n" + + +def usage_md_phase(agg: dict) -> str: + """The per-kind totals table with a total row, then cold starts, cache-hit ratio, priced total.""" + rows = [USAGE_PHASE_HEADER, "|---|---|---|---|---|---|---|"] + + def row(name, a): + return "| %s | %s | %s | %s | %s | %s | %s |" % ( + name, a["calls"], _n(a["input"]), _n(a["output"]), _n(a["cache_creation"]), + _n(a["cache_read"]), _cost(a["cost_usd"])) + for k, a in agg["kinds"].items(): + if a["calls"]: + rows.append(row(k, a)) + rows.append(row("total", agg["total"])) + if agg["cache_verdicts"]: + ratio = "%d/%d (%.1f%%)" % (agg["cache_hits"], agg["cache_verdicts"], + 100.0 * agg["cache_hits"] / agg["cache_verdicts"]) + else: + ratio = "n/a (no fork, continue or rebase call with a verdict)" + return "\n".join(rows) + "\n\n" + "\n".join([ + "- Cold starts: %d" % agg["cold_starts"], + "- Cache-hit ratio: %s" % ratio, + "- Priced total: %s" % _cost(agg["total"]["cost_usd"])]) + "\n" + + +def cmd_usage(args) -> tuple[dict | str, int]: + """`usage --run R [--block ID | --phase N] [--format json|md]`. R1 first: reconcile the ledger + against `results/` under the lock. Allowed inside a fork (N3).""" + state, _ = _load_run(args) + touch_orchestrator(state) + added = reconcile_ledger(state) + reg = read_registry(state) or {} + lines = read_ledger(state) + scope = None + if args.block is not None: + scope = {"block": args.block} + lines = [x for x in lines if x.get("block") is not None and str(x["block"]) == str(args.block)] + elif args.phase is not None: + scope = {"phase": _phase(args.phase)} + lines = [x for x in lines if x.get("phase") == scope["phase"]] + if args.format == "md": + if args.block is not None: + return usage_md_block(lines, reg.get("jobs") or {}), EXIT_OK + return usage_md_phase(usage_aggregate(lines)), EXIT_OK + agg = usage_aggregate(lines) + return {"state": state, "scope": scope, "reconciled": added, **agg, "rows": lines}, EXIT_OK + + HANDLERS = { "where": cmd_where, "init": cmd_init, @@ -3383,7 +3593,7 @@ HANDLERS = { "kill": cmd_kill, "_supervise": cmd_supervise, "rebase": cmd_rebase, - "usage": _not_implemented(phase=2), + "usage": cmd_usage, "tick": cmd_tick, } @@ -3412,7 +3622,17 @@ def main(argv=None) -> int: except Exception as e: # never break the stdout invariant traceback.print_exc(file=sys.stderr) out, code = {"error": "internal error", "type": type(e).__name__, "detail": str(e)}, EXIT_USAGE - _emit(out) + if isinstance(out, str): # `usage --format md`: one markdown fragment (plan 4.1 exception) + buf = getattr(sys.stdout, "buffer", None) + if buf is not None: # UTF-8 bytes whatever the console code page (the table has Δ, —) + sys.stdout.flush() + buf.write(out.encode("utf-8")) + buf.flush() + else: + sys.stdout.write(out) + sys.stdout.flush() + else: + _emit(out) return code diff --git a/plugins/spire/skills/forks/tests/test_forks.py b/plugins/spire/skills/forks/tests/test_forks.py index 0637001..7bd70ec 100644 --- a/plugins/spire/skills/forks/tests/test_forks.py +++ b/plugins/spire/skills/forks/tests/test_forks.py @@ -856,17 +856,6 @@ class InitTests(Base): # -------------------------------------------------------------------------------------- class CliTests(Base): - def test_phase2_not_implemented(self): - run_log = os.path.join(self.tmp, "r-run.md") - for argv in (("usage", "--phase", "1"),): # `tick` (2.1) and `rebase` (2.2) are implemented - with self.subTest(cmd=argv[0]): - code, out, raw = call([argv[0], "--run", run_log, *argv[1:]]) - self.assertEqual(code, 2) - self.assert_one_json_line(raw) - self.assertEqual(out["error"], "not implemented") - self.assertEqual(out["phase"], 2) - self.assertFalse(os.path.exists(run_log[:-3] + "-forks")) - def test_commands_need_init(self): run_log = os.path.join(self.tmp, "r-run.md") brief = os.path.join(self.tmp, "b.md") @@ -877,7 +866,8 @@ class CliTests(Base): ("continue", "--job", "j0001", "--brief", brief), ("wait", "--all"), ("list",), ("expire", "--all"), ("kill", "--job", "j0001"), ("_supervise", "--job", "j0001"), ("tick",), - ("rebase", "--base", "B1", "--as", "B2", "--brief", brief)) + ("rebase", "--base", "B1", "--as", "B2", "--brief", brief), + ("usage",), ("usage", "--phase", "1"), ("usage", "--block", "1.1", "--format", "md")) for argv in cases: with self.subTest(cmd=argv[0]): code, out, raw = call([argv[0], "--run", run_log, *argv[1:]]) @@ -1893,6 +1883,337 @@ class TickTests(Base): self.assertEqual(list(self.reg()["jobs"]), ["j0001"]) +def call_raw(argv): + """Like `call`, but for `usage --format md`: stdout is not JSON. Returns (exit code, raw stdout).""" + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + code = forks.main(list(argv)) + return code, out.getvalue() + + +SONNET = "claude-sonnet-4-5" +SESS_TRAIN = "aaaaaaaa-0000-4000-8000-000000000001" +SESS_FORK = "bbbbbbbb-0000-4000-8000-000000000002" +SESS_KILLED = "cccccccc-0000-4000-8000-000000000005" +SESS_PAR = "dddddddd-0000-4000-8000-000000000004" +SESS_PING = "eeeeeeee-0000-4000-8000-000000000006" +BLOCK_HEADER = ("| Block | Owner (model id from modelUsage) | Session | Commit | in | out | cache write " + "| cache read | cost Δ |") + + +class UsageTests(Base): + """`usage` against a fixed ledger fixture in a temp state dir (no `claude` call): the md text + exactly, the json totals, null usage, parallel Commit, R1 reconciliation, N3.""" + + def setUp(self): + super().setUp() + self.cwd = os.path.join(self.tmp, "proj") + os.makedirs(self.cwd) + self.git("init", "-q") + self.sha = [] + for i in range(3): + with open(os.path.join(self.cwd, "f.txt"), "w", newline="\n") as f: + f.write("%d\n" % i) + self.git("add", "f.txt") + self.git("commit", "-q", "-m", "c%d" % (i + 1)) + self.sha.append(self.git("rev-parse", "HEAD")) + os.makedirs(os.path.join(self.tmp, "reports")) + self.run_log = os.path.join(self.tmp, "reports", "2026-10-16-18-t-run.md") + self.state = forks.state_dir_for(self.run_log) + forks.ensure_state(self.state) + reg = forks.new_registry() + reg["config"] = {"launch": {"cwd": self.cwd}, "ttl_s": 3600, "pings": "on", "max_minutes": 90} + s = self.sha + specs = [ # job, kind, block, phase, mode, session, launch, end, models, tokens, cost, cache, flags + ("j0001", "train", None, 1, None, SESS_TRAIN, s[0], s[0], [TEST_MODEL], + (10, 20, 1000, 1000, 0, 0), 0.5, None, []), + ("j0002", "fork", "1.1", 1, "sequential", SESS_FORK, s[0], s[2], [TEST_MODEL], + (2, 1500, 3000, 3000, 0, 1234567), 1.2345, "hit", []), + ("j0003", "continue", "1.1", 1, "sequential", SESS_FORK, s[2], s[2], [TEST_MODEL, SONNET], + (3, 700, 200, 0, 200, 1300000), 0.25, "partial", ["cold_start"]), + ("j0004", "fork", "1.2", 1, "parallel", SESS_PAR, s[0], s[2], [TEST_MODEL], + (1, 100, 0, 0, 0, 50000), 0.1, "hit", []), + ("j0005", "fork", "1.3", 2, "sequential", SESS_KILLED, s[2], None, [], + None, None, None, []), + ("j0006", "ping", None, None, None, SESS_PING, None, None, [TEST_MODEL], + (1, 1, 50, 50, 0, 5000), 0.01, "partial", []), + ] + self.lines = [] + for job, kind, block, phase, mode, sess, head0, head1, models, tok, cost, cache, flags in specs: + reg["jobs"][job] = forks._new_entry(job, kind, base="B1", block=block, phase=phase, mode=mode, + cwd=self.cwd, session=sess, status="ok", + head_at_launch=head0) + if head1: + reg["jobs"][job]["head_at_end"] = head1 + tokens = ({k: None for k in forks.TOKEN_KEYS} if tok is None else + dict(zip(forks.TOKEN_KEYS, tok))) + self.lines.append({"job": job, "kind": kind, "phase": phase, "block": block, "base": "B1", + "session": sess, "models_used": models, "tokens": tokens, + "cost_delta_usd": cost, "cache": cache, "flags": flags, + "status": "killed" if tok is None else "ok"}) + reg["bases"]["B1"] = {"name": "B1", "status": "active", "session": "s-b1"} + with forks.RegistryLock(self.state): + forks.write_registry(self.state, reg) + for ln in self.lines: + forks.ledger_append(self.state, ln) + + def git(self, *args): + r = subprocess.run(["git", "-c", "user.name=t", "-c", "user.email=t@t", *args], cwd=self.cwd, + capture_output=True, timeout=60) + self.assertEqual(r.returncode, 0, r.stderr) + return r.stdout.decode().strip() + + def usage(self, *argv, code=0): + c, out, raw = call(["usage", "--run", self.run_log, *argv]) + self.assert_one_json_line(raw) + self.assertEqual(c, code, out) + return out + + def usage_md(self, *argv, code=0): + c, raw = call_raw(["usage", "--run", self.run_log, *argv, "--format", "md"]) + self.assertEqual(c, code, raw) + return raw + + def ledger_lines(self): + with open(forks.ledger_path(self.state), "rb") as f: + return [json.loads(x) for x in f.read().decode("utf-8").splitlines() if x.strip()] + + # --- md --- + def test_md_block_exact(self): + c2, c3 = self.sha[1][:7], self.sha[2][:7] + self.assertEqual(self.usage_md("--block", "1.1"), "\n".join([ + BLOCK_HEADER, + "|---|---|---|---|---|---|---|---|---|", + "| 1.1 | %s | bbbbbbbb | %s, %s | 2 | 1,500 | 3,000 | 1,234,567 | $1.2345 |" + % (TEST_MODEL, c2, c3), + "| 1.1 | %s, %s | bbbbbbbb | — | 3 | 700 | 200 | 1,300,000 | $0.2500 |" + % (TEST_MODEL, SONNET), + ""])) + + def test_md_block_parallel_commit_is_dash(self): + """A parallel fork shows an em dash even though HEAD moved during it (the Athanor commits).""" + self.assertNotEqual(self.sha[0], self.sha[2]) + self.assertEqual(self.usage_md("--block", "1.2"), "\n".join([ + BLOCK_HEADER, "|---|---|---|---|---|---|---|---|---|", + "| 1.2 | %s | dddddddd | — | 1 | 100 | 0 | 50,000 | $0.1000 |" % TEST_MODEL, ""])) + + def test_md_block_null_usage_renders_dash(self): + self.assertEqual(self.usage_md("--block", "1.3"), "\n".join([ + BLOCK_HEADER, "|---|---|---|---|---|---|---|---|---|", + "| 1.3 | — | cccccccc | — | — | — | — | — | — |", ""])) + + def test_md_block_unknown_is_header_only(self): + self.assertEqual(self.usage_md("--block", "9.9"), + BLOCK_HEADER + "\n|---|---|---|---|---|---|---|---|---|\n") + + def test_md_phase_exact(self): + self.assertEqual(self.usage_md("--phase", "1"), "\n".join([ + "| Kind | Calls | in | out | cache write | cache read | cost |", + "|---|---|---|---|---|---|---|", + "| train | 1 | 10 | 20 | 1,000 | 0 | $0.5000 |", + "| fork | 2 | 3 | 1,600 | 3,000 | 1,284,567 | $1.3345 |", + "| continue | 1 | 3 | 700 | 200 | 1,300,000 | $0.2500 |", + "| total | 4 | 16 | 2,320 | 4,200 | 2,584,567 | $2.0845 |", + "", + "- Cold starts: 1", + "- Cache-hit ratio: 2/3 (66.7%)", + "- Priced total: $2.0845", + ""])) + + def test_md_phase_with_null_usage_and_no_verdicts(self): + self.assertEqual(self.usage_md("--phase", "2"), "\n".join([ + "| Kind | Calls | in | out | cache write | cache read | cost |", + "|---|---|---|---|---|---|---|", + "| fork | 1 | 0 | 0 | 0 | 0 | $0.0000 |", + "| total | 1 | 0 | 0 | 0 | 0 | $0.0000 |", + "", + "- Cold starts: 0", + "- Cache-hit ratio: n/a (no fork, continue or rebase call with a verdict)", + "- Priced total: $0.0000", + ""])) + + def test_md_whole_run_is_a_phase_table_with_the_ping(self): + raw = self.usage_md() + self.assertTrue(raw.startswith("| Kind | Calls |"), raw) + self.assertIn("| ping | 1 | 1 | 1 | 50 | 5,000 | $0.0100 |\n", raw) + self.assertIn("| total | 6 | 17 | 2,321 | 4,250 | 2,589,567 | $2.0945 |\n", raw) + self.assertTrue(raw.endswith("- Priced total: $2.0945\n"), raw) + + def test_md_stdout_is_one_fragment_not_json(self): + raw = self.usage_md("--phase", "1") + self.assertFalse(raw.lstrip().startswith("{"), raw) + with self.assertRaises(ValueError): + json.loads(raw) + self.assertTrue(raw.endswith("\n")) + self.assertNotIn("\n\n\n", raw) + + def test_md_bytes_are_utf8_whatever_the_console(self): + """Delta and em dash are not in cp1252/ascii: the bytes must still be UTF-8.""" + env = child_env() + env["PYTHONIOENCODING"] = "ascii:strict" + r = subprocess.run([sys.executable, FORKS_PY, "usage", "--run", self.run_log, "--block", "1.3", + "--format", "md"], env=env, capture_output=True, timeout=120) + self.assertEqual(r.returncode, 0, r.stderr) + text = r.stdout.decode("utf-8") + self.assertEqual(text.splitlines()[0], BLOCK_HEADER) + self.assertIn("—", text) + self.assertNotIn(b"\r", r.stdout) + + # --- json --- + def test_json_totals_per_kind_and_overall(self): + out = self.usage() + self.assertEqual(out["scope"], None) + self.assertEqual(out["reconciled"], []) + self.assertEqual(list(out["kinds"]), ["train", "rebase", "fork", "continue", "ping"]) + k = out["kinds"] + self.assertEqual(k["rebase"], {"calls": 0, "input": 0, "output": 0, "cache_creation": 0, + "cache_creation_1h": 0, "cache_creation_5m": 0, "cache_read": 0, + "cost_usd": 0.0}) + self.assertEqual({x: k["fork"][x] for x in ("calls", "input", "output", "cache_creation", + "cache_creation_1h", "cache_creation_5m", + "cache_read")}, + {"calls": 3, "input": 3, "output": 1600, "cache_creation": 3000, + "cache_creation_1h": 3000, "cache_creation_5m": 0, "cache_read": 1284567}) + self.assertAlmostEqual(k["fork"]["cost_usd"], 1.3345, places=6) + self.assertEqual((k["continue"]["calls"], k["continue"]["cache_creation_5m"], + k["continue"]["cache_creation_1h"]), (1, 200, 0)) + self.assertEqual((k["ping"]["calls"], k["ping"]["cache_read"]), (1, 5000)) + t = out["total"] + self.assertEqual({x: t[x] for x in ("calls", "input", "output", "cache_creation", + "cache_creation_1h", "cache_creation_5m", "cache_read")}, + {"calls": 6, "input": 17, "output": 2321, "cache_creation": 4250, + "cache_creation_1h": 4050, "cache_creation_5m": 200, "cache_read": 2589567}) + self.assertAlmostEqual(t["cost_usd"], 2.0945, places=6) + self.assertAlmostEqual(t["cost_usd"], sum(x["cost_usd"] for x in k.values()), places=6) + self.assertEqual(out["cold_starts"], 1) + self.assertEqual((out["cache_hits"], out["cache_verdicts"]), (2, 3)) # ping excluded + self.assertAlmostEqual(out["cache_hit_ratio"], 2 / 3, places=3) + self.assertEqual([x["job"] for x in out["rows"]], ["j000%d" % i for i in range(1, 7)]) + + def test_json_scopes(self): + out = self.usage("--phase", "1") + self.assertEqual(out["scope"], {"phase": 1}) + self.assertEqual([x["job"] for x in out["rows"]], ["j0001", "j0002", "j0003", "j0004"]) + self.assertEqual(out["total"]["calls"], 4) + self.assertAlmostEqual(out["total"]["cost_usd"], 2.0845, places=6) + out = self.usage("--block", "1.1") + self.assertEqual(out["scope"], {"block": "1.1"}) + self.assertEqual([x["job"] for x in out["rows"]], ["j0002", "j0003"]) + self.assertEqual(out["kinds"]["fork"]["calls"], 1) + out = self.usage("--block", "1.3") # null usage: counted as a call, adds nothing + self.assertEqual((out["total"]["calls"], out["total"]["input"], out["total"]["cost_usd"]), + (1, 0, 0.0)) + self.assertIsNone(out["cache_hit_ratio"]) + self.assertEqual(self.usage("--block", "nope")["rows"], []) + + def test_format_json_explicit_and_errors_stay_json(self): + self.assertIn("kinds", self.usage("--format", "json")) + for argv in (["--block", "1.1", "--phase", "1"], ["--format", "yaml"]): + with self.subTest(argv=argv): + c, out, raw = call(["usage", "--run", self.run_log, *argv]) + self.assertEqual(c, 2, out) + self.assert_one_json_line(raw) + self.assertEqual(out["error"], "usage") + c, raw = call_raw(["usage", "--run", os.path.join(self.tmp, "none-run.md"), "--format", "md"]) + self.assertEqual(c, 2) + self.assertEqual(json.loads(raw)["error"], "no launch config for this run; call init first") + + def test_allowed_inside_a_fork(self): + os.environ[forks.CHILD_MARKER] = "1" + out = self.usage("--phase", "1") + self.assertEqual(out["total"]["calls"], 4) + self.assertIn("| fork | 2 |", self.usage_md("--phase", "1")) + + # --- R1 --- + def result_for(self, job, **kw): + res = {"job": job, "kind": "fork", "base": "B1", "block": "2.1", "phase": 2, "status": "ok", + "flags": [], "fingerprint": "ok", "session": "ffffffff-0000-4000-8000-000000000007", + "parent_session": "s-b1", "models_used": [TEST_MODEL], + "tokens": dict(zip(forks.TOKEN_KEYS, (4, 50, 600, 600, 0, 7000))), + "cost_delta_usd": 0.75, "cache": "hit", "base_state": "warm", "wall_s": 12.5, "turns": 3, + "denials": 0, "note": None, "raw": "raw/%s.jsonl" % job, "paths": None, "reason": None, + "model_requested": TEST_MODEL} + res.update(kw) + forks.write_json_atomic(forks.result_file(self.state, job), res) + return res + + def test_r1_result_without_ledger_line_gets_exactly_one(self): + # a finished result whose ledger line was lost (killed between registry write and append) + with forks.RegistryLock(self.state): + reg = forks.read_registry(self.state) + reg["jobs"]["j0007"] = forks._new_entry( + "j0007", "fork", base="B1", block="2.1", phase=2, status="ok", cwd=self.cwd, + ended_at="2026-10-16T17:00:00+00:00", cost_cum_usd=3.25, binary_version="2.1.999", + started_at="2026-10-16T16:59:00+00:00", head_at_launch=self.sha[0]) + forks.write_registry(self.state, reg) + self.result_for("j0007") + self.result_for("j0002") # has a line already: must stay at one + n0 = len(self.ledger_lines()) + out = self.usage("--phase", "2") + self.assertEqual(out["reconciled"], ["j0007"]) + lines = [x for x in self.ledger_lines() if x["job"] == "j0007"] + self.assertEqual(len(lines), 1) + self.assertEqual(len(self.ledger_lines()), n0 + 1) + ln = lines[0] + self.assertTrue(ln["reconciled"]) + self.assertEqual((ln["kind"], ln["block"], ln["phase"], ln["status"], ln["cache"], ln["turns"]), + ("fork", "2.1", 2, "ok", "hit", 3)) + self.assertEqual(ln["tokens"]["cache_read"], 7000) + self.assertEqual((ln["binary"], ln["cost_cumulative_usd"], ln["ended"], ln["started"]), + ("2.1.999", 3.25, "2026-10-16T17:00:00+00:00", "2026-10-16T16:59:00+00:00")) + self.assertEqual(ln["models_used"], [TEST_MODEL]) + # it is counted: phase 2 = the killed fork (null) + the reconciled one + self.assertEqual(out["total"]["calls"], 2) + self.assertEqual(out["total"]["cache_read"], 7000) + self.assertAlmostEqual(out["total"]["cost_usd"], 0.75, places=6) + # a second usage (either format) adds nothing + out = self.usage("--phase", "2") + self.assertEqual(out["reconciled"], []) + self.usage_md("--phase", "2") + self.assertEqual(len(self.ledger_lines()), n0 + 1) + self.assertEqual(len([x for x in self.ledger_lines() if x["job"] == "j0007"]), 1) + self.assertEqual(len([x for x in self.ledger_lines() if x["job"] == "j0002"]), 1) + + def test_r1_without_registry_entry_and_with_torn_ledger_tail(self): + self.result_for("j0009", block="3.1", phase=3) # no registry entry at all + with open(forks.ledger_path(self.state), "ab") as f: # a torn last line (no newline) + f.write(b'{"job": "j00') + out = self.usage("--phase", "3") + self.assertEqual(out["reconciled"], ["j0009"]) + self.assertEqual(out["total"]["calls"], 1) + self.assertIsNone(out["rows"][0]["cost_cumulative_usd"]) + self.assertEqual(self.usage("--phase", "3")["reconciled"], []) + + def test_r1_waits_for_the_registry_lock(self): + """With the lock held and a line missing, `usage` blocks until it is released, then + reconciles.""" + self.result_for("j0010", block="4.1", phase=4) + done = [] + lock = forks.RegistryLock(self.state) + lock.acquire() + try: + th = threading.Thread(target=lambda: done.append(self.usage("--phase", "4"))) + th.start() + time.sleep(1.0) + self.assertEqual(done, []) # blocked on the lock + finally: + lock.release() + th.join(30) + self.assertEqual(done[0]["reconciled"], ["j0010"]) + + def test_r1_nothing_missing_takes_no_lock(self): + self.result_for("j0002") # a result that has its ledger line + with forks.RegistryLock(self.state): # held: a pass with nothing to add must not wait + self.assertEqual(forks.reconcile_ledger(self.state), []) + self.assertEqual(len([x for x in self.ledger_lines() if x["job"] == "j0002"]), 1) + + def test_r1_corrupt_result_is_skipped(self): + with open(forks.result_file(self.state, "j0011"), "w") as f: + f.write("{not json") + self.assertEqual(self.usage("--phase", "1")["reconciled"], []) + self.assertNotIn("j0011", [x["job"] for x in self.ledger_lines()]) + + class LaunchTests(LaunchBase): def test_train_fork_continue_end_to_end(self): self.setup_run() @@ -2998,6 +3319,43 @@ class LaunchTests(LaunchBase): self.assertEqual(reg["bases"]["B1"]["status"], "active") self.assertNotIn("B2", reg["bases"]) + def test_head_at_end_and_usage_commit_column(self): + """2.3: the supervisor records `head_at_end`; `usage --block` lists the commits made during a + sequential fork, and the ledger/usage figures agree with the result.""" + self.setup_run() + self.train("B1") + self.finish("j0001") + out = self.fork_marked("turns=3", "sleep=1", extra=("--phase", "1")) + job = out["job"] + self.until_running(job) + before = self.git("rev-parse", "HEAD") + self.commit_file("made.py", "x = 1\n", "commit made during the fork") + after = self.git("rev-parse", "HEAD") + res = self.finish(job) + self.assertEqual(res["status"], "ok", res) + e = self.reg()["jobs"][job] + self.assertEqual(e["head_at_launch"], before) + self.assertEqual(e["head_at_end"], after) + self.assertEqual(self.reg()["jobs"]["j0001"]["head_at_end"], before) # the train too + c, raw = call_raw(["usage", "--run", self.run_log, "--block", "1.1", "--format", "md"]) + self.assertEqual(c, 0, raw) + rows = raw.splitlines() + self.assertEqual(rows[0], BLOCK_HEADER) + self.assertEqual(len(rows), 3, raw) + cells = [x.strip() for x in rows[2].strip("|").split("|")] + self.assertEqual(cells[0], "1.1") + self.assertEqual(cells[1], TEST_MODEL) + self.assertEqual(cells[2], res["session"][:8]) + self.assertEqual(cells[3], after[:7]) + t = res["tokens"] + self.assertEqual(cells[4:8], ["{:,}".format(t[k]) for k in ("input", "output", "cache_creation", + "cache_read")]) + self.assertEqual(cells[8], "${:.4f}".format(res["cost_delta_usd"])) + # the json view agrees with the ledger + c, js, raw = call(["usage", "--run", self.run_log, "--phase", "1"]) + self.assertEqual((c, js["reconciled"], js["total"]["calls"]), (0, [], 1)) + self.assertEqual(js["kinds"]["fork"]["cache_read"], res["tokens"]["cache_read"]) + def tearDownModule(): """No supervisor or stub of this suite may survive it (checked by command line, never by name)."""