dds

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.9"
# dependencies = []
# ///
"""dds — the discipline, one command.

`dds` runs the whole thing — witness, check, and the judged leg when
it's needed (batched, ANTHROPIC_API_KEY) — and dumps the issues:
every finding with its remedy, every verdict needing attention,
coverage, everything consciously carried. Exit 0 PASS · 1 FAIL ·
2 INCOMPLETE.

  dds                 the full run + the issues dump. Auto everything:
                      repo root discovered upward, base = merge-base
                      with the default branch, witness regenerated on
                      mechanical platforms, prior verdicts inherited
                      (.dds-verdicts.json / .dds-seed.json — only
                      disturbed nominees re-judge).
  dds --mechanical    skip the judged leg: free, fast, no API call.
                      Judgment reports as skipped; the exit code is
                      the MECHANICAL verdict (hard tier + delta
                      structural) — an honest partial answer, labeled.
  dds ci --base SHA   the machine run: witness (required) → verdict
                      reuse (GitHub env) → batched judge (API key) →
                      check → the same issues dump + full walk into
                      the log and .dds-check-status → dds-check.json →
                      provenance stamp. Nothing guessed; exit = status.

  Options: --mechanical · --base REF · --model M. That is all.

Plumbing (framework development): dds witness [--diff] · dds selftest
"""

import argparse
import json
import os
import subprocess
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)

import check_join  # noqa: E402


# ── color ────────────────────────────────────────────────────────────


class C:
    on = False

    @classmethod
    def _w(cls, code, s):
        return f"\033[{code}m{s}\033[0m" if cls.on else s

    @classmethod
    def red(cls, s):
        return cls._w("31", s)

    @classmethod
    def green(cls, s):
        return cls._w("32", s)

    @classmethod
    def yellow(cls, s):
        return cls._w("33", s)

    @classmethod
    def dim(cls, s):
        return cls._w("2", s)

    @classmethod
    def bold(cls, s):
        return cls._w("1", s)


def say(s=""):
    print(s)


# ── repo / base discovery ────────────────────────────────────────────


def git(repo, *args):
    r = subprocess.run(["git", "-C", repo, *args], capture_output=True, text=True)
    return r.stdout.strip() if r.returncode == 0 else None


def find_root(start):
    d = os.path.abspath(start)
    while True:
        if os.path.exists(os.path.join(d, ".git")):
            return d
        parent = os.path.dirname(d)
        if parent == d:
            return os.path.abspath(start)
        d = parent


def auto_base(repo):
    head = git(repo, "symbolic-ref", "refs/remotes/origin/HEAD")
    candidates = []
    if head:
        candidates.append(head.replace("refs/remotes/", ""))
    candidates += ["origin/main", "origin/master", "main", "master"]
    for ref in candidates:
        mb = git(repo, "merge-base", "HEAD", ref)
        if mb:
            return mb, ref
    return None, None


# ── witness (mechanical platforms) ───────────────────────────────────


def ensure_witness(repo, required=False):
    probe = check_join.Check(repo, None, None, None)
    probe.detect_platforms()
    if not probe.platforms:
        return True, None
    r = subprocess.run(
        [sys.executable, os.path.join(HERE, "witness.py"), "--repo", repo],
        capture_output=True,
        text=True,
    )
    if r.returncode == 0:
        return True, None
    if required:
        return False, (r.stderr or "").strip()[:400]
    if "ModuleNotFoundError" in (r.stderr or ""):
        return True, (
            "witness not regenerated (tree-sitter not installed — "
            "`pip install tree-sitter tree-sitter-elixir`); the check will "
            "report INCOMPLETE if no witness doc exists"
        )
    return True, f"witness generation failed: {(r.stderr or '').strip()[:200]}"


# ── the rendering ────────────────────────────────────────────────────


def tick(ok):
    return C.green("✓") if ok else C.red("✗")


