#!/usr/bin/env python3 """Mechanical lint for the Task26 candidate-2e0894ea runbook supersession. Checks the two runbooks against every amended-plan step and prohibition that is expressible as a document property, plus binding, supersession, preservation, command-citation, and hygiene rules. Read-only against the plan and the old runbooks. Writes lint-results.json next to itself (0600). """ import hashlib import json import os import re import stat import sys import tempfile from pathlib import Path HERE = Path(__file__).resolve().parent REPO = HERE.parents[3] GP_PATH = HERE / "task26-golden-path-2e0894ea.md" RC_PATH = HERE / "task26-recovery-runbook-2e0894ea.md" PLAN_PATH = REPO / ".omo/plans/dualcoach-production-readiness.md" OLD_GP = REPO / ".omo/evidence/dualcoach-golden-path-contract.md" OLD_RC = REPO / ".omo/evidence/dualcoach-recovery-runbook.md" CANDIDATE = "2e0894eac92bc396cc4723bf1f18ebc653b95018dd41574df435941c235da925" WHEEL = "af4a9d0a1ffffb6eb7551c1d6dc2b32853ca6d024332a4f8f5702bbf992f141b" PLAN_SHA = "7ace03c6dad33d2fc3ef223621cbca68a150fde8429932138e252fb8498ac582" OLD_CANDIDATE = "19ed0e9232e240553b9a96a2e6a33d10f1096be49e5ebafc019287d75984e45c" OLD_GP_SHA = "32a379d855c6e5af978bd9886e3bf49c616c7c20f1c1d5f100c19d8adfc5eb4a" OLD_RC_SHA = "ae2f5f9046c06f8f0b42693be5aa0d9c31cada8024ae1e5ab8ba19a3cf10f6fc" ACTOR = "8527916639" OWNER = "8693203710" gp = GP_PATH.read_text(encoding="utf-8") rc = RC_PATH.read_text(encoding="utf-8") both = gp + "\n" + rc results = [] def flow(text): """Collapse whitespace so line wraps don't split token matches.""" return re.sub(r"\s+", " ", text) gp_f, rc_f, both_f = flow(gp), flow(rc), flow(both) def check(cid, description, ok, detail=""): results.append({ "id": cid, "description": description, "status": "PASS" if ok else "FAIL", "detail": detail, }) return ok def tokens(cid, description, text, words): missing = [w for w in words if w not in text] return check(cid, description, not missing, "missing: %s" % missing if missing else "") def sha256(path): return hashlib.sha256(path.read_bytes()).hexdigest() # --- A. bindings --- for name, text in (("golden-path", gp_f), ("recovery", rc_f)): tokens(f"A-bindings-{name}", f"{name} carries candidate/wheel/plan/actor/owner/service bindings", text, [CANDIDATE, WHEEL, PLAN_SHA, ACTOR, OWNER, "hermes-gateway-dualcoachtest.service"]) # --- B. supersession --- for name, text in (("golden-path", gp_f), ("recovery", rc_f)): tokens(f"B-supersession-{name}", f"{name} declares supersession with old paths, shas, and the 19ed0e candidate", text, ["supersede", "dualcoach-golden-path-contract.md", "dualcoach-recovery-runbook.md", OLD_GP_SHA, OLD_RC_SHA, OLD_CANDIDATE]) tokens("B-preservation-note", "old runbooks declared byte-identical history", both, ["byte-identical"]) # --- C. preservation and plan hash (recomputed) --- check("C-old-golden-preserved", "old golden path sha unchanged", sha256(OLD_GP) == OLD_GP_SHA, sha256(OLD_GP)) check("C-old-recovery-preserved", "old recovery runbook sha unchanged", sha256(OLD_RC) == OLD_RC_SHA, sha256(OLD_RC)) check("C-plan-sha", "amended plan sha matches binding", sha256(PLAN_PATH) == PLAN_SHA, sha256(PLAN_PATH)) # --- D. amendment authorized actions (golden path) --- tokens("D-authorized", "golden path covers every authorized action from the ledger boundary", gp_f, ["archive-first", "pre-reset", "rebind", "one new", "invite", "one private-DM Start", "synthetic lifecycle", "one delivery", "explicitly sends", "disable", "cleanup reset"]) # --- E. amendment forbidden list (must be stated as forbidden somewhere) --- tokens("E-forbidden", "runbooks restate every forbidden action from the ledger boundary", both_f, ["archive restore", "prepopulation", "direct durable-state mutation", "raw Telegram listener", "update-cursor", "service replay", "forced card publication", "recovery shortcut", "real-customer activation", "historical evidence retargeting", "unrelated profile change", "commit", "push", "release", "register", "tag"]) # --- F. all-new IDs and states --- tokens("F-new-ids", "golden path pins all-new ID requirements and forbidden reused values", gp_f, ["customer_key", "cb_", "rc1_", "Onboarding revision", "checkin_revision", "draft token", "task22_dm_rehearsal", "task26_synthetic_rehearsal", "task26_same_actor_rehearsal", "cb_2NQV5sbkN-M6awycJH7X5g", "cb_6S6RABpDgZ165V02A7qEzw", "cb_rmDnfrqA6gkmjoQEdwxu0g", "wizard_995f04a3b8bc256fa13ff407", "3f44a18ea620d963"]) tokens("F-states", "golden path pins the full state sequences", gp_f, ["PREPARED", "REGISTERING", "AWAITING_CONSENT", "AWAITING_ACTIVATION", "ACTIVE", "CANCELLED", "EXPIRED", "FAILED", "collecting", "customer_attestation", "reconciling", "owner_review", "finalizing", "ready", "safety_hold", "generation_pending", "generating", "draft_created", "sent_audited", "enabled: false", "enabled: true"]) # --- G. golden path step coverage (Spec 2 + amendment lifecycle) --- tokens("G-steps", "golden path covers each lifecycle step surface", gp_f, ["Fail-closed preflight", "prepare_rehearsal_customer_invite", "^/start(?:@bot)?", "consent card", "Q1 through Q22", "attestation", "topic `59`", "owner DM", "readiness_cli", "customer_admin", "activate", "12-question check-in", "Q7", "exactly-once", "idempotency key", "sent_audited", "disable", "cleanup reset", "handset", "observer", "watch-ready", "loaded-byte proof", "direct_url.json"]) check("G-ordering", "observer subscription section precedes invite/action section", gp.find("Observer subscriptions") < gp.find("Exactly one invite"), "") # --- H. prohibition context: dangerous tokens only near prohibition wording --- PROHIBITION_HINT = re.compile(r"never|no |prohibit|forbid|don't|do not|void|abort|stop", re.IGNORECASE) for token in ("getUpdates", "drop_pending_updates", "`--all`", "sleep", " rm ", "rm`"): bad = [] for m in re.finditer(re.escape(token), both_f): window = both_f[max(0, m.start() - 160): m.end() + 60] if not PROHIBITION_HINT.search(window): bad.append(window.strip()[:90]) check(f"H-prohibition-context-{token.strip()}", f"'{token.strip()}' appears only in prohibition context", not bad, "; ".join(bad[:3])) # --- I. recovery runbook: procedures and per-procedure fields (Spec 3) --- proc_heads = re.findall(r"^### (R\w+)\. ", rc, flags=re.MULTILINE) check("I-procedure-count", "recovery runbook has at least 12 procedures", len(proc_heads) >= 12, str(proc_heads)) blocks = re.split(r"^### ", rc, flags=re.MULTILINE)[1:] field_fail = [] for block in blocks: title = block.splitlines()[0] for field in ("- Trigger", "- Expected", "- Rollback", "- Evidence"): if field not in block: field_fail.append(f"{title}: {field}") if ("- Action" not in block) and ("- Detection" not in block): field_fail.append(f"{title}: - Action/- Detection") check("I-procedure-fields", "every procedure carries trigger, action/detection, expected state, rollback, evidence", not field_fail, "; ".join(field_fail[:5])) tokens("I-procedure-topics", "recovery covers every Spec 3 scenario family", rc_f, ["Invite expired", "Claim collision", "Start payload", "consent", "revision conflict", "Safety hold", "Owner rejection", "failed callback", "Stale, duplicate", "Provider auth", "Delivery failure", "between approval and send", "Restart=always", "gateway.lock"]) # --- J. non-existent commands declared --- tokens("J-nonexistent-commands", "recovery declares the old planned dualcoach_admin interfaces non-existent", rc_f, ["process-state", "generation reconcile", "callback inspect", "card rebuild", "updates reconcile", "outbox reconcile", "customer disable", "do not exist"]) # --- K. exact supported command citations --- tokens("K-commands", "runbooks cite the exact supported command surfaces", both_f, ["systemctl --user start hermes-gateway-dualcoachtest.service", "systemctl --user stop hermes-gateway-dualcoachtest.service", "systemctl --user restart hermes-gateway-dualcoachtest.service", "journalctl --user -u hermes-gateway-dualcoachtest.service", "dualcoach_admin provider-auth check --json", "checkin_cli.readiness_cli", "checkin_cli.customer_admin", "nutrition_onboarding_cli", "--no-index --force-reinstall", "verify_candidate.py", "flock -n", "expire_unbound", "dry-run", "verify", "execute", "--contract", "--permission", "--candidate", "--wheel-sha256", "--plan-sha256", "--approval", "--run-id"]) bad_admin = [s for s in re.findall(r"dualcoach_admin\s+([a-z-]+)", both) if s not in ("provider-auth",)] check("K-no-invented-admin", "no dualcoach_admin subcommand cited besides provider-auth", not bad_admin, str(bad_admin)) bad_gw = [s for s in re.findall(r"gateway (status|start|stop|restart|install|uninstall|setup|list|run|enroll|migrate-legacy)", both) if s not in ("status", "run")] check("K-gateway-cli-scope", "hermes gateway CLI cited only for read-only status (plus unit ExecStart run)", not bad_gw, str(bad_gw)) # --- L. open blockers gated fail-closed --- tokens("L-blockers", "golden path marks every open preflight blocker with owner and closure evidence", gp_f, ["OPEN (B1)", "OPEN (B2)", "OPEN (B3)", "OPEN (B5)", "st_01a0054d", "st_01a00547", "4091167"]) # --- M. reset + cleanup (Task 25 / amendment) --- tokens("M-reset", "archive-first pre/post reset with no-restore rule", gp_f, ["Archive-first pre-reset", "Archive-first cleanup reset", "No restore", "rehearsal-reset-archives", "dry-run", "verify", "execute"]) # --- N. other profiles --- for name, text in (("golden-path", gp_f), ("recovery", rc_f)): tokens(f"N-other-profiles-{name}", f"{name} pins other-profile non-touch", text, ["physique-coach", "quarantine", "e04e3a0a13a73bd7cf2259898e3c9198e5860a4a468a5342c7b0b537a4a19e1e", "9ec0b0f3251f944d9d89a4b6cc3e5a45e15025b52d3a5c0d6a60ea04fa1dc031"]) # --- O. secrets and redaction --- tokens("O-redaction", "secrets and redaction rules present", both_f, ["0600", "edact", "bot token", "plaintext", "customer answer text"]) # --- P. evidence receipts --- tokens("P-receipts", "evidence receipt inventories present", gp_f, ["preflight-receipt", "pre-reset-receipt", "deployment-rebind-receipt", "invite-preparation-receipt", "claim-receipt", "owner-review-receipt", "readiness-receipt", "activation-checklist-receipt", "checkin-receipt", "generation-receipt", "send-receipt", "delivery-receipt", "disable-receipt", "cleanup-reset-receipt"]) tokens("P-recovery-receipts", "recovery requires a receipt per action", rc_f, ["0600 receipt"]) # --- Q. rollback/abort matrix --- check("Q-abort-matrix", "recovery carries a rollback/abort matrix with all lifecycle stages", "Rollback and abort matrix" in rc and rc.count("| Post-") >= 4 and "| Preflight" in rc, "") # --- R. invariants expressible in docs --- tokens("R-invariants", "exactly-once/idempotency/append-only/no-silent-failure invariants stated", both_f, ["exactly-once", "idempotency", "ppend-only", "No silent failure", "lease", "consumed-update"]) # --- S. hygiene: no em/en dashes, no fixed sleeps --- check("S-no-emdash", "no em or en dashes", ("—" not in both) and ("–" not in both), "") check("S-no-sleep-command", "no fixed-sleep command usage", not re.search(r"`[^`]*\bsleep\b", both), "") # --- T. file modes --- for name, path in (("golden-path", GP_PATH), ("recovery", RC_PATH)): mode = stat.S_IMODE(path.stat().st_mode) check(f"T-mode-{name}", f"{path.name} is 0600", mode == 0o600, oct(mode)) overall = all(r["status"] == "PASS" for r in results) out = { "schema": "task26-runbook-lint-results-v1", "linter": Path(__file__).name, "candidate": CANDIDATE, "wheel_sha256": WHEEL, "plan_sha256": PLAN_SHA, "overall": "PASS" if overall else "FAIL", "check_count": len(results), "checks": results, } out_path = HERE / "lint-results.json" fd, tmp = tempfile.mkstemp(prefix=".lint.", dir=str(HERE)) try: os.fchmod(fd, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as fh: json.dump(out, fh, indent=2, sort_keys=False) fh.write("\n") fh.flush() os.fsync(fh.fileno()) os.replace(tmp, out_path) except BaseException: try: os.unlink(tmp) except FileNotFoundError: pass raise print(json.dumps({"overall": out["overall"], "checks": len(results), "failures": [r["id"] for r in results if r["status"] == "FAIL"]}, indent=1)) sys.exit(0 if overall else 1)