Skip to content

Commit f708470

Browse files
eduralphclaude
andcommitted
feat(gates): gate-promotion lifecycle — surface advisory checks ready to promote (#156)
Promoting a check from advisory (gating=false) to gating was a manual flip with no signal for WHEN it's earned. A check may now carry promote_after=N: gates.promotion_candidates() scans frozen cycles and returns each advisory check that has PASSED in its N most-recent frozen runs. `pdca gates --promotions` lists them — 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 blocks. Documented in pdca.toml + docs/05. Adds test_gate_promotion.py (6 cases). Suite 282 green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 0806b00 commit f708470

5 files changed

Lines changed: 199 additions & 2 deletions

File tree

docs/05-check.md

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,12 @@ They become NEEDS-HUMAN items.
6666
> *new* failure is `[delta]` (your fix may have caused it). You'll see a `[delta]`
6767
> bite in [step 06](06-signoff.md).
6868
69+
**Promoting a check.** A new gate should earn the right to block. Give a check
70+
`promote_after = N` and run `pdca gates --promotions`: it lists the advisory checks that
71+
have **passed in their N most-recent frozen cycles** — earned promotion from advisory to
72+
gating. It's a hint; you flip `gating = true` yourself (nothing is auto-mutated). That is
73+
the Act "promote a check" delta, with a concrete trigger (issue #156).
74+
6975
### Delegating to a host runner
7076

7177
If your project already single-sources its gates in its own runner (`cargo xtask`,

template/pdca.toml.jinja

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -296,7 +296,10 @@ argv = ["claude", "--agent", "publisher", "--permission-mode", "acceptEdits"]
296296
# runtime, conformance (T1/T2/T4), and interface/E2E tiers all audit code the
297297
# *current* fix did not introduce, so gating them on pre-existing/legacy
298298
# non-conformance is wrong. Promote a tier to gating once its targeted artifacts
299-
# are clean. For interface/E2E, gate a SMOKE test ("does the app start"), not the
299+
# are clean — give a check `promote_after = N` and `pdca gates --promotions` lists the
300+
# advisory checks that passed in their N most-recent frozen cycles (earned promotion; you
301+
# flip `gating`, nothing is auto-mutated). For interface/E2E, gate a SMOKE test ("does the
302+
# app start"), not the
300303
# full suite — the full suite mixes green tests with known-bug repros, so it is a
301304
# characterization, not a pass/fail signal. Cite each tier's rules back to the
302305
# project's normative ruleset (docs/INTEGRATION.md §4) so the gate is auditable.

template/src/pdca_harness/cli.py

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,8 @@ def main(argv: list[str] | None = None) -> int:
9090
p_gates = sub.add_parser("gates", help="run the deterministic Check gates (driver + CI share this)")
9191
p_gates.add_argument("issue_id", nargs="?")
9292
p_gates.add_argument("--working-tree", action="store_true", help="repo-scoped gates only (the CI merge re-gate)")
93+
p_gates.add_argument("--promotions", action="store_true",
94+
help="list advisory checks clean for their promote_after cycles (#156)")
9395

9496
p_reval = sub.add_parser("revalidate",
9597
help="re-run gates on a COMPLETE bundle vs the current engine; write a dated stamp (never re-decides §9)")
@@ -397,6 +399,8 @@ def _gates(cfg: Config, args: argparse.Namespace) -> int:
397399
The single-sourced entry point: the driver runs gates per bundle during Do,
398400
CI runs ``pdca gates --working-tree`` on the PR — same impl, same pdca.toml.
399401
"""
402+
if getattr(args, "promotions", False):
403+
return _gates_promotions(cfg)
400404
if args.working_tree:
401405
result = gates.run_working_tree(cfg)
402406
else:
@@ -412,6 +416,21 @@ def _gates(cfg: Config, args: argparse.Namespace) -> int:
412416
return 1 if result["overall"] == "fail" else 0
413417

414418

419+
def _gates_promotions(cfg: Config) -> int:
420+
"""List advisory checks that have earned promotion to gating (#156) — hint-only."""
421+
cands = gates.promotion_candidates(cfg)
422+
if not cands:
423+
print("no advisory checks ready to promote "
424+
"(none with `promote_after` clean across the threshold of recent cycles)")
425+
return 0
426+
print("Advisory checks that have earned promotion to gating "
427+
"(flip `gating = true` in pdca.toml):")
428+
for c in cands:
429+
print(f" - {c['id']}: {c['label']} "
430+
f"(passed ≥ {c['threshold']} most-recent frozen cycles)")
431+
return 0
432+
433+
415434
def _revalidate(cfg: Config, args: argparse.Namespace) -> int:
416435
"""Re-gate a COMPLETE bundle against the current engine; write a dated stamp.
417436

template/src/pdca_harness/gates.py

Lines changed: 77 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,12 +33,13 @@
3333
from __future__ import annotations
3434

3535
import json
36+
import re
3637
import shlex
3738
import shutil
3839
import sys
3940
from pathlib import Path
4041

41-
from . import brief, lane, progress, worktree
42+
from . import brief, lane, progress, state, worktree
4243
from .config import Config
4344

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

5051

52+
# ----------------------------------------------------------------------------
53+
# Gate-promotion lifecycle (issue #156): a check may carry ``promote_after = N``; once it
54+
# has PASSED in its N most-recent frozen cycles it has earned promotion from advisory to
55+
# gating. ``pdca gates --promotions`` lists the ready ones — hint-only, the human flips
56+
# ``gating`` (nothing is auto-mutated). De-risks a new (often Act-proposed) gate, which
57+
# should prove itself advisory before it is allowed to block.
58+
# ----------------------------------------------------------------------------
59+
GATES_JSON = "check-gates.json"
60+
_PROMO_DATE = re.compile(r"(\d{4}-\d{2}-\d{2})")
61+
62+
63+
def _gates_record(d: Path) -> dict | None:
64+
"""A bundle's frozen ``check-gates.json``, or None if absent/unreadable."""
65+
p = d / GATES_JSON
66+
if not p.exists():
67+
return None
68+
try:
69+
return json.loads(p.read_text(encoding="utf-8"))
70+
except (ValueError, OSError):
71+
return None
72+
73+
74+
def _signoff_date(d: Path) -> str:
75+
"""A recency key for ordering frozen cycles — the first ISO date in SUMMARY.md (the §9
76+
sign-off date in practice), or "" when none."""
77+
s = d / "SUMMARY.md"
78+
if not s.exists():
79+
return ""
80+
m = _PROMO_DATE.search(s.read_text(encoding="utf-8"))
81+
return m.group(1) if m else ""
82+
83+
84+
def _check_result(rec: dict, check_id: str) -> str | None:
85+
"""The result this gate record holds for ``check_id`` (``pass`` / ``fail`` /
86+
``unverifiable``), or None when the check didn't run / isn't recorded."""
87+
for row in rec.get("rows", []):
88+
if row.get("rule_id") == check_id:
89+
res = row.get("result")
90+
return res if res in ("pass", "fail", "unverifiable") else None
91+
return None
92+
93+
94+
def promotion_candidates(cfg: Config) -> list[dict]:
95+
"""Advisory checks (``gating = false``) carrying ``promote_after = N`` that have PASSED
96+
in their N most-recent frozen runs — earned promotion to gating. Each:
97+
``{id, label, threshold}``. Hint-only; the human flips ``gating``."""
98+
advisory = [c for c in cfg.gates_checks
99+
if c.get("promote_after") and not bool(c.get("gating", True))]
100+
if not advisory or not cfg.bundle_root.exists():
101+
return []
102+
frozen = sorted((d for d in cfg.bundle_root.glob("issue_*")
103+
if d.is_dir() and state.state(d) == state.COMPLETE),
104+
key=_signoff_date, reverse=True) # newest first
105+
records = [rec for rec in (_gates_record(d) for d in frozen) if rec]
106+
out: list[dict] = []
107+
for chk in advisory:
108+
try:
109+
n = int(chk["promote_after"])
110+
except (TypeError, ValueError):
111+
continue
112+
if n < 1:
113+
continue
114+
ran: list[str] = []
115+
for rec in records:
116+
res = _check_result(rec, chk.get("id", ""))
117+
if res is not None:
118+
ran.append(res)
119+
if len(ran) >= n:
120+
break
121+
if len(ran) >= n and all(r == "pass" for r in ran[:n]):
122+
out.append({"id": chk.get("id", ""), "label": chk.get("label", ""),
123+
"threshold": n})
124+
return out
125+
126+
51127
def run_gates(d: Path, cfg: Config) -> dict:
52128
"""Run every gate for bundle ``d`` (both repo- and bundle-scoped); write JSON."""
53129
rows = _run_checks(cfg, cwd=cfg.root, bundle=d, scopes=("repo", "bundle"))
Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
"""Slice for the gate-promotion lifecycle (issue #156) — `gates.promotion_candidates`.
2+
3+
An advisory check carrying `promote_after = N` earns promotion to gating once it has PASSED
4+
in its N most-recent frozen cycles (a hint; the human flips `gating`). Frozen
5+
`check-gates.json` records drive it; `state` is mocked COMPLETE so the test needn't assemble
6+
full SUMMARY sign-offs. Run from the project root:
7+
PYTHONPATH=src python -m unittest discover -s tests
8+
"""
9+
10+
from __future__ import annotations
11+
12+
import json
13+
import shutil
14+
import tempfile
15+
import unittest
16+
from pathlib import Path
17+
from unittest import mock
18+
19+
from pdca_harness import gates, state
20+
from pdca_harness.config import Config, LeafConfig
21+
22+
_CHECK = {"id": "C5-prod", "label": "test exercises production", "scope": "bundle",
23+
"gating": False, "promote_after": 3}
24+
25+
26+
def _cfg(root: Path, checks: list[dict]) -> Config:
27+
return Config(
28+
root=root, bundle_root=root / "results", process_dir=root / "process",
29+
templates_dir=root / "templates", default_branch="main", tracker_system="github",
30+
tracker_url="", issue_id_example="#1",
31+
builder=LeafConfig(mode="stub"), reviewer=LeafConfig(mode="stub"),
32+
gates_checks=checks)
33+
34+
35+
class Promotion(unittest.TestCase):
36+
def setUp(self) -> None:
37+
self.tmp = Path(tempfile.mkdtemp())
38+
39+
def tearDown(self) -> None:
40+
shutil.rmtree(self.tmp, ignore_errors=True)
41+
42+
def _bundle(self, cfg: Config, name: str, date: str, result: str) -> None:
43+
d = cfg.bundle(name)
44+
d.mkdir(parents=True)
45+
(d / "check-gates.json").write_text(
46+
json.dumps({"rows": [{"rule_id": "C5-prod", "result": result}]}),
47+
encoding="utf-8")
48+
(d / "SUMMARY.md").write_text(
49+
f"## 9. Check sign-off\n- By / date: t / {date}\n", encoding="utf-8")
50+
51+
def _candidates(self, cfg: Config) -> list[dict]:
52+
with mock.patch.object(gates.state, "state", return_value=state.COMPLETE):
53+
return gates.promotion_candidates(cfg)
54+
55+
def _three(self, cfg: Config, results: tuple[str, str, str]) -> None:
56+
for (name, date), res in zip(
57+
[("A", "2026-06-01"), ("B", "2026-06-02"), ("C", "2026-06-03")], results):
58+
self._bundle(cfg, name, date, res)
59+
60+
def test_ready_when_clean_for_threshold(self) -> None:
61+
cfg = _cfg(self.tmp, [_CHECK])
62+
self._three(cfg, ("pass", "pass", "pass"))
63+
self.assertEqual([c["id"] for c in self._candidates(cfg)], ["C5-prod"])
64+
65+
def test_not_ready_when_most_recent_failed(self) -> None:
66+
cfg = _cfg(self.tmp, [_CHECK])
67+
self._three(cfg, ("pass", "pass", "fail")) # newest (C, 06-03) failed
68+
self.assertEqual(self._candidates(cfg), [])
69+
70+
def test_unverifiable_breaks_the_streak(self) -> None:
71+
cfg = _cfg(self.tmp, [_CHECK])
72+
self._three(cfg, ("pass", "pass", "unverifiable"))
73+
self.assertEqual(self._candidates(cfg), [])
74+
75+
def test_not_ready_below_threshold(self) -> None:
76+
cfg = _cfg(self.tmp, [_CHECK])
77+
self._bundle(cfg, "A", "2026-06-01", "pass")
78+
self._bundle(cfg, "B", "2026-06-02", "pass") # only 2 runs, threshold is 3
79+
self.assertEqual(self._candidates(cfg), [])
80+
81+
def test_gating_check_not_a_candidate(self) -> None:
82+
cfg = _cfg(self.tmp, [{**_CHECK, "gating": True}])
83+
self._three(cfg, ("pass", "pass", "pass"))
84+
self.assertEqual(self._candidates(cfg), [])
85+
86+
def test_without_promote_after_not_a_candidate(self) -> None:
87+
cfg = _cfg(self.tmp, [{k: v for k, v in _CHECK.items() if k != "promote_after"}])
88+
self._three(cfg, ("pass", "pass", "pass"))
89+
self.assertEqual(self._candidates(cfg), [])
90+
91+
92+
if __name__ == "__main__":
93+
unittest.main()

0 commit comments

Comments
 (0)