Merge pull request 'feat(#496): enforce canonical state comment validation before posting' (#497) from feat/issue-496-canonical-comment-validator into master

final_report_validator.py

@@ -196,6 +196,23 @@ _PAGINATION_PROOF_RE = re.compile(
     re.IGNORECASE,
 )
 _GIT_FETCH_RE = re.compile(r"\bgit\s+fetch\b", re.IGNORECASE)
+_CANONICAL_VALIDATION_REJECTED_RE = re.compile(
+    r"canonical comment validation failed|"
+    r"canonical_comment_validation|"
+    r'"allowed"\s*:\s*false',
+    re.IGNORECASE,
+)
+_COMMENT_POSTED_CLAIM_RE = re.compile(
+    r"(?:issue comments posted\s*:\s+(?!none\b)\S|"
+    r"pr comments posted\s*:\s+(?!none\b)\S|"
+    r"comment_id\s*[:=]\s*\d+|"
+    r"gitea comment (?:was )?posted|"
+    r"posted (?:issue|pr|canonical) (?:state )?comment|"
+    r"comment posted successfully|"
+    r"mcp/gitea mutations\s*:\s*[^;\n]*comment posted|"
+    r"reconciliation mutations\s*:\s*[^;\n]*comment posted)",
+    re.IGNORECASE,
+)
 _READONLY_DIAG_RE = re.compile(
     r"read[- ]only diagnostics\s*:\s*(.+)$",
     re.IGNORECASE | re.MULTILINE,
@@ -382,6 +399,43 @@ def _rule_shared_email_disclosure(report_text: str) -> list[dict[str, str]]:
     )
 
 
+def _rule_shared_canonical_comment_post_claim(
+    report_text: str,
+    *,
+    action_log: list[dict] | None = None,
+) -> list[dict[str, str]]:
+    """#496: reports must not claim comment posted when validator rejected."""
+    text = report_text or ""
+    rejected_in_report = bool(_CANONICAL_VALIDATION_REJECTED_RE.search(text))
+    rejected_in_log = False
+    if action_log:
+        for entry in action_log:
+            validation = entry.get("canonical_comment_validation") or {}
+            if validation.get("allowed") is False:
+                rejected_in_log = True
+                break
+            result = entry.get("result") or {}
+            nested = result.get("canonical_comment_validation") or {}
+            if nested.get("allowed") is False:
+                rejected_in_log = True
+                break
+    if not (rejected_in_report or rejected_in_log):
+        return []
+    if not _COMMENT_POSTED_CLAIM_RE.search(text):
+        return []
+    return [
+        validator_finding(
+            "shared.canonical_comment_post_claim",
+            "block",
+            "MCP/Gitea mutations",
+            "report claims a Gitea comment was posted while canonical comment "
+            "validation rejected the workflow comment",
+            "do not claim comment posted when validator fail-closed; repair "
+            "the canonical comment body and retry posting",
+        )
+    ]
+
+
 def _rule_reviewer_legacy_workspace_mutations(
     report_text: str,
     *,
@@ -1372,11 +1426,16 @@ _SHARED_CLEANUP_PROOF_RULES = (
     _rule_shared_mcp_native_cleanup_proof,
 )
 
+_SHARED_CANONICAL_COMMENT_RULES = (
+    _rule_shared_canonical_comment_post_claim,
+)
+
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "review_pr": [
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_legacy_workspace_mutations,
         _rule_reviewer_vague_mutations_none,
@@ -1408,6 +1467,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reconcile_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         *_SHARED_CLEANUP_PROOF_RULES,
         _rule_reconcile_stale_author_fields,
@@ -1425,6 +1485,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
     ],
@@ -1432,6 +1493,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_shared_issue_acceptance_gate,
         _rule_reviewer_vague_mutations_none,
@@ -1443,12 +1505,14 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
     "inventory": [
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reconcile_pagination_proof,
     ],
@@ -1456,6 +1520,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
 }