def issues_report(doc, mechanical=False):
    """The one output: every issue with its remedy, judgment, coverage,
    the carried ledger, notes."""
    out = []
    st = doc["status"]
    color = {"PASS": C.green, "FAIL": C.red, "INCOMPLETE": C.yellow}[st]
    plat = " · ".join(p["name"] for p in doc.get("platforms") or [])
    regs = " · ".join(
        r["name"] + (" (sketch)" if r.get("sketch") else "") for r in doc.get("regimes") or []
    )
    out.append(
        f"{C.bold('dds')} · protocol {doc.get('protocol')} · "
        f"base {str(doc.get('base'))[:9]} — {color(C.bold(st))}"
        + (C.yellow(" (mechanical only)") if mechanical else "")
    )
    if plat or regs:
        out.append(C.dim(f"platform {plat or '—'}   profiles {regs or '—'}"))

    # issues, grouped by what kind of fix each needs
    findings = [f for f in doc.get("findings") or [] if not f.get("known")]
    if findings:
        out.append("")
        out.append(C.bold("issues to fix"))
        kind_head = {
            "mechanical": "mechanical — a tool repairs these",
            "referential": "referential — a link, id, or vocabulary to correct",
            "code": "code — the tree needs the change",
            "judgmental": "judgmental — a human decides, via PR",
        }
        for kind in ("mechanical", "referential", "code", "judgmental"):
            items = [f for f in findings if (f.get("fix") or {}).get("kind") == kind]
            if not items:
                continue
            out.append("  " + C.bold(kind_head[kind]))
            for f in items:
                gate = C.red(" ← blocks") if f["tier"] != "JUDGED" else ""
                prov = C.dim(" (this change)") if f["delta"] else C.dim(" (pre-existing)")
                out.append(f"    {C.yellow(f['artifact'])}{gate}{prov}")
                out.append(C.dim(f"      {f['text'][:220]}"))
                out.append(C.dim("      fix: ") + (f.get("fix") or {}).get("hint", ""))

    # judgment needing attention
    verds = doc.get("verdicts") or []
    rest = [v for v in verds if v["verdict"] != "conformant"]
    missing = doc.get("missing_verdicts") or []
    if rest or missing:
        out.append("")
        out.append(C.bold("judgment"))
        for v in rest:
            mark = {"partial": C.yellow, "unwitnessed": C.red}.get(v["verdict"], C.dim)
            out.append(f"  {mark(v['verdict']):<22} {os.path.basename(v['adr'])[:-3]}")
            if v.get("basis"):
                out.append(C.dim(f"    why: {v['basis'][:220]}"))
            gap = next((e for e in v.get("evidence") or [] if e.get("note")), None)
            if v["verdict"] == "partial" and gap:
                out.append(C.dim(f"    gap: {gap['note'][:180]}"))
        skip_note = (
            "skipped (--mechanical)" if mechanical
            else "set ANTHROPIC_API_KEY (or run an agent /dds-check)"
        )
        for m in missing:
            out.append(f"  {C.yellow('unjudged'):<22} {os.path.basename(m)[:-3]}"
                       + C.dim(f"  — {skip_note}"))

    # the coverage walk
    risks = doc.get("risks") or []
    if risks:
        out.append("")
        out.append(C.bold("coverage"))
        vmap = {v["adr"]: v for v in verds}
        noms = {n["adr"]: n for n in doc.get("nominees") or []}
        for r in risks:
            head = f"  {tick(r['covered'])} {r['risk']}"
            if not r["covered"]:
                head += C.red("  GAP — no Accepted covering ADR")
            out.append(head)
            for c in r["adrs"]:
                v = vmap.get(c["adr"])
                n = noms.get(c["adr"]) or {}
                line = f"      {c['stem']} ({c['status']})"
                if n.get("conformance"):
                    line += C.yellow(f" · Conformance: {n['conformance']}")
                if v:
                    mk = {"conformant": C.green, "partial": C.yellow}.get(v["verdict"], C.dim)
                    line += f" — {mk(v['verdict'])}"
                elif c["adr"] in missing:
                    line += C.dim(" — unjudged")
                out.append(line)

    # everything consciously carried or awaiting judgment
    reg = doc.get("register") or {}
    lines = []
    for r in reg.get("risks") or []:
        for res in r.get("residuals") or []:
            if not res["state"].startswith("closed"):
                mark = C.red("open    ") if res["state"].startswith("open") else C.green("accepted")
                lines.append(f"  {mark} {res['id']} " + C.dim(f"[{r['id']}] {res['text'][:110]}"))
    for n in doc.get("nominees") or []:
        if n.get("conformance"):
            lines.append(
                f"  {C.yellow('declared')} {os.path.basename(n['adr'])[:-3]} "
                + C.dim(f"Conformance: {n['conformance']} — designed, not yet realized")
            )
        if "interim" in (n.get("status") or ""):
            lines.append(
                f"  {C.yellow('interim ')} {os.path.basename(n['adr'])[:-3]} "
                + C.dim(n["status"] or "")
            )
    for d in doc.get("divergences") or []:
        mark = C.red("STALE   ") if d.get("stale") else C.dim("diverged")
        lines.append(
            f"  {mark} {d['file']} · {d['slug']} ({d['kind']})"
            + (C.dim(f" — {d['why']}") if d.get("why") else "")
        )
    for x in reg.get("datums") or []:
        if x.get("draft"):
            lines.append(f"  {C.dim('draft   ')} {x['key']}")
        if x.get("holes"):
            lines.append(f"  {C.dim('holes   ')} {x['key']} " + C.dim(", ".join(x["holes"])[:110]))
    for x in (reg.get("activities") or []) + (reg.get("governance") or []):
        if x.get("holes"):
            lines.append(f"  {C.dim('holes   ')} {x['stable_id']} " + C.dim(", ".join(x["holes"])[:110]))
    if lines:
        out.append("")
        out.append(C.bold("carried — consciously, or awaiting judgment"))
        out += lines

    # notes + degradations
    notes = (doc.get("stability_notes") or []) + (doc.get("aging_proposed") or [])
    degraded = doc.get("degraded") or []
    if notes or degraded:
        out.append("")
        out.append(C.bold("notes"))
        for n in notes:
            out.append(C.dim(f"  {n[:220]}"))
        for g in degraded:
            out.append(C.dim(f"  degraded: {g[:220]}"))

    if len(out) <= 2:
        out.append(C.green("clean — nothing to fix, nothing pending"))
    return "\n".join(out)


