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:
@@ -0,0 +1,217 @@
|
||||
"""Unit tests for canonical_comment_validator (#496)."""
|
||||
|
||||
import unittest
|
||||
|
||||
import canonical_comment_validator as ccv
|
||||
|
||||
FULL_SHA = "a" * 40
|
||||
|
||||
_VALID_ISSUE = f"""## Canonical Issue State
|
||||
|
||||
STATE: ready-for-author
|
||||
WHO_IS_NEXT: author
|
||||
NEXT_ACTION: Implement canonical comment validator and wire MCP gates
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
You are the author LLM for issue #496. Add canonical_comment_validator.py
|
||||
and wire fail-closed gates into gitea_create_issue_comment.
|
||||
```
|
||||
WHAT_HAPPENED: Issue filed for MCP validation wall
|
||||
WHY: Workflow comments must carry durable next-action state
|
||||
RELATED_PRS: none
|
||||
BLOCKERS: none
|
||||
VALIDATION: unit tests not yet run
|
||||
LAST_UPDATED_BY: prgs-author
|
||||
"""
|
||||
|
||||
_VALID_PR = f"""## Canonical PR State
|
||||
|
||||
STATE: ready-for-review
|
||||
WHO_IS_NEXT: reviewer
|
||||
NEXT_ACTION: Run focused pytest and review the validator wiring diff
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
Review PR for issue #496. Confirm casual comments still post and workflow
|
||||
comments without canonical fields are rejected before the Gitea API call.
|
||||
```
|
||||
WHAT_HAPPENED: Author opened PR with validator module
|
||||
WHY: Fail-closed enforcement belongs at the MCP boundary
|
||||
ISSUE: #496
|
||||
HEAD_SHA: {FULL_SHA}
|
||||
REVIEW_STATUS: pending
|
||||
MERGE_READY: false
|
||||
BLOCKERS: none
|
||||
VALIDATION: not yet run
|
||||
LAST_UPDATED_BY: prgs-author
|
||||
"""
|
||||
|
||||
_VALID_SUPERSEDED = """## Canonical PR State
|
||||
|
||||
STATE: superseded
|
||||
WHO_IS_NEXT: reconciler
|
||||
NEXT_ACTION: Close superseded issue after canonical issue #495 lands
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
Close issue #494 as superseded by #495 once templates and docs are merged.
|
||||
```
|
||||
WHY: Duplicate tracking issue replaced by durable canonical template work
|
||||
CANONICAL_ITEM: issue #495
|
||||
SUPERSEDED_ITEM: issue #494
|
||||
CLOSE_OR_KEEP_OPEN: close superseded issue #494 after #495 merges
|
||||
"""
|
||||
|
||||
_VALID_BLOCKED = """## Canonical Issue State
|
||||
|
||||
STATE: blocked
|
||||
WHO_IS_NEXT: author
|
||||
NEXT_ACTION: Fix failing validator unit tests before requesting re-review
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
Run pytest tests/test_canonical_comment_validator.py and repair failures.
|
||||
```
|
||||
WHAT_HAPPENED: Validator tests failed in CI
|
||||
WHY: Blocked until test suite is green
|
||||
RELATED_PRS: PR #500
|
||||
BLOCKERS: pytest failures; unblock after all validator tests pass
|
||||
VALIDATION: pytest failed
|
||||
LAST_UPDATED_BY: prgs-author
|
||||
"""
|
||||
|
||||
_VALID_READY_TO_MERGE = f"""## Canonical PR State
|
||||
|
||||
STATE: ready-to-merge
|
||||
WHO_IS_NEXT: merger
|
||||
NEXT_ACTION: Merge PR after confirming approval_at_current_head
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
Merge PR #500 for issue #496 after live mergeable check passes.
|
||||
```
|
||||
WHAT_HAPPENED: Reviewer approved at current head
|
||||
WHY: All gates passed and head SHA is current
|
||||
ISSUE: #496
|
||||
HEAD_SHA: {FULL_SHA}
|
||||
REVIEW_STATUS: approved / approval_at_current_head
|
||||
MERGE_READY: true
|
||||
BLOCKERS: none
|
||||
VALIDATION: pytest passed; reviewer approved at head {FULL_SHA}
|
||||
LAST_UPDATED_BY: prgs-reviewer
|
||||
"""
|
||||
|
||||
|
||||
class TestCanonicalCommentDetection(unittest.TestCase):
|
||||
def test_casual_comment_not_workflow(self):
|
||||
self.assertFalse(ccv.is_workflow_changing_comment("Thanks, I will check this."))
|
||||
|
||||
def test_workflow_trigger_detected(self):
|
||||
self.assertTrue(ccv.is_workflow_changing_comment("Blocked, author should fix."))
|
||||
|
||||
def test_machine_marker_exempt(self):
|
||||
body = "<!-- mcp-review-lease:v1 -->\nphase: claimed"
|
||||
self.assertFalse(ccv.is_workflow_changing_comment(body))
|
||||
|
||||
|
||||
class TestCanonicalCommentValidation(unittest.TestCase):
|
||||
def test_casual_comment_allowed(self):
|
||||
result = ccv.assess_canonical_comment("Thanks, I will check this.")
|
||||
self.assertTrue(result["allowed"])
|
||||
self.assertFalse(result["is_workflow_comment"])
|
||||
|
||||
def test_workflow_comment_missing_fields_rejected(self):
|
||||
result = ccv.assess_canonical_comment("Blocked, author should fix.")
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertTrue(result["is_workflow_comment"])
|
||||
self.assertIn("WHO_IS_NEXT", result["missing_fields"])
|
||||
self.assertIn("correction_message", result)
|
||||
self.assertIn("Suggested template", result["correction_message"])
|
||||
|
||||
def test_missing_next_prompt_rejected(self):
|
||||
body = """STATE: blocked
|
||||
WHO_IS_NEXT: author
|
||||
NEXT_ACTION: Repair the failing validator unit tests in CI
|
||||
WHY: Tests must pass before merge
|
||||
BLOCKERS: pytest failed; unblock after tests pass
|
||||
VALIDATION: failed
|
||||
"""
|
||||
result = ccv.assess_canonical_comment(body)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("NEXT_PROMPT", result["missing_fields"] + result["vague_fields"])
|
||||
|
||||
def test_vague_next_action_rejected(self):
|
||||
body = _VALID_ISSUE.replace(
|
||||
"Implement canonical comment validator and wire MCP gates",
|
||||
"continue",
|
||||
)
|
||||
result = ccv.assess_canonical_comment(body)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("NEXT_ACTION", result["vague_fields"])
|
||||
|
||||
def test_invalid_who_is_next_rejected(self):
|
||||
body = _VALID_ISSUE.replace("WHO_IS_NEXT: author", "WHO_IS_NEXT: llm")
|
||||
result = ccv.assess_canonical_comment(body)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("WHO_IS_NEXT", result["vague_fields"])
|
||||
|
||||
def test_valid_issue_comment_allowed(self):
|
||||
result = ccv.assess_canonical_comment(_VALID_ISSUE)
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
def test_issue_comment_mentioning_pr_requires_related_prs(self):
|
||||
body = _VALID_ISSUE.replace("RELATED_PRS: none", "RELATED_PRS:")
|
||||
body = body.replace("Issue filed", "Issue filed; see PR #500")
|
||||
result = ccv.assess_canonical_comment(body)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("RELATED_PRS", result["missing_fields"])
|
||||
|
||||
def test_pr_review_requires_head_sha(self):
|
||||
body = _VALID_PR.replace(f"HEAD_SHA: {FULL_SHA}", "HEAD_SHA:")
|
||||
result = ccv.assess_canonical_comment(body, context="pr_review")
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("HEAD_SHA", result["missing_fields"])
|
||||
|
||||
def test_ready_to_merge_requires_approval_and_sha_proof(self):
|
||||
body = _VALID_READY_TO_MERGE.replace("approval_at_current_head", "pending")
|
||||
body = body.replace(f"HEAD_SHA: {FULL_SHA}", "HEAD_SHA: pending")
|
||||
body = body.replace("reviewer approved at head", "reviewer pending at head")
|
||||
body = body.replace("approved /", "pending /")
|
||||
result = ccv.assess_canonical_comment(body, context="pr_comment")
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertTrue(result["extra_reasons"])
|
||||
|
||||
def test_ready_to_merge_valid_allowed(self):
|
||||
result = ccv.assess_canonical_comment(_VALID_READY_TO_MERGE, context="pr_comment")
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
def test_superseded_requires_canonical_and_superseded_items(self):
|
||||
body = _VALID_SUPERSEDED.replace("CANONICAL_ITEM: issue #495", "CANONICAL_ITEM:")
|
||||
result = ccv.assess_canonical_comment(body, context="supersession")
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("CANONICAL_ITEM", result["missing_fields"])
|
||||
|
||||
def test_superseded_valid_allowed(self):
|
||||
result = ccv.assess_canonical_comment(_VALID_SUPERSEDED, context="supersession")
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
def test_blocked_requires_unblock_condition(self):
|
||||
body = _VALID_BLOCKED.replace(
|
||||
"BLOCKERS: pytest failures; unblock after all validator tests pass",
|
||||
"BLOCKERS: pytest failures",
|
||||
)
|
||||
result = ccv.assess_canonical_comment(body)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("BLOCKERS", result["vague_fields"])
|
||||
|
||||
def test_blocked_valid_allowed(self):
|
||||
result = ccv.assess_canonical_comment(_VALID_BLOCKED)
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
def test_correction_lists_missing_fields(self):
|
||||
result = ccv.assess_canonical_comment("ready for merge")
|
||||
self.assertFalse(result["allowed"])
|
||||
message = result["correction_message"]
|
||||
self.assertIn("Missing fields", message)
|
||||
self.assertIn("WHO_IS_NEXT", message)
|
||||
self.assertIn("Suggested template", message)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -498,6 +498,31 @@ class TestCanonicalReconcileSchema(unittest.TestCase):
|
||||
)
|
||||
|
||||
|
||||
class TestCanonicalCommentPostClaim(unittest.TestCase):
|
||||
def test_blocks_post_claim_when_validator_rejected_in_report(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- MCP/Gitea mutations: issue comment posted",
|
||||
"- Issue comments posted: 1 canonical state comment",
|
||||
"canonical comment validation failed (fail closed before posting).",
|
||||
'- canonical_comment_validation: {"allowed": false}',
|
||||
])
|
||||
result = assess_final_report_validator(report, "work_issue")
|
||||
rule_ids = [f["rule_id"] for f in result["findings"]]
|
||||
self.assertIn("shared.canonical_comment_post_claim", rule_ids)
|
||||
self.assertTrue(result["blocked"])
|
||||
|
||||
def test_allows_none_when_validator_rejected_without_post_claim(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Issue comments posted: none",
|
||||
"canonical comment validation failed (fail closed before posting).",
|
||||
])
|
||||
result = assess_final_report_validator(report, "work_issue")
|
||||
rule_ids = [f["rule_id"] for f in result["findings"]]
|
||||
self.assertNotIn("shared.canonical_comment_post_claim", rule_ids)
|
||||
|
||||
|
||||
class TestEntryPoint(unittest.TestCase):
|
||||
def test_unknown_task_kind_blocks(self):
|
||||
result = assess_final_report_validator("report", "unknown_mode")
|
||||
|
||||
@@ -3087,6 +3087,36 @@ class TestIssueCommentTools(unittest.TestCase):
|
||||
self.assertFalse(result["success"])
|
||||
mock_api.assert_not_called()
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_casual_comment_allowed(self, _auth, mock_api):
|
||||
mock_api.return_value = self._comment(556, "gitea-author", "casual")
|
||||
with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
|
||||
result = gitea_create_issue_comment(
|
||||
issue_number=9,
|
||||
body="Thanks, I will check this.",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertTrue(result["performed"])
|
||||
mock_api.assert_called_once()
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_workflow_comment_missing_fields_blocked(self, _auth, mock_api):
|
||||
with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
|
||||
result = gitea_create_issue_comment(
|
||||
issue_number=9,
|
||||
body="Blocked, author should fix.",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(result["success"])
|
||||
self.assertFalse(result["performed"])
|
||||
self.assertIn("canonical_comment_validation", result)
|
||||
self.assertFalse(result["canonical_comment_validation"]["allowed"])
|
||||
self.assertTrue(result["canonical_comment_validation"]["is_workflow_comment"])
|
||||
mock_api.assert_not_called()
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_missing_issue_error_is_redacted(self, _auth, mock_api):
|
||||
|
||||
Reference in New Issue
Block a user