#!/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 # How many passes the budget looks back over. One pass is a coin flip — # a `meta` pass reads 100% and a `product` pass reads 0%, and neither # says anything about the balance the budget is about. Three is the # smallest window in which a ratio means something and still moves. TRAILING_PASSES = 3 META_SOFT_PCT = 25 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 mark = "ok " if share <= META_SOFT_PCT else "OVER" print(f"\n meta budget [{mark}] {share:.0f}% over the last " f"{len(window)} pass(es) (soft {META_SOFT_PCT}%, InnerLoop v1.6)") 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. print("\n workplans") closed = [] for wid, title, status, tasks, _kind in plans: done = sum(1 for _, s, _ in tasks if s == "done") if status == "done" and done == len(tasks): closed.append(wid.replace("CB-WP-", "")) continue short = (title or "")[:44] print(f" {wid} {status:<12} {done}/{len(tasks)} done {short}") if closed: print(f" {len(closed)} closed and complete: " + " ".join(closed)) 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)) 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())