coulomb/clay-borg
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
clay-borg/tools/status.py
tegwick 38f237fc5c CB-WP-0009-T01: the meta budget measures the pass it governs 
ADR-0006 D1. cb-cost gains pass_costs, which buckets responses into
workplan windows in a single read of the transcripts — calling collect()
once per boundary would re-read every transcript per window, and status
is supposed to stay cheap enough that nobody replaces it with ls.

make status now reports the share over a trailing three passes with the
per-pass breakdown, keeps the lifetime figure labelled NOT the metric,
and prints the D2 exemption so the next reader does not re-derive the
standoff where the budget blocked its own repair. One pass would be a
coin flip: a meta pass reads 100%, a product pass 0%. Three is the
smallest window where the ratio means something and still moves.

First reading: 36% over the last three passes against 49% lifetime,
with CB-WP-0008 at 0%.

The first run reported $0.00 for every window. Transcript stamps are
UTC Z and git prints a local offset, so the string comparison put every
response before the first boundary. Both the fix and a self-test for it
are in; reverting the conversion turns four checks red.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-01 15:38:35 +02:00

402 lines
15 KiB
Python
Raw Blame History

#!/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)"))
print("\n workplans")
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):
print(f" {wid} {status:<12} {done}/{len(tasks)}")
continue
short = (title or "")[:44]
print(f" {wid} {status:<12} {done}/{len(tasks)} done {short}")
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())
Reference in a new issue View git blame Copy permalink