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


# ------------------------------------------------------------------ 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.5 — soft 25% meta budget. Reported, never gated.
    try:
        _, _, mod = cost_lines()
        rep = mod.collect("-home-worsch-clay-borg", None)
        detail = rep["by_task_detail"]
        by_kind = {"product": 0.0, "meta": 0.0, "mixed": 0.0}
        attributed = 0.0
        for wid, _t, _s, tasks, kind in plans:
            for tid, _ts, _p in tasks:
                d = detail.get(tid)
                if d and kind in by_kind:
                    by_kind[kind] += d["cost"]
                    attributed += d["cost"]
        if attributed > 0:
            # `mixed` splits evenly; stated rather than hidden.
            meta = by_kind["meta"] + by_kind["mixed"] / 2
            share = 100 * meta / attributed
            mark = "ok  " if share <= 25 else "OVER"
            print(f"\n  meta budget  [{mark}] {share:.0f}% of "
                  f"${attributed:,.2f} attributed  (soft 25%, InnerLoop v1.5)")
            print(f"    product ${by_kind['product']:,.2f}  "
                  f"meta ${by_kind['meta']:,.2f}  "
                  f"mixed ${by_kind['mixed']:,.2f} (split 50/50)")
            print("    NOTE: attributed tasks only — early workplans used bare")
            print("          T## ids that collide across passes and are excluded.")
    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))

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