feat(#496): fail-closed canonical comment validation before Gitea posts

Add canonical_comment_validator and wire it into issue comments, PR review
bodies, reconcile post_comment, and structured comment helpers. Workflow-changing
comments without complete next-action state are rejected before any API call.
Includes unit/MCP integration tests, runbook docs, and a final-report rule
blocking false "comment posted" claims when validation failed.
This commit is contained in:
2026-07-08 22:46:22 -04:00
parent 9a2e585a9e
commit 84d921a52f
7 changed files with 855 additions and 0 deletions
+61
View File
@@ -829,6 +829,7 @@ import review_merge_state_machine # noqa: E402
import pr_work_lease # noqa: E402
import native_mcp_preference # noqa: E402
import worktree_cleanup_audit # noqa: E402
import canonical_comment_validator as ccv # noqa: E402
# Keyed issue-lock storage (#443): per remote/org/repo/issue files under
@@ -3268,6 +3269,13 @@ def _evaluate_pr_review_submission(
result["pr_work_lease"] = lease_block
return result
if (body or "").strip():
gate = _canonical_comment_gate(body, context="pr_review")
if gate["blocked"]:
reasons.extend(gate["reasons"])
result["canonical_comment_validation"] = gate["canonical_comment_validation"]
return result
result["would_perform"] = True
if not live:
reasons.append(
@@ -5207,6 +5215,12 @@ def gitea_reconcile_already_landed_pr(
"gitea.pr.comment"
)
return result
gate = _canonical_comment_gate(comment_body)
if gate["blocked"]:
result["success"] = False
result["reasons"].extend(gate["reasons"])
result["canonical_comment_validation"] = gate["canonical_comment_validation"]
return result
comment_url = f"{base}/issues/{pr_number}/comments"
with _audited(
"comment_pr",
@@ -6444,6 +6458,15 @@ def gitea_create_issue_comment(
blocked["permission_report"] = _permission_block_report(
"gitea.issue.comment")
return blocked
canon_gate = _canonical_comment_gate(body, context="issue_comment")
if canon_gate["blocked"]:
return {
"success": False,
"performed": False,
"issue_number": issue_number,
"reasons": canon_gate["reasons"],
"canonical_comment_validation": canon_gate["canonical_comment_validation"],
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"
@@ -7973,6 +7996,34 @@ def gitea_audit_config() -> dict:
return report
def _canonical_comment_gate(
body: str,
*,
context: str | None = None,
force_workflow: bool = False,
) -> dict:
"""Fail-closed canonical state validation before comment/review POST (#496)."""
assessment = ccv.assess_canonical_comment(
body,
context=context,
force_workflow=force_workflow,
)
if assessment.get("allowed"):
return {
"blocked": False,
"canonical_comment_validation": assessment,
"reasons": [],
}
correction = (assessment.get("correction_message") or "").strip()
return {
"blocked": True,
"canonical_comment_validation": assessment,
"reasons": [
correction or "canonical comment validation failed (fail closed before posting)"
],
}
def _post_structured_issue_comment(
*,
issue_number: int,
@@ -7982,8 +8033,18 @@ def _post_structured_issue_comment(
org: str | None,
repo: str | None,
audit_op: str = "create_issue_comment",
comment_context: str | None = None,
) -> dict:
"""Post an issue-thread comment after permission gates already passed."""
gate = _canonical_comment_gate(body, context=comment_context)
if gate["blocked"]:
return {
"success": False,
"performed": False,
"issue_number": issue_number,
"reasons": gate["reasons"],
"canonical_comment_validation": gate["canonical_comment_validation"],
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"