clay-borg/tools/status.py

#!/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())