#!/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"): import shutil as _shutil n = len(doc["missing_verdicts"]) if _shutil.which("claude") and not doc.get("needs_extraction"): # the local tier: plan tokens, no API key — the gate replays # and verifies identically whoever judged import local_judge say(C.dim(f"judging {n} nominee(s) via local claude (plan tokens)…")) try: local_judge.run(repo, base, verdicts, lambda s: say(C.dim(s))) except Exception as e: say(C.yellow(f"local judge failed ({str(e)[:120]}) — reporting what stands")) doc = run_check(repo, base, verdicts, seed) elif os.environ.get("ANTHROPIC_API_KEY"): say(C.dim(f"judging {n} nominee(s) (batched API)…")) 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( "judged leg not run — no local `claude` (plan tokens) and no " "ANTHROPIC_API_KEY (--mechanical makes the skip explicit)" )) 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_upgrade(args): """Install or upgrade the framework in this repo from the bundled bytes: canon (install / Amendment-4 reconcile), skills into .claude/skills, the version pin, and the pending-migration list. Refuses on the default branch (it rewrites docs/) unless --force.""" import upgrade as upgrade_mod repo = find_root(".") branch = git(repo, "branch", "--show-current") default = (git(repo, "symbolic-ref", "refs/remotes/origin/HEAD") or "").split("/")[-1] or "main" if branch in (default, "main", "master") and not args.force: say(C.red(f"refusing on the default branch ({branch}) — upgrade rewrites docs/; branch first (or --force)")) return 2 report, rc = upgrade_mod.run(repo) for line in report: mark = line.split()[0] color = {"STOPPED": C.red, "ERROR": C.red, "STALE": C.red, "MIGRATION": C.yellow, "ACTION": C.yellow, "NOTE": C.yellow}.get(mark, C.dim) say(color(line) if color is not C.dim else C.dim(line)) say(C.dim("review with git diff; nothing was committed")) return rc 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", "upgrade", "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)") pu = sub.add_parser("upgrade", help="install/upgrade the framework: canon + skills + pin + pending migrations") pu.add_argument("--force", action="store_true", help="allow running on the default branch") 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, "upgrade": cmd_upgrade, "witness": cmd_witness, "selftest": cmd_selftest, }[args.cmd](args) sys.exit(rc) if __name__ == "__main__": main()