feat(revalidate): add pdca revalidate — re-gate a frozen bundle (#11)

template/src/pdca_harness/act.py

@@ -19,7 +19,7 @@
 from dataclasses import dataclass, field
 from pathlib import Path
 
-from . import state
+from . import revalidate, state
 from .config import Config
 
 _DATE_RE = re.compile(r"(\d{4}-\d{2}-\d{2})")
@@ -35,6 +35,7 @@ class ActEntry:
     needs_human: list[str] = field(default_factory=list)  # §6 items (cleared or not)
     unproven: list[str] = field(default_factory=list)  # §7 unproven lines
     act_candidates: list[str] = field(default_factory=list)  # §10 hints
+    reval_deltas: list[str] = field(default_factory=list)  # revalidation stamps (#11)
 
 
 def frozen_bundles(cfg: Config) -> list[Path]:
@@ -77,8 +78,14 @@ def render_index(entries: list[ActEntry], pats: dict[str, list[str]]) -> str:
             f"- §6 NEEDS-HUMAN ({len(e.needs_human)}): " + ("; ".join(e.needs_human) or "—"),
             f"- §7 unproven ({len(e.unproven)}): " + ("; ".join(e.unproven) or "—"),
             f"- §10 Act candidates ({len(e.act_candidates)}): " + ("; ".join(e.act_candidates) or "—"),
-            "",
         ]
+        # Only when present — a frozen gate result the current engine now contradicts
+        # (esp. a frozen FAIL now PASS = stale artifact, or a frozen PASS now FAIL =
+        # regression). Surfaced here so Act can tell stale records from real failures.
+        if e.reval_deltas:
+            lines.append(f"- revalidation deltas ({len(e.reval_deltas)}): "
+                         + "; ".join(e.reval_deltas))
+        lines.append("")
     lines += ["## Recurring signals (appear in >1 cycle)"]
     any_pat = False
     for label, items in pats.items():
@@ -144,6 +151,7 @@ def _extract(summary: Path, bundle: Path) -> ActEntry:
         needs_human=_checkitems(s6),
         unproven=_unproven(s7),
         act_candidates=_candidates(s10),
+        reval_deltas=revalidate.deltas(bundle),  # frozen-gate staleness surfaced (#11)
     )
 
 

template/src/pdca_harness/cli.py

@@ -13,7 +13,7 @@
 import sys
 from pathlib import Path
 
-from . import act, brief, driver, flow, gates, publish, queue, signoff, state
+from . import act, brief, driver, flow, gates, publish, queue, revalidate, signoff, state
 from .config import Config
 
 # Ordering for the cheap-first sign-off queue (docs 03 §sign-off queue).
@@ -64,6 +64,11 @@ def main(argv: list[str] | None = None) -> int:
     p_gates.add_argument("issue_id", nargs="?")
     p_gates.add_argument("--working-tree", action="store_true", help="repo-scoped gates only (the CI merge re-gate)")
 
+    p_reval = sub.add_parser("revalidate",
+                             help="re-run gates on a COMPLETE bundle vs the current engine; write a dated stamp (never re-decides §9)")
+    p_reval.add_argument("issue_id")
+    p_reval.add_argument("--date", help="ISO date for the stamp (default: today)")
+
     p_actidx = sub.add_parser("act-index", help="read-only index of frozen cycles + recurring signals")
     p_actidx.add_argument("--since", help="only cycles signed off on/after this ISO date")
 
@@ -106,6 +111,8 @@ def main(argv: list[str] | None = None) -> int:
         return _queue(cfg)
     if args.cmd == "gates":
         return _gates(cfg, args)
+    if args.cmd == "revalidate":
+        return _revalidate(cfg, args)
     if args.cmd == "act-index":
         return _act_index(cfg, args)
     if args.cmd == "act-log":
@@ -291,6 +298,29 @@ def _gates(cfg: Config, args: argparse.Namespace) -> int:
     return 1 if result["overall"] == "fail" else 0
 
 
+def _revalidate(cfg: Config, args: argparse.Namespace) -> int:
+    """Re-gate a COMPLETE bundle against the current engine; write a dated stamp.
+
+    Reuses the single-sourced gate runner (``gates.run_gates_dry`` — no write to the
+    frozen ``check-gates.json``) and records ``revalidation-<date>.json``. Refuses a
+    non-COMPLETE bundle; never re-decides §9. Exits nonzero iff a row changed, so a
+    delta is visible to the caller; an unchanged result is a quiet confirmation.
+    """
+    d = cfg.bundle(args.issue_id)
+    if not d.exists():
+        print(f"no such bundle: {d}", file=sys.stderr)
+        return 1
+    if state.state(d) != state.COMPLETE:
+        print(f"revalidate refuses {d.name}: not COMPLETE (state {state.state(d)}). "
+              "Revalidation re-gates a frozen bundle; finish sign-off first.",
+              file=sys.stderr)
+        return 2
+    date = args.date or datetime.date.today().isoformat()
+    result = revalidate.revalidate(cfg, d, date)
+    print(revalidate.render_md(result))
+    return 1 if result["changed"] else 0
+
+
 def _act_index(cfg: Config, args: argparse.Namespace) -> int:
     """Print the read-only Act bundle index across frozen cycles."""
     entries = act.index(cfg, since=args.since)