# ── the run ──────────────────────────────────────────────────────────


EXIT = {"PASS": 0, "FAIL": 1, "INCOMPLETE": 2}


def mechanical_status(doc):
    """The verdict ignoring the judged leg: the hard tier plus any
    unacknowledged mechanical finding — what --mechanical honestly
    answers. Residuals are the only exception channel here too."""
    hard = any(f["tier"] == "HARD" for f in doc.get("findings") or [])
    unacknowledged = any(
        f["tier"] in ("STRUCTURAL", "CUBE") and not f.get("known")
        for f in doc.get("findings") or []
    )
    if doc.get("needs_extraction"):
        return "INCOMPLETE"
    return "FAIL" if (hard or unacknowledged) else "PASS"


def run_check(repo, base, verdicts, seed):
    chk = check_join.Check(
        repo, base, ".dds-check/facts",
        verdicts if verdicts and os.path.exists(verdicts) else None,
        prior_path=seed if seed and os.path.exists(seed) else None,
    )
    return chk.run()


def cmd_run(args):
    repo = find_root(".")
    base = args.base
    if not base:
        base, ref = auto_base(repo)
        if not base:
            say(C.red("cannot infer a base (no origin default branch) — pass --base"))
            return 2
        say(C.dim(f"base: merge-base with {ref} = {base[:9]}"))
    ok, hint = ensure_witness(repo)
    if hint:
        say(C.yellow(f"note: {hint}"))

    verdicts = os.path.join(repo, ".dds-verdicts.json")
    seed = os.path.join(repo, ".dds-seed.json")
    doc = run_check(repo, base, verdicts, seed)

    if not args.mechanical and doc.get("missing_verdicts"):
        if os.environ.get("ANTHROPIC_API_KEY"):
            say(C.dim(f"judging {len(doc['missing_verdicts'])} nominee(s) (batched)…"))
            cmd = [
                sys.executable, os.path.join(HERE, "batch_judge.py"),
                "--repo", repo, "--base", base, "--out", verdicts,
            ]
            if args.model:
                cmd += ["--model", args.model]
            inherit = verdicts if os.path.exists(verdicts) else seed
            if inherit and os.path.exists(inherit):
                cmd += ["--seed", inherit]
            rc = subprocess.call(cmd)
            if rc != 0:
                say(C.yellow(f"judge exited {rc} — reporting what stands"))
            doc = run_check(repo, base, verdicts, seed)
        else:
            say(C.yellow(
                "no ANTHROPIC_API_KEY — judged leg not run "
                "(--mechanical to make this explicit and get the mechanical verdict)"
            ))

    say(issues_report(doc, mechanical=args.mechanical))
    if args.mechanical:
        return EXIT[mechanical_status(doc)]
    return EXIT[doc["status"]]


