tegwick
6281cd546e
49 turns and $6.87 per CB-RES-0003 went to grep/ls/wc answering "what is
the state of this repo". tools/status.py answers it in 22 lines: commit
and working-tree state, every workplan with task counts, the next todo
task, spend since the last commit against CB-01/CB-02, and the three fast
gates.
Two constraints are deliberate. It does not build — a status command that
takes two minutes gets replaced by `ls` within a day. And it states its
own limit in the output ("slow gates not run here"), so a green status
cannot be misread as a green `make all`; the self-test asserts that line
is present, and that the whole report stays under 40 lines.
The positive control is the refusal to be confidently empty: a parser
that found zero workplans and zero tasks would print a clean, wrong
picture, which is worse than the greps it replaces. It also caught a
dot-all regex that returned twenty paragraphs of the preceding task's
prose as the "task heading".
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
279 lines
9.9 KiB
Python
279 lines
9.9 KiB
Python
#!/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
|
|
|
|
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
|
|
|
|
|
|
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 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
|
|
|
|
|
|
# ------------------------------------------------------------------ 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 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}")
|
|
|
|
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 = 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)])
|
|
== ("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")])]) 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))
|
|
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))
|
|
|
|
# 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())
|