🤖 forks: usage rendering and ledger reconciliation (block 2.3)

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/<job>.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
 
 

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)."""