feat: verify canonical workflow citation and version in review reports (Closes #296)

The canonical PR review/merge workflow already lives in skills/llm-project-workflow/workflows/review-merge-pr.md and is exposed through mcp_list_project_skills / mcp_get_skill_guide (#333-#335), and its final-report schema already demands a workflow version. What was missing is enforcement: add compute_workflow_hash (short sha256 version hash of workflow text) and assess_review_workflow_source to review_proofs. Review reports must cite the canonical workflow file and carry a populated Workflow version field; when the current canonical hash is supplied, a mismatched citation fails as a stale workflow copy, directing the agent to reload via mcp_get_skill_guide. Fails closed. Tests cover hash determinism and content sensitivity, matching-hash pass, missing citation, missing version field, stale hash, version none, and offline validation without a canonical hash. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 09:19:33 -04:00
parent e017d80256
commit 6946534f3e
2 changed files with 170 additions and 0 deletions
@@ -14,6 +14,7 @@ is usable from prompts, harness assertions, and tests. The MCP-level gates
 here weakens or replaces them.
 """

+import hashlib
 import re

 import issue_duplicate_gate
@@ -3097,6 +3098,67 @@ def assess_issue_filing_duplicate_summary(
    }


+# ── Canonical review-workflow citation and version proof (Issue #296) ────────
+#
+# The canonical PR review/merge workflow lives in
+# skills/llm-project-workflow/workflows/review-merge-pr.md and is exposed
+# via mcp_get_skill_guide. Review reports must prove they loaded it and
+# cite the version hash used, so stale or shortened prompt copies are
+# rejected.
+
+REVIEW_WORKFLOW_MARKERS = (
+    "workflows/review-merge-pr.md",
+    "review-merge-pr.md",
+)
+
+
+def compute_workflow_hash(workflow_text):
+    """Return the short (12-hex) sha256 version hash of a workflow text."""
+    digest = hashlib.sha256((workflow_text or "").encode("utf-8"))
+    return digest.hexdigest()[:12]
+
+
+def assess_review_workflow_source(report_text, *, canonical_hash=None):
+    """#296: review reports must cite the canonical workflow and version.
+
+    The report must reference the canonical workflow file and carry a
+    populated ``Workflow version`` (or ``Workflow hash``) field. When
+    *canonical_hash* — ``compute_workflow_hash`` of the current
+    workflows/review-merge-pr.md content — is supplied, the cited hash
+    must match it, rejecting stale copies.
+
+    Returns {'complete', 'downgraded', 'reasons'}; fails closed.
+    """
+    lower = (report_text or "").lower()
+    reasons = []
+
+    if not any(marker in lower for marker in REVIEW_WORKFLOW_MARKERS):
+        reasons.append(
+            "review report missing canonical workflow source citation "
+            "(workflows/review-merge-pr.md)"
+        )
+
+    fields = _report_labeled_fields(report_text)
+    version = fields.get("workflow version") or fields.get("workflow hash")
+    if not version or version.strip().lower().startswith(("none", "unknown")):
+        reasons.append(
+            "review report missing populated 'Workflow version' field "
+            "(version/hash of the canonical workflow used)"
+        )
+    elif canonical_hash and canonical_hash.lower() not in version.lower():
+        reasons.append(
+            f"review report cites stale workflow version '{version}'; "
+            f"current canonical hash is '{canonical_hash}' — reload the "
+            "workflow via mcp_get_skill_guide"
+        )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
 def assess_create_issue_workflow_source(report_text: str) -> dict:
    """#337: create-issue reports must cite the canonical workflow file."""
    lower = (report_text or "").lower()
@@ -0,0 +1,108 @@
+"""Tests for canonical review-workflow citation and version proof (#296).
+
+PR review/merge reports must prove they loaded the canonical workflow
+(workflows/review-merge-pr.md) and cite the workflow version/hash used;
+a stale or missing hash fails so prompt drift is caught.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import (  # noqa: E402
+    assess_review_workflow_source,
+    compute_workflow_hash,
+)
+
+
+CANONICAL_TEXT = "# review-merge-pr workflow v1\nstep one\n"
+CANONICAL_HASH = compute_workflow_hash(CANONICAL_TEXT)
+
+
+def _report(citation=True, version=CANONICAL_HASH):
+    lines = ["Task: review next eligible PR"]
+    if citation:
+        lines.append(
+            "Workflow source: skills/llm-project-workflow/"
+            "workflows/review-merge-pr.md (loaded via mcp_get_skill_guide)"
+        )
+    if version is not None:
+        lines.append(f"- Workflow version: {version}")
+    return "\n".join(lines) + "\n"
+
+
+class TestComputeWorkflowHash(unittest.TestCase):
+    def test_deterministic(self):
+        self.assertEqual(
+            compute_workflow_hash(CANONICAL_TEXT),
+            compute_workflow_hash(CANONICAL_TEXT),
+        )
+
+    def test_changes_with_content(self):
+        self.assertNotEqual(
+            compute_workflow_hash(CANONICAL_TEXT),
+            compute_workflow_hash(CANONICAL_TEXT + "extra rule\n"),
+        )
+
+    def test_short_hex(self):
+        h = compute_workflow_hash(CANONICAL_TEXT)
+        self.assertEqual(len(h), 12)
+        int(h, 16)  # raises if not hex
+
+
+class TestReviewWorkflowSource(unittest.TestCase):
+    def test_citation_and_matching_hash_passes(self):
+        result = assess_review_workflow_source(
+            _report(), canonical_hash=CANONICAL_HASH
+        )
+        self.assertTrue(result["complete"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_missing_citation_fails(self):
+        result = assess_review_workflow_source(
+            _report(citation=False), canonical_hash=CANONICAL_HASH
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("review-merge-pr" in r.lower() for r in result["reasons"])
+        )
+
+    def test_missing_version_field_fails(self):
+        result = assess_review_workflow_source(
+            _report(version=None), canonical_hash=CANONICAL_HASH
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("workflow version" in r.lower() for r in result["reasons"])
+        )
+
+    def test_stale_hash_fails(self):
+        result = assess_review_workflow_source(
+            _report(version="deadbeef0000"), canonical_hash=CANONICAL_HASH
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("stale" in r.lower() or "does not match" in r.lower()
+                for r in result["reasons"])
+        )
+
+    def test_version_none_fails(self):
+        result = assess_review_workflow_source(
+            _report(version="none"), canonical_hash=CANONICAL_HASH
+        )
+        self.assertFalse(result["complete"])
+
+    def test_without_canonical_hash_field_still_required(self):
+        # No canonical hash supplied (e.g. offline validation): citation
+        # and a populated version field are still required.
+        self.assertTrue(
+            assess_review_workflow_source(_report())["complete"]
+        )
+        self.assertFalse(
+            assess_review_workflow_source(_report(version=None))["complete"]
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()