#!/usr/bin/env python3 """One-shot orientation: what is the state of this repo? CB-WP-0004 T03. CB-RES-0003 measured 49 turns and $6.87 of `grep`/`ls`/`wc` answering exactly this question at the start of a session — which workplan is active, which tasks are open, what is uncommitted, how much has been spent, what is failing. It is deliberately **cheap**: git, the workplan files, the transcripts and the two fast Python gates. It does not build. Anything that needs cargo is named as a command to run, not run here — a status command that takes two minutes gets replaced by `ls` within a day. Stated limit: this reports the *fast* gates only. A green `status` is not a green `make all`, and the output says so rather than letting a reader infer it. Usage: python3 tools/status.py python3 tools/status.py --self-test """ import glob import io import os import re import subprocess import sys from contextlib import redirect_stdout from repo import ROOT, enter_root PIN = "fc76445" # CB-WP-0001 acceptance pin; see specs/CostAccounting.md §7 def git(*args): return subprocess.run(["git", "-C", ROOT, *args], capture_output=True, text=True).stdout.strip() # ------------------------------------------------------------- workplans def parse_workplan(path): """(id, title, status, tasks) where tasks is [(id, status, priority)].""" text = open(path).read() def fm(key): m = re.search(rf'^{key}:\s*"?(.*?)"?\s*$', text, re.M) return m.group(1) if m else None kind = fm("kind") or "untagged" tasks = [] for m in re.finditer(r"```task\n(.*?)```", text, re.S): body = m.group(1) def f(key, b=body): mm = re.search(rf"^{key}:\s*(\S+)\s*$", b, re.M) return mm.group(1) if mm else None tasks.append((f("id"), f("status"), f("priority"))) return fm("id"), fm("title"), fm("status"), tasks, kind def workplans(): out = [] for path in sorted(glob.glob(os.path.join(ROOT, "workplans", "*.md"))): out.append(parse_workplan(path)) return out def next_task(plans): """First todo task of the first non-done workplan, with its heading.""" for wid, _title, status, tasks, _kind in plans: if status == "done": continue for tid, tstatus, prio in tasks: if tstatus == "todo": return wid, tid, prio return None def task_heading(task_id): """The `## Task: ...` heading immediately above this task block.""" for path in sorted(glob.glob(os.path.join(ROOT, "workplans", "*.md"))): text = open(path).read() # The heading is one line and the block follows it immediately. # An `re.S` dot here matched from a much earlier heading and # returned twenty paragraphs of prose as a "heading". m = re.search(rf"## Task: ([^\n]*)\n+```task\n(?:[^\n`]*\n)*?" rf"id: {re.escape(task_id)}\s", text) if m: return m.group(1).strip() return "" # ------------------------------------------------------------- fast gates def run_tool(script, *args): """(exit_code, output) for one of our Python gates.""" r = subprocess.run([sys.executable, os.path.join(ROOT, "tools", script), *args], capture_output=True, text=True, cwd=ROOT) return r.returncode, (r.stdout + r.stderr).strip() def cost_lines(): """(open_spend, pinned_total) — measured, never estimated.""" import importlib.util spec = importlib.util.spec_from_file_location( "cb_cost", os.path.join(ROOT, "tools", "cb-cost.py")) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) rep = mod.collect("-home-worsch-clay-borg", None) open_spend = rep["by_task"].get(mod.OPEN_REMAINDER, 0.0) return open_spend, rep["total"], mod # ------------------------------------------------------- the meta budget # **The ratio and the window are a pair** (InnerLoop v1.7, CB-WP-0019 T05). # One meta pass among `n`, costing what its neighbours cost, reads `1/n`: # 33% over three, 25% over four, 20% over five. So 80/20 — the split the # maintainer set — is *one pass in five at normal cost*, and that is a # five-pass window. The same 20% over three would additionally demand the # meta pass be half-price, which does not make meta work rarer, only # rushed. Changing either number without the other changes the rule. TRAILING_PASSES = 5 META_SOFT_PCT = 20 # What the budget is for, in one line, because for three arguments it had # a threshold and no purpose: most of the spend goes on the task at hand; # some must go on control, review, and improving the process that carries # the work forward. META_PURPOSE = ("most spend on the task at hand; some on control, review " "and improving the process") def meta_phase(root=None): """The declared phase setting from gates.toml, or None. Returns `(pct, reason, review_by)`. A phase setting is allowed to move the line — stage 0 and a stabilisation phase do not deserve the same ratio — but only **declared, argued and expiring**. One with no reason or no expiry is refused rather than honoured, so the line cannot be raised quietly to pass a breach. """ path = os.path.join(root or ROOT, "gates.toml") try: text = open(path).read() except OSError: return None m = re.search(r"^meta_phase\s*=\s*\{(.+?)\}", text, re.M | re.S) if not m: return None body = m.group(1) def field(name): f = re.search(name + r'\s*=\s*"([^"]*)"', body) return f.group(1) if f else None pct = re.search(r"pct\s*=\s*(\d+)", body) reason, review_by = field("reason"), field("review_by") if not pct or not reason or not review_by: raise ValueError( "meta_phase needs pct, reason and review_by — a threshold with " "no argument or no expiry is a dial, not a threshold") return int(pct.group(1)), reason, review_by def workplan_starts(): """[(id, kind, first_commit_iso)] — when each pass began. A pass starts at the commit that *added* its workplan file, which is the only boundary git records and the same one `cb-cost --since` has been windowed on by hand since CB-WP-0004 T05. """ import datetime out = [] for path in sorted(glob.glob(os.path.join(ROOT, "workplans", "*.md"))): wid, _title, _status, _tasks, kind = parse_workplan(path) rel = os.path.relpath(path, ROOT) added = git("log", "--diff-filter=A", "--format=%cI", "--", rel) first = added.splitlines()[-1].strip() if added else "" if not (wid and first): continue # Transcript timestamps are UTC `...Z`; git prints a local offset. # Comparing the two as strings silently buckets everything into # `_before`, which is how this first read reported $0. first = ( datetime.datetime.fromisoformat(first) .astimezone(datetime.timezone.utc) .isoformat() .replace("+00:00", "Z") ) out.append((wid, kind, first)) return sorted(out, key=lambda r: r[2]) def meta_budget(plans): """ADR-0006 D1: report the share for the window the budget governs.""" _, _, mod = cost_lines() starts = workplan_starts() if not starts: print("\n meta budget UNAVAILABLE — no workplan start commits found") return costs = mod.pass_costs("-home-worsch-clay-borg", [(wid, when) for wid, _k, when in starts]) window = starts[-TRAILING_PASSES:] kinds = {wid: kind for wid, kind, _ in starts} total = sum(costs[wid]["cost"] for wid, _k, _w in window) # `mixed` splits evenly; stated rather than hidden. weight = {"product": 0.0, "meta": 1.0, "mixed": 0.5} meta = sum(costs[wid]["cost"] * weight.get(kinds[wid], 0.5) for wid, _k, _w in window) if total <= 0: print("\n meta budget UNAVAILABLE — the trailing window measured $0") return share = 100 * meta / total phase = meta_phase() soft = phase[0] if phase else META_SOFT_PCT mark = "ok " if share <= soft else "OVER" print(f"\n meta budget [{mark}] {share:.0f}% over the last " f"{len(window)} pass(es) (soft {soft}%, InnerLoop v1.7)") print(f" for: {META_PURPOSE}") if phase: print(f" phase setting {phase[0]}% until {phase[2]} — {phase[1]}") print(f" (reverts to {META_SOFT_PCT}% on that date unless re-argued)") for wid, kind, _w in window: c = costs[wid] print(f" {wid} {kind:<8} ${c['cost']:>7,.2f} " f"{c['responses']:>4} response(s)") # History, kept and labelled — the whole point of ADR-0006 D1 is that # this number is not what the target compares against. lifetime = sum(costs[wid]["cost"] for wid, _k, _w in starts) life_meta = sum(costs[wid]["cost"] * weight.get(kind, 0.5) for wid, kind, _ in starts) if lifetime > 0: print(f" history, all {len(starts)} passes: " f"{100 * life_meta / lifetime:.0f}% — NOT the metric " f"(ADR-0006 D1)") print(" NOTE: repairing the instrument that reports a breach is always") print(" in budget (ADR-0006 D2); other above-line meta work needs") print(" `authorized_above_budget:` in the workplan frontmatter.") # ------------------------------------------------------------------ report def report(): plans = workplans() head = git("log", "-1", "--format=%h %s") branch = git("rev-parse", "--abbrev-ref", "HEAD") dirty = [ln for ln in git("status", "--porcelain").splitlines() if ln] print("clay-borg — loop status") print(f" commit {head}") print(f" branch {branch}" + (f" ({len(dirty)} uncommitted)" if dirty else " (clean)")) # The list grows by one line per pass forever, and this report has a # length limit it is meant to keep. Collapse the fully-closed ones to # a single line rather than raising the limit — the precedent is # LOOP-LINT's four loadability breaches, each fixed structurally. render_workplans(plans) report_tail(plans) def render_workplans(plans): """The workplan block. Extracted so its behaviour can be asserted on rather than re-derived in a test (ADR-0010 D2: controls are stated over what the tool does, not over a literal the test wrote).""" print("\n workplans") closed = [] for wid, title, status, tasks, _kind in plans: # `cancel` is a CLOSED outcome, not an open one. Counting only # `done` left CB-WP-0005 reading "5/8" and CB-WP-0007 "2/6" in # every status run since they closed — permanently wrong numbers, # which teach the reader to skip the column. The two figures are # kept separate so a cancellation is still visible rather than # laundered into completion. done = sum(1 for _, s, _ in tasks if s == "done") cancelled = sum(1 for _, s, _ in tasks if s == "cancel") if status == "done" and done + cancelled == len(tasks): closed.append(wid.replace("CB-WP-", "")) continue short = (title or "")[:44] note = f", {cancelled} cancelled" if cancelled else "" print(f" {wid} {status:<12} {done}/{len(tasks)} done{note} {short}") if closed: print(f" {len(closed)} closed and complete: " + " ".join(closed)) def report_tail(plans): """Next task, spend, budgets, fast gates — the rest of the report.""" nxt = next_task(plans) if nxt: wid, tid, prio = nxt print(f"\n next task {tid} ({prio}) — {task_heading(tid)}") else: print("\n next task none — every workplan task is done") # Spend. Measured from the transcripts; CB-01/CB-02 thresholds live in # specs/CostAccounting.md §7. print("\n spend") try: open_spend, total, _mod = cost_lines() flag = "" if open_spend > 22.00: flag = " HARD BREACH — commit or decompose" elif open_spend > 10.00: flag = " soft limit passed" print(f" since last commit ${open_spend:>8,.2f} " f"[soft $10.00 / hard $22.00]{flag}") print(f" measured to date ${total:>8,.2f} (all sessions, this repo)") except Exception as e: print(f" UNAVAILABLE — {e}") # InnerLoop v1.6 — soft 25% meta budget over a trailing pass window. # Reported, never gated. try: meta_budget(plans) except Exception as e: print(f"\n meta budget UNAVAILABLE — {e}") print("\n fast gates") code, out = run_tool("loop-lint.py") last = out.splitlines()[-1] if out else "?" print(f" loop-lint {'ok ' if code == 0 else 'FAIL'} {last}") code, out = run_tool("rule-coverage.py") first = next((ln for ln in out.splitlines() if ln.startswith("AM-1 ")), "?") prov = next((ln for ln in out.splitlines() if ln.startswith("provisional U-item")), None) print(f" coverage {'ok ' if code == 0 else 'warn'} {first.strip()}") if prov: ages = re.findall(r"age=(\d+)d", out) oldest = max((int(a) for a in ages), default=0) print(f" provisional {'ok ' if oldest <= 30 else 'warn'} " f"{prov.strip()}, oldest {oldest}d") print("\n slow gates not run here — `make all` (build, tests, sim, bench)") return 0 # --------------------------------------------------------------- self-test def self_test(): """Each assertion pins a failure this tool must detect. The control that matters is the harness-does-nothing class: an orientation command that silently found no workplans and no tasks would print a confident, empty, wrong picture — which is worse than the greps it replaces. """ results = [] def check(name, ok, detail=""): results.append((name, ok, detail)) # CB-WP-0019 T05: the ratio and the window are a pair. If either moves # without the other, "80/20" stops meaning one pass in five at normal # cost — which is the whole content of the rule. check("the meta ratio matches its window", abs(META_SOFT_PCT - 100 / TRAILING_PASSES) < 1e-9, f"{META_SOFT_PCT}% over {TRAILING_PASSES} passes; one meta pass at " f"parity reads {100 / TRAILING_PASSES:.0f}%") # A workplan whose tasks are done-or-cancelled is CLOSED. Counting # only `done` left two real workplans reading 5/8 and 2/6 forever. def rendered(tasks): out = io.StringIO() saved, sys.stdout = sys.stdout, out try: render_workplans([("CB-WP-0000", "fixture", "done", tasks, "meta")]) finally: sys.stdout = saved return out.getvalue() both = rendered([("T01", "done", "high"), ("T02", "cancel", "low")]) check("a done+cancelled workplan collapses to closed", "closed and complete" in both and "1/2" not in both, both.strip().splitlines()[-1].strip()) open_one = rendered([("T01", "done", "high"), ("T02", "todo", "low")]) check("a workplan with a real todo does NOT collapse", "1/2 done" in open_one, open_one.strip().splitlines()[-1].strip()) check("the budget states what it is for", bool(META_PURPOSE) and "task at hand" in META_PURPOSE) # A phase setting may move the line, but only declared, argued and # expiring. One with no reason or no expiry is refused rather than # honoured — otherwise it is a dial for passing breaches. import tempfile def phase_of(body): with tempfile.TemporaryDirectory() as d: open(os.path.join(d, "gates.toml"), "w").write(body) return meta_phase(d) check("a phase setting with no gates.toml is absent, not an error", phase_of("# nothing here") is None) good = 'meta_phase = { pct = 35, reason = "stage 0 hardening", review_by = "2026-12-31" }' check("a fully declared phase setting is honoured", phase_of(good) == (35, "stage 0 hardening", "2026-12-31")) for bad, why in [ ('meta_phase = { pct = 90, review_by = "2026-12-31" }', "no reason"), ('meta_phase = { pct = 90, reason = "because" }', "no expiry"), ]: try: phase_of(bad) check(f"a phase setting with {why} is refused", False, "it was honoured") except ValueError: check(f"a phase setting with {why} is refused", True) def check(name, ok, detail=""): results.append((name, ok, detail)) sample = ( "---\nid: CB-WP-0009\ntitle: \"Sample\"\nstatus: in_progress\n---\n\n" "## Task: first\n\n```task\nid: CB-WP-0009-T01\nstatus: done\n" "priority: high\n```\n\n" "## Task: second\n\n```task\nid: CB-WP-0009-T02\nstatus: todo\n" "priority: medium\n```\n" ) import tempfile with tempfile.NamedTemporaryFile("w", suffix=".md", delete=False) as fh: fh.write(sample) tmp = fh.name try: wid, title, status, tasks, _kind = parse_workplan(tmp) check("workplan frontmatter parses", (wid, title, status) == ("CB-WP-0009", "Sample", "in_progress"), f"{wid} {status}") check("both task blocks parse with status and priority", tasks == [("CB-WP-0009-T01", "done", "high"), ("CB-WP-0009-T02", "todo", "medium")]) check("next task is the first todo, not the first task", next_task([(wid, title, status, tasks, "meta")]) == ("CB-WP-0009", "CB-WP-0009-T02", "medium")) check("a fully-done workplan yields no next task", next_task([("X", "t", "done", [("X-T01", "done", "high")], "meta")]) is None) finally: os.unlink(tmp) # Positive control: refuse to be confidently empty. plans = workplans() check("finds real workplans in this repo", len(plans) >= 4, f"{len(plans)} workplan(s)") check("finds real tasks in this repo", sum(len(t) for *_, t in plans) >= 20, f"{sum(len(t) for *_, t in plans)} task(s)") check("every real workplan has an id and a status", all(w and s for w, _, s, _, _ in plans)) # InnerLoop v1.5: the meta budget needs every workplan classified, or # the ratio is computed over an unknown denominator. check("every workplan declares kind: product|meta|mixed", all(k in ("product", "meta", "mixed") for *_, k in plans), ", ".join(f"{w}={k}" for w, *_, k in plans)) check("every real task has an id and a status", all(i and s for _, _, _, ts, _ in plans for i, s, _ in ts)) # A heading must be one line. The first version used a dot-all match # and returned the prose of the preceding task as the heading. nxt = next_task(workplans()) if nxt: h = task_heading(nxt[1]) check("task heading is a single short line", "\n" not in h and len(h) < 80, repr(h[:60])) check("task heading resolves to something", bool(h)) # ADR-0006 D1: the budget must measure the window it governs. starts = workplan_starts() check("every workplan has a start commit", len(starts) >= len(plans) - 1, f"{len(starts)} start(s) for {len(plans)} workplan(s)") check("start commits are UTC and comparable to transcript stamps", all(w.endswith("Z") for _i, _k, w in starts), # A local-offset stamp compares as a string against `...Z` and # silently buckets every response before the first boundary, # which is exactly how this reported $0 on its first run. ", ".join(w for _i, _k, w in starts[:1])) check("start commits are in ascending order", [w for _i, _k, w in starts] == sorted(w for _i, _k, w in starts)) buf = io.StringIO() try: with redirect_stdout(buf): meta_budget(plans) text = buf.getvalue() check("budget reports a windowed share", "over the last" in text) check("budget labels the cumulative figure as history", "NOT the metric" in text) check("budget states the instrument-repair exemption", "in budget (ADR-0006 D2)" in text) except Exception as e: check("budget reports a windowed share", False, str(e)) # The report itself must run and produce substance, not a stub. buf = io.StringIO() try: with redirect_stdout(buf): report() text = buf.getvalue() check("report runs end to end", "clay-borg — loop status" in text) check("report states its own limit rather than implying green", "slow gates" in text and "make all" in text) check("report is short enough to read at a glance", len(text.splitlines()) <= 40, f"{len(text.splitlines())} lines") except Exception as e: check("report runs end to end", False, str(e)) print("status self-test (positive control)") ok = True for name, passed, det in results: print(f" [{'ok ' if passed else 'FAIL'}] {name}" + (f" — {det}" if det else "")) ok &= passed return 0 if ok else 1 def main(): enter_root() if "--self-test" in sys.argv: return self_test() return report() if __name__ == "__main__": sys.exit(main())