coulomb/clay-borg
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
clay-borg/tools/status.py
tegwick 5c00c72d3b
Some checks failed
ci / check (push) Has been cancelled
Details
status: a cancelled task is a closed one 
CB-WP-0005 read 5/8 and CB-WP-0007 read 2/6 in every status run since
they closed, because only 'done' counted and their remaining tasks are
'cancel'. Two permanently-wrong numbers teach the reader to skip the
column. Both now collapse into 'closed and complete' -- 18 of them --
while an open workplan with cancellations still shows the count
separately, so a cancellation is visible rather than laundered into
completion.

The workplan block is extracted as render_workplans so the control is
stated over what the tool PRINTS rather than over a literal the test
wrote: a done+cancelled fixture must collapse, and one with a real todo
must not. The first version of that control asserted arithmetic on its
own input, which is the tautological shape ADR-0010 D2 demoted.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-03 18:55:11 +02:00

531 lines
21 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
# **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())
Reference in a new issue View git blame Copy permalink