template/src/pdca_harness/gates.py

@@ -45,6 +45,16 @@ def run_working_tree(cfg: Config) -> dict:
     return _finalize(rows, name="working-tree", write_to=None)
 
 
+def run_gates_dry(d: Path, cfg: Config) -> dict:
+    """Run every gate for bundle ``d`` against the CURRENT engine WITHOUT writing the
+    frozen ``check-gates.json`` — the gate runner behind ``pdca revalidate`` (issue #11).
+
+    Same single-sourced ``_run_checks`` as :func:`run_gates`, but ``write_to=None`` so a
+    re-gate of an already-COMPLETE bundle never mutates its frozen record."""
+    rows = _run_checks(cfg, cwd=cfg.root, bundle=d, scopes=("repo", "bundle"))
+    return _finalize(rows, name=d.name, write_to=None)
+
+
 # ----------------------------------------------------------------------------
 def _bundle_target(
     bundle: Path | None,

template/src/pdca_harness/revalidate.py

@@ -0,0 +1,122 @@
+"""``pdca revalidate`` — re-gate a frozen bundle against the current engine (issue #11).
+
+A bundle's ``check-gates.json`` is written once at Check time and frozen when the
+bundle goes COMPLETE; that immutability is correct — the bundle is the record of what
+was decided. But the gates run against a *moving* substrate (the engine code, the
+conformance ruleset, the dependency repos under test). When those improve, a frozen
+``FAIL`` the current engine would never reproduce becomes indistinguishable from a real
+failure the human knowingly accepted.
+
+``revalidate`` re-runs the **same single-sourced gate set** as ``pdca gates``
+(:func:`gates.run_gates_dry` — which never writes the frozen file) against the current
+engine and records an **additive, dated** stamp ``revalidation-<date>.json`` recording
+each row's ``old → new`` result. It **never** mutates ``check-gates.json`` /
+``check-gates.md`` or ``SUMMARY.md`` §9 — the original decision stands. A changed result
+in *either* direction is a delta; a frozen ``PASS`` now ``FAIL`` is a real regression
+signal.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+
+from . import gates
+from .config import Config
+
+
+def revalidate(cfg: Config, d: Path, date: str) -> dict:
+    """Re-gate COMPLETE bundle ``d`` against the current engine; write a dated stamp.
+
+    Returns the revalidation result and writes ``revalidation-<date>.json`` into the
+    bundle — additive (one stamp per date, never overwriting a prior one) and never
+    touching the frozen ``check-gates.json`` / ``check-gates.md`` / §9.
+    """
+    frozen = json.loads((d / "check-gates.json").read_text(encoding="utf-8"))
+    fresh = gates.run_gates_dry(d, cfg)
+
+    old_by = {_row_key(r): r for r in frozen.get("rows", [])}
+    new_by = {_row_key(r): r for r in fresh.get("rows", [])}
+    # Union, preserving the frozen order then any rows the current engine added.
+    keys = list(old_by) + [k for k in new_by if k not in old_by]
+
+    rows = []
+    for key in keys:
+        o, n = old_by.get(key), new_by.get(key)
+        ref = o or n
+        old_res = o["result"] if o else None
+        new_res = n["result"] if n else None
+        rows.append({
+            "check": ref["check"],
+            "element": ref.get("element", ""),
+            "rule_id": ref.get("rule_id", ""),
+            "gating": (n or o).get("gating", False),
+            "old": old_res,
+            "new": new_res,
+            "changed": old_res != new_res,
+        })
+
+    result = {
+        "date": date,
+        "engine_rev": _engine_rev(cfg.root),
+        "bundle": d.name,
+        "frozen_overall": frozen.get("overall"),
+        "current_overall": fresh.get("overall"),
+        "changed": any(r["changed"] for r in rows),
+        # A gating row that was PASS and is now FAIL is the load-bearing signal.
+        "regression": any(r["gating"] and r["old"] == "pass" and r["new"] == "fail"
+                          for r in rows),
+        "rows": rows,
+    }
+    (d / f"revalidation-{date}.json").write_text(
+        json.dumps(result, indent=2) + "\n", encoding="utf-8")
+    return result
+
+
+def render_md(result: dict) -> str:
+    """A compact old→new delta table; changed rows are the deltas, the rest confirm."""
+    head = f"# Revalidation — {result['bundle']} — {result['date']}"
+    rev = f" (engine {result['engine_rev']})" if result.get("engine_rev") else ""
+    verdict = ("REGRESSION — a frozen PASS is now FAIL" if result["regression"]
+               else "deltas found" if result["changed"]
+               else "no change — frozen record confirmed against the current engine")
+    lines = [head + rev, "", f"**{verdict}.** The frozen check-gates.json is unchanged.",
+             "", "| Check | Old | New | Δ | Gating |", "|---|---|---|---|---|"]
+    for r in result["rows"]:
+        delta = "→" if r["changed"] else ""
+        lines.append(f"| {r['check']} | {r['old'] or '—'} | {r['new'] or '—'} | "
+                     f"{delta} | {'yes' if r['gating'] else 'no'} |")
+    return "\n".join(lines) + "\n"
+
+
+def deltas(d: Path) -> list[str]:
+    """One-line summaries of every changed row across ``d``'s revalidation stamps.
+
+    Read-only; for the Act bundle index to surface staleness where Act already looks.
+    """
+    out: list[str] = []
+    for stamp in sorted(d.glob("revalidation-*.json")):
+        try:
+            res = json.loads(stamp.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            continue
+        for r in res.get("rows", []):
+            if r.get("changed"):
+                out.append(f"{r['check']} {r['old']}→{r['new']} ({res.get('date', '?')})")
+    return out
+
+
+def _row_key(row: dict) -> tuple[str, str, str]:
+    """Stable identity for a gate row across re-gates: matrix element + rule + label."""
+    return (row.get("element", ""), row.get("rule_id", ""), row.get("check", ""))
+
+
+def _engine_rev(root: Path) -> str:
+    """Short git rev of the engine under test — best-effort provenance, never fatal."""
+    try:
+        out = subprocess.run(["git", "rev-parse", "--short", "HEAD"], cwd=root,
+                             capture_output=True, text=True, check=True)
+        return out.stdout.strip()
+    except Exception:  # noqa: BLE001 — provenance is a nicety, absence is not an error
+        return ""

template/tests/test_revalidate.py

@@ -0,0 +1,144 @@
+"""Offline slice for `pdca revalidate` — re-gate a frozen bundle (stdlib unittest).
+
+Proves the issue-#11 contract: revalidate re-runs the single-sourced gates against the
+current engine, writes an additive dated stamp, NEVER mutates the frozen
+check-gates.json / check-gates.md / §9, refuses a non-COMPLETE bundle, reports a changed
+row in either direction, and surfaces deltas where Act looks. Deterministic real gates
+(`true` / `false`) flip a gate's result between freeze and revalidate with no Claude /
+Docker. Run from the project root:  PYTHONPATH=src python -m unittest discover -s tests
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+from pdca_harness import act, cli, gates, revalidate, state
+from pdca_harness.config import Config, LeafConfig
+
+# A real bundle-scoped gate keyed on a stable (element, rule_id, check) so a frozen row
+# and a fresh row line up; only the cmd's exit code (true/false) decides pass/fail.
+_GATE = {"id": "C4", "tier": "C4", "label": "verify", "scope": "bundle", "gating": True}
+_PASS = {**_GATE, "cmd": "true"}
+_FAIL = {**_GATE, "cmd": "false"}
+
+
+def _stub_config(root: Path) -> Config:
+    return Config(
+        root=root,
+        bundle_root=root / "results",
+        process_dir=root / "process",
+        templates_dir=root / "templates",
+        default_branch="main",
+        tracker_system="github",
+        tracker_url="",
+        issue_id_example="#1",
+        builder=LeafConfig(mode="stub", family="claude"),
+        reviewer=LeafConfig(mode="stub", family="codex"),
+        planner=LeafConfig(mode="stub", family="claude", interactive=True),
+        signoff=LeafConfig(mode="stub", family="claude", interactive=True),
+        publisher=LeafConfig(mode="stub", family="claude", interactive=True),
+        act=LeafConfig(mode="stub", family="claude", interactive=True),
+    )
+
+
+class Revalidate(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = Path(tempfile.mkdtemp())
+        self.cfg = _stub_config(self.tmp)
+
+    def tearDown(self) -> None:
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def _complete_bundle(self, iid: str, *, frozen_gate: dict) -> Path:
+        """A COMPLETE (frozen) bundle whose check-gates.json was written by `frozen_gate`."""
+        d = self.cfg.bundle(iid)
+        d.mkdir(parents=True)
+        (d / "brief.md").write_text("- **Slug:** reval\n", encoding="utf-8")
+        (d / "patch.diff").write_text("--- a\n+++ b\n", encoding="utf-8")
+        self.cfg.gates_checks = [frozen_gate]
+        gates.run_gates(d, self.cfg)  # writes the frozen check-gates.json / .md
+        (d / "SUMMARY.md").write_text(
+            "## 9. Check sign-off\n- Outcome: accepted\n- By / date: t / 2026-06-04\n",
+            encoding="utf-8")
+        self.assertEqual(state.state(d), state.COMPLETE)
+        return d
+
+    def test_stamp_written_and_frozen_files_untouched(self) -> None:
+        # Frozen FAIL (old engine); the current engine PASSes. Revalidate records the
+        # delta but leaves the frozen record byte-for-byte intact.
+        d = self._complete_bundle("REVAL", frozen_gate=_FAIL)
+        before = {name: (d / name).read_bytes()
+                  for name in ("check-gates.json", "check-gates.md", "SUMMARY.md")}
+        self.cfg.gates_checks = [_PASS]  # engine since fixed
+        result = revalidate.revalidate(self.cfg, d, "2026-06-12")
+
+        self.assertTrue((d / "revalidation-2026-06-12.json").exists())
+        for name, blob in before.items():
+            self.assertEqual((d / name).read_bytes(), blob,
+                             f"revalidate must not touch the frozen {name}")
+        self.assertTrue(result["changed"])
+        self.assertFalse(result["regression"])  # FAIL→PASS is a stale artifact, not a regression
+        c4 = next(r for r in result["rows"] if r["element"] == "C4")
+        self.assertEqual((c4["old"], c4["new"]), ("fail", "pass"))
+
+    def test_regression_when_frozen_pass_now_fails(self) -> None:
+        # Frozen PASS; the current engine FAILs the same gate — the load-bearing signal.
+        d = self._complete_bundle("REG", frozen_gate=_PASS)
+        self.cfg.gates_checks = [_FAIL]
+        result = revalidate.revalidate(self.cfg, d, "2026-06-12")
+        self.assertTrue(result["changed"])
+        self.assertTrue(result["regression"])
+        c4 = next(r for r in result["rows"] if r["element"] == "C4")
+        self.assertEqual((c4["old"], c4["new"]), ("pass", "fail"))
+
+    def test_unchanged_is_a_quiet_confirmation(self) -> None:
+        # Same gate result at freeze and now → no delta; the CLI exits 0.
+        d = self._complete_bundle("SAME", frozen_gate=_PASS)
+        self.cfg.gates_checks = [_PASS]
+        rc = cli._revalidate(self.cfg, SimpleNamespace(issue_id="SAME", date="2026-06-12"))
+        self.assertEqual(rc, 0)
+        stamp = json.loads((d / "revalidation-2026-06-12.json").read_text(encoding="utf-8"))
+        self.assertFalse(stamp["changed"])
+
+    def test_cli_exit_nonzero_on_delta(self) -> None:
+        d = self._complete_bundle("DELTA", frozen_gate=_FAIL)
+        self.cfg.gates_checks = [_PASS]
+        rc = cli._revalidate(self.cfg, SimpleNamespace(issue_id="DELTA", date="2026-06-12"))
+        self.assertEqual(rc, 1)  # a changed row is surfaced to the caller
+        self.assertTrue((d / "revalidation-2026-06-12.json").exists())
+
+    def test_refuses_non_complete_bundle(self) -> None:
+        # A bundle that is only PLANNED (brief, no patch) must be refused, no stamp.
+        d = self.cfg.bundle("PARTIAL")
+        d.mkdir(parents=True)
+        (d / "brief.md").write_text("- **Slug:** x\n", encoding="utf-8")
+        self.assertNotEqual(state.state(d), state.COMPLETE)
+        rc = cli._revalidate(self.cfg, SimpleNamespace(issue_id="PARTIAL", date="2026-06-12"))
+        self.assertEqual(rc, 2)
+        self.assertEqual(list(d.glob("revalidation-*.json")), [])
+
+    def test_missing_bundle_returns_one(self) -> None:
+        rc = cli._revalidate(self.cfg, SimpleNamespace(issue_id="GHOST", date=None))
+        self.assertEqual(rc, 1)
+
+    def test_act_index_surfaces_revalidation_delta(self) -> None:
+        # A COMPLETE bundle carrying a revalidation delta shows it in the Act index,
+        # so Act can tell a stale frozen FAIL from a real accepted failure.
+        d = self._complete_bundle("ACTREVAL", frozen_gate=_FAIL)
+        self.cfg.gates_checks = [_PASS]
+        revalidate.revalidate(self.cfg, d, "2026-06-12")
+        entries = act.index(self.cfg)
+        entry = next(e for e in entries if e.bundle.name == "issue_ACTREVAL")
+        self.assertTrue(entry.reval_deltas)
+        rendered = act.render_index(entries, act.patterns(entries))
+        self.assertIn("revalidation deltas", rendered)
+        self.assertIn("fail→pass", rendered)
+
+
+if __name__ == "__main__":
+    unittest.main()