feat: comment-then-stop policy for partial PR reconciliation (Closes #302) #375

Merged
sysadmin merged 3 commits from feat/issue-302-partial-reconciliation-policy into master 2026-07-07 08:51:13 -05:00
4 changed files with 322 additions and 0 deletions
+165
View File
@@ -3965,6 +3965,168 @@ def assess_reviewer_baseline_validation_proof(
# ---------------------------------------------------------------------------
# Partial reconciliation policy (#302)
# ---------------------------------------------------------------------------
# Durable policy choice for already-landed PR reconciliation when
# ``gitea.pr.close`` is unavailable: Option B, comment-then-stop.
PARTIAL_RECONCILIATION_POLICY = "comment_then_stop"
PARTIAL_RECONCILIATION_COMMENT_FIELDS = (
"PR head SHA",
"target branch SHA",
"ancestor proof",
"linked issue status",
"required missing capability",
)
def resolve_partial_reconciliation_plan(
*,
ancestor_proven: bool,
capabilities: dict | None,
) -> dict:
"""#302: decide reconciliation mutations from proven capabilities.
Implements the comment-then-stop policy (Option B): with ancestor
proof but no ``close_pr`` capability, a reconciliation comment with
the full proof is posted when ``comment_pr`` is proven, then the
workflow stops for an authorized close. Missing comment capability
degrades to a recovery handoff with no Gitea mutation. Unproven
ancestry blocks every mutation regardless of capability.
*capabilities* maps ``close_pr``/``comment_pr``/``close_issue``/
``comment_issue`` to proven booleans; ``None`` or missing keys fail
closed.
"""
caps = capabilities or {}
if not ancestor_proven:
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "GATE_NOT_PROVEN",
"allowed_mutations": (),
"required_comment_fields": (),
"report_requirements": (
"state that ancestry was not proven and no Gitea mutation "
"was performed",
),
"reasons": [
"ancestor proof missing; no reconciliation mutation may "
"run (#302)"
],
"safe_next_action": (
"run the already-landed ancestry check before any "
"reconciliation mutation"
),
}
if caps.get("close_pr") is True:
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "FULL_RECONCILE_CLOSE_ALLOWED",
"allowed_mutations": ("close_pr", "comment_pr"),
"required_comment_fields": (),
"report_requirements": (
"report the PR close result",
),
"reasons": [],
"safe_next_action": (
"close the already-landed PR with the proven close_pr "
"capability and report the close result"
),
}
if caps.get("comment_pr") is True:
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "PARTIAL_RECONCILE_COMMENT_THEN_STOP",
"allowed_mutations": ("comment_pr",),
"required_comment_fields": PARTIAL_RECONCILIATION_COMMENT_FIELDS,
"report_requirements": (
"report that the reconciliation comment was posted",
"report the missing close capability that prevented the "
"PR close",
),
"reasons": [
"close_pr capability missing; policy is comment-then-stop "
"(#302)"
],
"safe_next_action": (
"post one reconciliation comment carrying the ancestor "
"proof, then stop for a human or authorized close"
),
}
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "RECOVERY_HANDOFF_ONLY",
"allowed_mutations": (),
"required_comment_fields": (),
"report_requirements": (
"report that no reconciliation comment was posted and name "
"the missing capability",
"report that no Gitea mutation was performed",
),
"reasons": [
"close_pr and comment_pr capabilities missing; recovery "
"handoff only (#302)"
],
"safe_next_action": (
"produce a recovery handoff recording the intended comment "
"and required capabilities; perform no Gitea mutation"
),
}
def assess_partial_reconciliation_report(
report_text: str | None,
*,
plan: dict,
) -> dict:
"""#302: final reports must explain why comments were or were not posted."""
text = (report_text or "").lower()
outcome = (plan or {}).get("outcome", "")
reasons: list[str] = []
if outcome == "FULL_RECONCILE_CLOSE_ALLOWED":
if "close" not in text or "result" not in text:
reasons.append(
"full reconciliation report must state the PR close "
"result (#302)"
)
elif outcome == "PARTIAL_RECONCILE_COMMENT_THEN_STOP":
if "comment" not in text:
reasons.append(
"partial reconciliation report must state that the "
"reconciliation comment was posted (#302)"
)
if "capability" not in text and "close_pr" not in text:
reasons.append(
"partial reconciliation report must name the missing "
"close capability (#302)"
)
elif outcome == "RECOVERY_HANDOFF_ONLY":
if "comment" not in text or "capability" not in text:
reasons.append(
"recovery handoff report must explain that no comment was "
"posted and name the missing capability (#302)"
)
if "no gitea mutation" not in text and "no mutation" not in text:
reasons.append(
"recovery handoff report must state that no Gitea "
"mutation was performed (#302)"
)
return {
"complete": not reasons,
"block": bool(reasons),
"reasons": reasons,
}
# ---------------------------------------------------------------------------
# Already-landed review gate (#292)
# ---------------------------------------------------------------------------
@@ -4130,6 +4292,9 @@ def assess_already_landed_report_state(
}
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Full-suite failure approval gate (#323)
# ---------------------------------------------------------------------------
@@ -248,6 +248,28 @@ Post a reconciliation comment only if:
If comment capability is missing, record the intended comment in the final
handoff only.
## 12A. Partial reconciliation policy (#302)
The durable policy when `close_pr` capability is missing is
**comment-then-stop** (`resolve_partial_reconciliation_plan` in
`review_proofs.py` enforces it):
* `close_pr` proven → full reconciliation: close the PR and report the
close result (`FULL_RECONCILE_CLOSE_ALLOWED`).
* `close_pr` missing, `comment_pr` proven → post exactly one
reconciliation comment carrying PR head SHA, target branch SHA,
ancestor proof, linked issue status, and the required missing
capability, then stop for a human or authorized close
(`PARTIAL_RECONCILE_COMMENT_THEN_STOP`).
* `comment_pr` also missing → no Gitea mutation; produce a recovery
handoff recording the intended comment and required capabilities
(`RECOVERY_HANDOFF_ONLY`).
* Ancestry not proven → no mutation regardless of capability
(`GATE_NOT_PROVEN`).
Final reports must explain why the comment was or was not posted and
name the missing capability (`assess_partial_reconciliation_report`).
## 13. PR close rules
Close a PR only if:
+6
View File
@@ -78,6 +78,12 @@ def test_reconcile_landed_workflow_contract():
assert "canonical: true" in text
assert "Do not review or merge normal PRs" in text
assert "Already-landed proof" in text
# Issue #302: partial reconciliation policy must stay documented.
assert "## 12A. Partial reconciliation policy (#302)" in text
assert "comment-then-stop" in text
assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text
assert "RECOVERY_HANDOFF_ONLY" in text
assert "resolve_partial_reconciliation_plan" in text
def test_create_issue_workflow_contract():
+129
View File
@@ -26,6 +26,8 @@ from review_proofs import ( # noqa: E402
assess_already_landed_report_state,
assess_already_landed_review_gate,
assess_author_pr_report,
assess_partial_reconciliation_report,
resolve_partial_reconciliation_plan,
assess_validation_environment_proof,
assess_full_suite_failure_approval_gate,
assess_queue_ordering_proof,
@@ -2641,6 +2643,133 @@ class TestReviewerBaselineValidationProof(unittest.TestCase):
self.assertFalse(report["baseline_validation_proven"])
class TestPartialReconciliationPlan(unittest.TestCase):
"""Issue #302: policy when PR-close capability is missing (Option B)."""
def _plan(self, **overrides):
kwargs = {
"ancestor_proven": True,
"capabilities": {
"close_pr": False,
"comment_pr": True,
"close_issue": True,
"comment_issue": True,
},
}
kwargs.update(overrides)
return resolve_partial_reconciliation_plan(**kwargs)
def test_close_capability_present_allows_full_reconcile(self):
plan = self._plan(capabilities={
"close_pr": True, "comment_pr": True,
"close_issue": True, "comment_issue": True,
})
self.assertEqual(plan["outcome"], "FULL_RECONCILE_CLOSE_ALLOWED")
self.assertIn("close_pr", plan["allowed_mutations"])
def test_close_missing_with_comment_posts_comment_then_stops(self):
plan = self._plan()
self.assertEqual(
plan["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP")
self.assertEqual(plan["allowed_mutations"], ("comment_pr",))
for field in (
"PR head SHA",
"target branch SHA",
"ancestor proof",
"linked issue status",
"required missing capability",
):
self.assertIn(field, plan["required_comment_fields"])
def test_comment_capability_missing_produces_handoff_only(self):
plan = self._plan(capabilities={
"close_pr": False, "comment_pr": False,
"close_issue": True, "comment_issue": True,
})
self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY")
self.assertEqual(plan["allowed_mutations"], ())
def test_all_capabilities_missing_produces_handoff_only(self):
plan = self._plan(capabilities={
"close_pr": False, "comment_pr": False,
"close_issue": False, "comment_issue": False,
})
self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY")
self.assertEqual(plan["allowed_mutations"], ())
def test_unproven_ancestry_blocks_all_mutations(self):
plan = self._plan(ancestor_proven=False, capabilities={
"close_pr": True, "comment_pr": True,
"close_issue": True, "comment_issue": True,
})
self.assertEqual(plan["outcome"], "GATE_NOT_PROVEN")
self.assertEqual(plan["allowed_mutations"], ())
def test_missing_capability_dict_fails_closed(self):
plan = self._plan(capabilities=None)
self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY")
self.assertEqual(plan["allowed_mutations"], ())
class TestPartialReconciliationReport(unittest.TestCase):
"""Issue #302: reports must explain why comments were or were not posted."""
def _plan(self, outcome):
capabilities = {
"FULL_RECONCILE_CLOSE_ALLOWED": {
"close_pr": True, "comment_pr": True,
"close_issue": True, "comment_issue": True,
},
"PARTIAL_RECONCILE_COMMENT_THEN_STOP": {
"close_pr": False, "comment_pr": True,
"close_issue": True, "comment_issue": True,
},
"RECOVERY_HANDOFF_ONLY": {
"close_pr": False, "comment_pr": False,
"close_issue": False, "comment_issue": False,
},
}[outcome]
return resolve_partial_reconciliation_plan(
ancestor_proven=True, capabilities=capabilities)
def test_partial_report_requires_comment_and_missing_capability(self):
plan = self._plan("PARTIAL_RECONCILE_COMMENT_THEN_STOP")
good = (
"Reconciliation comment posted on PR #278 with ancestor proof. "
"PR close skipped: close capability gitea.pr.close missing; "
"stopped for authorized close."
)
result = assess_partial_reconciliation_report(good, plan=plan)
self.assertTrue(result["complete"])
bad = "Stopped. Nothing done."
result = assess_partial_reconciliation_report(bad, plan=plan)
self.assertTrue(result["block"])
def test_handoff_only_report_requires_no_comment_explanation(self):
plan = self._plan("RECOVERY_HANDOFF_ONLY")
good = (
"No reconciliation comment posted: comment capability missing. "
"No Gitea mutation performed; recovery handoff produced."
)
result = assess_partial_reconciliation_report(good, plan=plan)
self.assertTrue(result["complete"])
bad = "Recovery handoff produced."
result = assess_partial_reconciliation_report(bad, plan=plan)
self.assertTrue(result["block"])
def test_full_reconcile_report_requires_close_result(self):
plan = self._plan("FULL_RECONCILE_CLOSE_ALLOWED")
good = "PR close result: closed PR #278 after ancestor proof."
result = assess_partial_reconciliation_report(good, plan=plan)
self.assertTrue(result["complete"])
bad = "Reconciliation done."
result = assess_partial_reconciliation_report(bad, plan=plan)
self.assertTrue(result["block"])
class TestAlreadyLandedReviewGate(unittest.TestCase):
"""Issue #292: PR head already on target blocks approval and merge."""