Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/05-check.md
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,12 @@ They become NEEDS-HUMAN items.
> *new* failure is `[delta]` (your fix may have caused it). You'll see a `[delta]`
> bite in [step 06](06-signoff.md).

**Promoting a check.** A new gate should earn the right to block. Give a check
`promote_after = N` and run `pdca gates --promotions`: it lists the advisory checks that
have **passed in their N most-recent frozen cycles** — earned promotion from advisory to
gating. It's a hint; you flip `gating = true` yourself (nothing is auto-mutated). That is
the Act "promote a check" delta, with a concrete trigger (issue #156).

### Delegating to a host runner

If your project already single-sources its gates in its own runner (`cargo xtask`,
Expand Down
5 changes: 4 additions & 1 deletion template/pdca.toml.jinja
Original file line number Diff line number Diff line change
Expand Up @@ -296,7 +296,10 @@ argv = ["claude", "--agent", "publisher", "--permission-mode", "acceptEdits"]
# runtime, conformance (T1/T2/T4), and interface/E2E tiers all audit code the
# *current* fix did not introduce, so gating them on pre-existing/legacy
# non-conformance is wrong. Promote a tier to gating once its targeted artifacts
# are clean. For interface/E2E, gate a SMOKE test ("does the app start"), not the
# are clean — give a check `promote_after = N` and `pdca gates --promotions` lists the
# advisory checks that passed in their N most-recent frozen cycles (earned promotion; you
# flip `gating`, nothing is auto-mutated). For interface/E2E, gate a SMOKE test ("does the
# app start"), not the
# full suite — the full suite mixes green tests with known-bug repros, so it is a
# characterization, not a pass/fail signal. Cite each tier's rules back to the
# project's normative ruleset (docs/INTEGRATION.md §4) so the gate is auditable.
Expand Down
19 changes: 19 additions & 0 deletions template/src/pdca_harness/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,8 @@ def main(argv: list[str] | None = None) -> int:
p_gates = sub.add_parser("gates", help="run the deterministic Check gates (driver + CI share this)")
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_gates.add_argument("--promotions", action="store_true",
help="list advisory checks clean for their promote_after cycles (#156)")

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)")
Expand Down Expand Up @@ -397,6 +399,8 @@ def _gates(cfg: Config, args: argparse.Namespace) -> int:
The single-sourced entry point: the driver runs gates per bundle during Do,
CI runs ``pdca gates --working-tree`` on the PR — same impl, same pdca.toml.
"""
if getattr(args, "promotions", False):
return _gates_promotions(cfg)
if args.working_tree:
result = gates.run_working_tree(cfg)
else:
Expand All @@ -412,6 +416,21 @@ def _gates(cfg: Config, args: argparse.Namespace) -> int:
return 1 if result["overall"] == "fail" else 0


def _gates_promotions(cfg: Config) -> int:
"""List advisory checks that have earned promotion to gating (#156) — hint-only."""
cands = gates.promotion_candidates(cfg)
if not cands:
print("no advisory checks ready to promote "
"(none with `promote_after` clean across the threshold of recent cycles)")
return 0
print("Advisory checks that have earned promotion to gating "
"(flip `gating = true` in pdca.toml):")
for c in cands:
print(f" - {c['id']}: {c['label']} "
f"(passed ≥ {c['threshold']} most-recent frozen cycles)")
return 0


def _revalidate(cfg: Config, args: argparse.Namespace) -> int:
"""Re-gate a COMPLETE bundle against the current engine; write a dated stamp.

Expand Down
78 changes: 77 additions & 1 deletion template/src/pdca_harness/gates.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,12 +33,13 @@
from __future__ import annotations

import json
import re
import shlex
import shutil
import sys
from pathlib import Path

from . import brief, lane, progress, worktree
from . import brief, lane, progress, state, worktree
from .config import Config

# A gate that cannot RUN its mechanical check (vs. running and failing) declares so:
Expand All @@ -48,6 +49,81 @@
UNVERIFIABLE_MARKER = "PDCA-UNVERIFIABLE:"


# ----------------------------------------------------------------------------
# Gate-promotion lifecycle (issue #156): a check may carry ``promote_after = N``; once it
# has PASSED in its N most-recent frozen cycles it has earned promotion from advisory to
# gating. ``pdca gates --promotions`` lists the ready ones — hint-only, the human flips
# ``gating`` (nothing is auto-mutated). De-risks a new (often Act-proposed) gate, which
# should prove itself advisory before it is allowed to block.
# ----------------------------------------------------------------------------
GATES_JSON = "check-gates.json"
_PROMO_DATE = re.compile(r"(\d{4}-\d{2}-\d{2})")


def _gates_record(d: Path) -> dict | None:
"""A bundle's frozen ``check-gates.json``, or None if absent/unreadable."""
p = d / GATES_JSON
if not p.exists():
return None
try:
return json.loads(p.read_text(encoding="utf-8"))
except (ValueError, OSError):
return None


def _signoff_date(d: Path) -> str:
"""A recency key for ordering frozen cycles — the first ISO date in SUMMARY.md (the §9
sign-off date in practice), or "" when none."""
s = d / "SUMMARY.md"
if not s.exists():
return ""
m = _PROMO_DATE.search(s.read_text(encoding="utf-8"))
Comment thread
eduralph marked this conversation as resolved.
Outdated
return m.group(1) if m else ""


def _check_result(rec: dict, check_id: str) -> str | None:
"""The result this gate record holds for ``check_id`` (``pass`` / ``fail`` /
``unverifiable``), or None when the check didn't run / isn't recorded."""
for row in rec.get("rows", []):
if row.get("rule_id") == check_id:
res = row.get("result")
return res if res in ("pass", "fail", "unverifiable") else None
return None


def promotion_candidates(cfg: Config) -> list[dict]:
"""Advisory checks (``gating = false``) carrying ``promote_after = N`` that have PASSED
in their N most-recent frozen runs — earned promotion to gating. Each:
``{id, label, threshold}``. Hint-only; the human flips ``gating``."""
advisory = [c for c in cfg.gates_checks
if c.get("promote_after") and not bool(c.get("gating", True))]
if not advisory or not cfg.bundle_root.exists():
return []
frozen = sorted((d for d in cfg.bundle_root.glob("issue_*")
if d.is_dir() and state.state(d) == state.COMPLETE),
key=_signoff_date, reverse=True) # newest first
records = [rec for rec in (_gates_record(d) for d in frozen) if rec]
out: list[dict] = []
for chk in advisory:
try:
n = int(chk["promote_after"])
except (TypeError, ValueError):
continue
if n < 1:
continue
ran: list[str] = []
for rec in records:
res = _check_result(rec, chk.get("id", ""))
if res is not None:
ran.append(res)
if len(ran) >= n:
break
if len(ran) >= n and all(r == "pass" for r in ran[:n]):
out.append({"id": chk.get("id", ""), "label": chk.get("label", ""),
"threshold": n})
return out


def run_gates(d: Path, cfg: Config) -> dict:
"""Run every gate for bundle ``d`` (both repo- and bundle-scoped); write JSON."""
rows = _run_checks(cfg, cwd=cfg.root, bundle=d, scopes=("repo", "bundle"))
Expand Down
93 changes: 93 additions & 0 deletions template/tests/test_gate_promotion.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
"""Slice for the gate-promotion lifecycle (issue #156) — `gates.promotion_candidates`.

An advisory check carrying `promote_after = N` earns promotion to gating once it has PASSED
in its N most-recent frozen cycles (a hint; the human flips `gating`). Frozen
`check-gates.json` records drive it; `state` is mocked COMPLETE so the test needn't assemble
full SUMMARY sign-offs. 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 unittest import mock

from pdca_harness import gates, state
from pdca_harness.config import Config, LeafConfig

_CHECK = {"id": "C5-prod", "label": "test exercises production", "scope": "bundle",
"gating": False, "promote_after": 3}


def _cfg(root: Path, checks: list[dict]) -> 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"), reviewer=LeafConfig(mode="stub"),
gates_checks=checks)


class Promotion(unittest.TestCase):
def setUp(self) -> None:
self.tmp = Path(tempfile.mkdtemp())

def tearDown(self) -> None:
shutil.rmtree(self.tmp, ignore_errors=True)

def _bundle(self, cfg: Config, name: str, date: str, result: str) -> None:
d = cfg.bundle(name)
d.mkdir(parents=True)
(d / "check-gates.json").write_text(
json.dumps({"rows": [{"rule_id": "C5-prod", "result": result}]}),
encoding="utf-8")
(d / "SUMMARY.md").write_text(
f"## 9. Check sign-off\n- By / date: t / {date}\n", encoding="utf-8")

def _candidates(self, cfg: Config) -> list[dict]:
with mock.patch.object(gates.state, "state", return_value=state.COMPLETE):
return gates.promotion_candidates(cfg)

def _three(self, cfg: Config, results: tuple[str, str, str]) -> None:
for (name, date), res in zip(
[("A", "2026-06-01"), ("B", "2026-06-02"), ("C", "2026-06-03")], results):
self._bundle(cfg, name, date, res)

def test_ready_when_clean_for_threshold(self) -> None:
cfg = _cfg(self.tmp, [_CHECK])
self._three(cfg, ("pass", "pass", "pass"))
self.assertEqual([c["id"] for c in self._candidates(cfg)], ["C5-prod"])

def test_not_ready_when_most_recent_failed(self) -> None:
cfg = _cfg(self.tmp, [_CHECK])
self._three(cfg, ("pass", "pass", "fail")) # newest (C, 06-03) failed
self.assertEqual(self._candidates(cfg), [])

def test_unverifiable_breaks_the_streak(self) -> None:
cfg = _cfg(self.tmp, [_CHECK])
self._three(cfg, ("pass", "pass", "unverifiable"))
self.assertEqual(self._candidates(cfg), [])

def test_not_ready_below_threshold(self) -> None:
cfg = _cfg(self.tmp, [_CHECK])
self._bundle(cfg, "A", "2026-06-01", "pass")
self._bundle(cfg, "B", "2026-06-02", "pass") # only 2 runs, threshold is 3
self.assertEqual(self._candidates(cfg), [])

def test_gating_check_not_a_candidate(self) -> None:
cfg = _cfg(self.tmp, [{**_CHECK, "gating": True}])
self._three(cfg, ("pass", "pass", "pass"))
self.assertEqual(self._candidates(cfg), [])

def test_without_promote_after_not_a_candidate(self) -> None:
cfg = _cfg(self.tmp, [{k: v for k, v in _CHECK.items() if k != "promote_after"}])
self._three(cfg, ("pass", "pass", "pass"))
self.assertEqual(self._candidates(cfg), [])


if __name__ == "__main__":
unittest.main()
Loading