def cmd_ci(args):
    """The machine run. Its log and status file carry the SAME issues
    dump a human would see — a red CI run explains itself."""
    repo = find_root(".")
    if not args.base:
        say(C.red("dds ci: --base is required (machines do not guess baselines)"))
        return 2

    ok, err = ensure_witness(repo, required=True)
    if not ok:
        say(C.red("ci: witness generation failed — the check would attest silence"))
        say(err or "")
        return 2
    say("ci: witness step done (or docs-side only)")

    verdicts = os.path.join(repo, ".dds-verdicts.json")
    seed = os.path.join(repo, ".dds-seed.json")
    if os.environ.get("GITHUB_REPOSITORY") and os.environ.get("GITHUB_OUTPUT"):
        tree = git(repo, "rev-parse", "HEAD^{tree}")
        rc = subprocess.call(
            [sys.executable, os.path.join(HERE, "reuse_verdicts.py"), tree], cwd=repo
        )
        if rc not in (0,):
            say(f"ci: reuse step exited {rc} — judging fresh")
    else:
        say("ci: no GitHub env — verdict reuse skipped")

    if not os.path.exists(verdicts):
        if os.environ.get("ANTHROPIC_API_KEY"):
            cmd = [
                sys.executable, os.path.join(HERE, "batch_judge.py"),
                "--repo", repo, "--base", args.base, "--out", verdicts,
            ]
            if os.path.exists(seed):
                cmd += ["--seed", seed]
            rc = subprocess.call(cmd)
            if rc != 0:
                say(f"ci: batch judge exited {rc} — the completeness gate decides")
        else:
            say("ci: no ANTHROPIC_API_KEY — judged leg not run (INCOMPLETE if nominees exist)")

    doc = run_check(repo, args.base, verdicts, seed)
    issues_text = issues_report(doc)
    walk_text = check_join.render(doc)
    say(issues_text)
    say()
    say("── full walk " + "─" * 50)
    say(walk_text)
    with open(os.path.join(repo, "dds-check.json"), "w", encoding="utf-8") as f:
        json.dump(doc, f, indent=2)
    first = "FAIL" if doc["status"] in ("FAIL", "INCOMPLETE") else "PASS"
    with open(os.path.join(repo, ".dds-check-status"), "w", encoding="utf-8") as f:
        f.write(first + "\n\n" + issues_text + "\n\n" + walk_text + "\n")

    if os.environ.get("GITHUB_SHA"):
        subprocess.call([sys.executable, os.path.join(HERE, "stamp_artifact.py")], cwd=repo)
    else:
        say("ci: no GITHUB_SHA — provenance stamp skipped")

    say(f"ci: {doc['status']} — dds-check.json + .dds-check-status written")
    return EXIT[doc["status"]]


def cmd_witness(args):
    cmd = [sys.executable, os.path.join(HERE, "witness.py"), "--repo", find_root(".")]
    if args.diff:
        cmd.append("--diff")
    return subprocess.call(cmd)


def cmd_selftest(args):
    fixtures = args.fixtures
    if not fixtures:
        guess = os.path.abspath(os.path.join(HERE, "..", "..", "..", "..", "spec", "fixtures"))
        fixtures = guess if os.path.isdir(guess) else None
    if not fixtures:
        say(C.red("no fixtures dir found — pass one explicitly"))
        return 2
    return check_join.selftest(fixtures)


COMMANDS = ("run", "ci", "witness", "selftest")


def main():
    ap = argparse.ArgumentParser(
        prog="dds",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=__doc__,
    )
    ap.add_argument("--no-color", action="store_true", help=argparse.SUPPRESS)
    sub = ap.add_subparsers(dest="cmd", metavar="")

    pr = sub.add_parser("run", help="the full run + issues dump (default; `dds` alone)")
    pr.add_argument("--mechanical", action="store_true",
                    help="skip the judged leg: free, fast; exit = the mechanical verdict")
    pr.add_argument("--base")
    pr.add_argument("--model")
    pci = sub.add_parser("ci", help="the machine run: witness → reuse → judge → check → artifacts")
    pci.add_argument("--base", help="the merge base (required — machines do not guess)")
    pw = sub.add_parser("witness", help="plumbing: (re)generate the ephemeral witness")
    pw.add_argument("--diff", action="store_true")
    ps = sub.add_parser("selftest", help="plumbing: the fixture conformance corpus")
    ps.add_argument("fixtures", nargs="?")

    argv = sys.argv[1:]
    if not argv or (argv[0] not in COMMANDS and argv[0] not in ("-h", "--help")):
        argv = ["run"] + argv

    args = ap.parse_args(argv)
    C.on = sys.stdout.isatty() and not args.no_color and not os.environ.get("NO_COLOR")
    rc = {
        "run": cmd_run,
        "ci": cmd_ci,
        "witness": cmd_witness,
        "selftest": cmd_selftest,
    }[args.cmd](args)
    sys.exit(rc)


if __name__ == "__main__":
    main()