diff --git a/canonical_comment_validator.py b/canonical_comment_validator.py new file mode 100644 index 0000000..a2d05a0 --- /dev/null +++ b/canonical_comment_validator.py @@ -0,0 +1,408 @@ +"""Fail-closed validation for workflow-changing Gitea comments (#496).""" + +from __future__ import annotations + +import re +from typing import Any + +VALID_ROLES = frozenset({ + "controller", + "author", + "reviewer", + "merger", + "reconciler", + "user", +}) + +_BASE_REQUIRED = ("STATE", "WHO_IS_NEXT", "NEXT_ACTION", "NEXT_PROMPT", "WHY") + +_ISSUE_REQUIRED = _BASE_REQUIRED + ("BLOCKERS", "VALIDATION") +_PR_REQUIRED = _BASE_REQUIRED + ( + "ISSUE", + "HEAD_SHA", + "REVIEW_STATUS", + "MERGE_READY", + "BLOCKERS", + "VALIDATION", +) +_SUPERSESSION_REQUIRED = _BASE_REQUIRED + ( + "CANONICAL_ITEM", + "SUPERSEDED_ITEM", + "CLOSE_OR_KEEP_OPEN", +) + +_MACHINE_MARKERS = ( + "", + "", + "", +) + +_WORKFLOW_TRIGGERS = re.compile( + r"\b(?:" + r"blocked|unblocked|ready(?:\s+for\s+(?:review|merge|author))?|" + r"ready-to-merge|approved|approve|request\s+changes|changes\s+requested|" + r"superseded|duplicate|canonical|next\s+action|next\s+actor|who\s+is\s+next|" + r"author\s+should|reviewer\s+should|merger\s+should|reconciler\s+should|" + r"controller\s+should|issue\s+complete|pr\s+open|pr\s+merged|close\s+this|" + r"do\s+not\s+merge|needs?\s+rebase|stale\s+approval|contaminated\s+review|" + r"merge\s+ready|ready\s+for\s+merge" + r")\b", + re.IGNORECASE, +) + +_CANONICAL_HEADINGS = ( + "## Canonical Issue State", + "## Canonical PR State", + "## Canonical Discussion Summary", +) + +_FIELD_RE = re.compile( + r"^([A-Z][A-Z0-9_]*)\s*:\s*(.*)$", + re.MULTILINE, +) + +_VAGUE_NEXT_ACTIONS = frozenset({ + "continue", + "handle this", + "fix it", + "follow up", + "follow-up", + "see above", + "see review", + "tbd", + "todo", + "as needed", + "proceed", + "next steps", + "will check", + "investigate", +}) + +_FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE) +_SHORT_SHA_RE = re.compile(r"\b[0-9a-f]{7,40}\b", re.IGNORECASE) +_PR_REF_RE = re.compile(r"(?:PR\s*#|pull\s*#)\d+|\b#\d{2,}\b", re.IGNORECASE) +_APPROVAL_PROOF_RE = re.compile( + r"\b(?:approved|approval_at_current_head|APPROVE)\b", + re.IGNORECASE, +) +_UNBLOCK_RE = re.compile( + r"\b(?:unblock|until|after|once|when|requires?|must)\b", + re.IGNORECASE, +) + + +def _parse_fields(body: str) -> dict[str, str]: + """Parse KEY: value fields, including values continued on following lines.""" + fields: dict[str, str] = {} + current_key: str | None = None + current_lines: list[str] = [] + + def _flush() -> None: + nonlocal current_key, current_lines + if current_key is not None: + fields[current_key] = "\n".join(current_lines).strip() + current_key = None + current_lines = [] + + for line in (body or "").splitlines(): + if line.startswith("## "): + _flush() + continue + match = re.match(r"^([A-Z][A-Z0-9_]*)\s*:\s*(.*)$", line) + if match: + _flush() + current_key = match.group(1).strip().upper() + rest = match.group(2) + current_lines = [rest] if rest else [] + elif current_key is not None: + current_lines.append(line) + _flush() + return fields + + +def _is_machine_generated(body: str) -> bool: + text = body or "" + return any(marker in text for marker in _MACHINE_MARKERS) + + +def _has_canonical_heading(body: str) -> bool: + return any(h in (body or "") for h in _CANONICAL_HEADINGS) + + +def is_workflow_changing_comment(body: str) -> bool: + """True when comment text implies a workflow/state transition.""" + text = (body or "").strip() + if not text: + return False + if _is_machine_generated(text): + return False + if _has_canonical_heading(text): + return True + if _WORKFLOW_TRIGGERS.search(text): + return True + fields = _parse_fields(text) + if "STATE" in fields or "WHO_IS_NEXT" in fields: + return True + return False + + +def infer_comment_context(body: str, *, explicit: str | None = None) -> str: + if explicit: + return explicit + text = body or "" + if "## Canonical Discussion Summary" in text: + return "discussion_summary" + if "## Canonical PR State" in text: + return "pr_comment" + if "## Canonical Issue State" in text: + return "issue_comment" + fields = _parse_fields(text) + if fields.get("CANONICAL_ITEM") or fields.get("SUPERSEDED_ITEM"): + return "supersession" + if any(k in fields for k in ("HEAD_SHA", "REVIEW_STATUS", "MERGE_READY", "ISSUE")): + return "pr_comment" + return "issue_comment" + + +def _required_fields_for_context(context: str) -> tuple[str, ...]: + if context in ("pr_comment", "pr_review"): + return _PR_REQUIRED + if context == "supersession": + return _SUPERSESSION_REQUIRED + if context == "discussion_summary": + return _BASE_REQUIRED + ("DECISION", "SUBSTANTIVE_COMMENTS") + return _ISSUE_REQUIRED + + +def _is_vague_next_action(value: str) -> bool: + normalized = re.sub(r"\s+", " ", (value or "").strip().lower()) + normalized = normalized.rstrip(".") + if not normalized: + return True + if normalized in _VAGUE_NEXT_ACTIONS: + return True + if len(normalized) < 12: + return True + return False + + +def _next_prompt_ok(value: str) -> bool: + text = (value or "").strip() + if len(text) < 40: + return False + if text.lower() in {"n/a", "none", "tbd", "todo"}: + return False + return True + + +def _state_value(fields: dict[str, str]) -> str: + return (fields.get("STATE") or "").strip().lower() + + +def _suggested_template(context: str) -> str: + if context == "pr_comment" or context == "pr_review": + return ( + "## Canonical PR State\n\n" + "STATE:\n" + "WHO_IS_NEXT:\n" + "NEXT_ACTION:\n" + "NEXT_PROMPT:\n" + "```text\n\n```\n" + "WHAT_HAPPENED:\n" + "WHY:\n" + "ISSUE:\n" + "HEAD_SHA:\n" + "REVIEW_STATUS:\n" + "MERGE_READY:\n" + "BLOCKERS:\n" + "VALIDATION:\n" + "LAST_UPDATED_BY:\n" + ) + if context == "supersession": + return ( + "## Canonical PR State\n\n" + "STATE:\nsuperseded\n" + "WHO_IS_NEXT:\nreconciler\n" + "NEXT_ACTION:\n" + "NEXT_PROMPT:\n" + "WHY:\n" + "CANONICAL_ITEM:\n" + "SUPERSEDED_ITEM:\n" + "CLOSE_OR_KEEP_OPEN:\n" + ) + if context == "discussion_summary": + return ( + "## Canonical Discussion Summary\n\n" + "STATE:\n" + "WHO_IS_NEXT:\n" + "DECISION:\n" + "WHY:\n" + "SUBSTANTIVE_COMMENTS:\n" + "NEXT_ACTION:\n" + "NEXT_PROMPT:\n" + ) + return ( + "## Canonical Issue State\n\n" + "STATE:\n" + "WHO_IS_NEXT:\n" + "NEXT_ACTION:\n" + "NEXT_PROMPT:\n" + "```text\n\n```\n" + "WHAT_HAPPENED:\n" + "WHY:\n" + "RELATED_PRS:\n" + "BLOCKERS:\n" + "VALIDATION:\n" + "LAST_UPDATED_BY:\n" + ) + + +def _build_correction_message( + *, + missing_fields: list[str], + vague_fields: list[str], + extra_reasons: list[str], + context: str, +) -> str: + parts = [ + "Canonical comment validation failed (fail closed before posting).", + ] + if missing_fields: + parts.append("Missing fields: " + ", ".join(missing_fields) + ".") + if vague_fields: + parts.append("Vague or invalid fields: " + ", ".join(vague_fields) + ".") + parts.extend(extra_reasons) + parts.append("Fill the suggested template and retry.") + parts.append("Suggested template:\n" + _suggested_template(context)) + return "\n".join(parts) + + +def assess_canonical_comment( + body: str, + *, + context: str | None = None, + force_workflow: bool = False, +) -> dict[str, Any]: + """Validate outgoing comment text before a Gitea mutation.""" + text = (body or "").strip() + ctx = infer_comment_context(text, explicit=context) + + if not text: + return { + "allowed": True, + "is_workflow_comment": False, + "context": ctx, + "missing_fields": [], + "vague_fields": [], + "correction_message": "", + "suggested_template": "", + } + + if _is_machine_generated(text): + return { + "allowed": True, + "is_workflow_comment": False, + "context": ctx, + "missing_fields": [], + "vague_fields": [], + "correction_message": "", + "suggested_template": "", + } + + workflow = force_workflow or is_workflow_changing_comment(text) + if not workflow: + return { + "allowed": True, + "is_workflow_comment": False, + "context": ctx, + "missing_fields": [], + "vague_fields": [], + "correction_message": "", + "suggested_template": "", + } + + fields = _parse_fields(text) + required = _required_fields_for_context(ctx) + missing = [name for name in required if not (fields.get(name) or "").strip()] + + vague: list[str] = [] + extra: list[str] = [] + + who = (fields.get("WHO_IS_NEXT") or "").strip().lower() + if who and who not in VALID_ROLES: + vague.append("WHO_IS_NEXT") + extra.append( + f"WHO_IS_NEXT must be one of: {', '.join(sorted(VALID_ROLES))}." + ) + + next_action = fields.get("NEXT_ACTION") or "" + if next_action and _is_vague_next_action(next_action): + vague.append("NEXT_ACTION") + + next_prompt = fields.get("NEXT_PROMPT") or "" + if not _next_prompt_ok(next_prompt): + if "NEXT_PROMPT" not in missing: + vague.append("NEXT_PROMPT") + + state = _state_value(fields) + blockers = (fields.get("BLOCKERS") or "").strip() + if "blocked" in state: + if not blockers or blockers.lower() in {"none", "n/a"}: + missing.append("BLOCKERS (unblock condition)") + elif not _UNBLOCK_RE.search(blockers): + vague.append("BLOCKERS") + extra.append("BLOCKED state requires an explicit unblock condition in BLOCKERS.") + + if "superseded" in state: + canon = (fields.get("CANONICAL_ITEM") or fields.get("SUPERSEDED_BY") or "").strip() + superseded = (fields.get("SUPERSEDED_ITEM") or fields.get("SUPERSEDES") or "").strip() + if not canon and "CANONICAL_ITEM" not in missing: + missing.append("CANONICAL_ITEM") + if not superseded and "SUPERSEDED_ITEM" not in missing: + missing.append("SUPERSEDED_ITEM") + + if "ready-to-merge" in state or "ready to merge" in state: + head_sha = fields.get("HEAD_SHA") or "" + merge_ready = fields.get("MERGE_READY") or "" + review_status = fields.get("REVIEW_STATUS") or "" + validation = fields.get("VALIDATION") or "" + proof_blob = " ".join((head_sha, merge_ready, review_status, validation)) + has_sha = bool(_FULL_SHA_RE.search(proof_blob) or _SHORT_SHA_RE.search(proof_blob)) + has_approval = bool(_APPROVAL_PROOF_RE.search(proof_blob)) + if not has_sha or not has_approval: + extra.append( + "ready-to-merge STATE requires approval proof and HEAD_SHA in " + "REVIEW_STATUS, MERGE_READY, HEAD_SHA, or VALIDATION." + ) + + if ctx in ("pr_comment", "pr_review"): + head_sha = (fields.get("HEAD_SHA") or "").strip() + if not head_sha or not _SHORT_SHA_RE.search(head_sha): + if "HEAD_SHA" not in missing: + missing.append("HEAD_SHA") + + if ctx == "issue_comment" and _PR_REF_RE.search(text): + related = (fields.get("RELATED_PRS") or "").strip() + if not related or related.lower() in {"none", "n/a", "-"}: + missing.append("RELATED_PRS") + + allowed = not missing and not vague and not extra + correction = "" + if not allowed: + correction = _build_correction_message( + missing_fields=missing, + vague_fields=vague, + extra_reasons=extra, + context=ctx, + ) + + return { + "allowed": allowed, + "is_workflow_comment": True, + "context": ctx, + "missing_fields": missing, + "vague_fields": vague, + "extra_reasons": extra, + "correction_message": correction, + "suggested_template": _suggested_template(ctx) if not allowed else "", + } \ No newline at end of file diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index e2c1c1a..9a6e97a 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -887,6 +887,55 @@ touched release state names the exact tag/commit and why. Design debates belong in **discussion/RFC issues** (e.g. #100 `profiles.json v2`) — comment on the issue, create no branches/PRs, and end the comment with this handoff. +## Canonical comment validation (#496) + +Workflow-changing issue/PR/review comments must carry durable next-action +state. Casual discussion is still allowed. + +The MCP server runs `canonical_comment_validator.assess_canonical_comment` +**before** posting through: + +- `gitea_create_issue_comment` +- `gitea_submit_pr_review` / `gitea_dry_run_pr_review` (non-empty review bodies) +- `gitea_reconcile_already_landed_pr` when `post_comment=True` +- internal structured comment helpers (machine lease/heartbeat markers stay exempt) + +Detection examples: + +- **Allowed:** `Thanks, I will check this.` +- **Rejected:** `Blocked, author should fix.` (workflow trigger without canonical fields) + +When validation fails, the tool returns `canonical_comment_validation` with +`allowed: false`, `missing_fields`, `vague_fields`, `correction_message`, and +`suggested_template`. **No Gitea API call is made.** + +Minimum workflow comment fields: + +```text +STATE: +WHO_IS_NEXT: +NEXT_ACTION: +NEXT_PROMPT: +WHY: +``` + +`WHO_IS_NEXT` must be one of: `controller`, `author`, `reviewer`, `merger`, +`reconciler`, `user`. + +Issue comments also require `RELATED_PRS`, `BLOCKERS`, and `VALIDATION` when +they mention PR work. PR comments/reviews also require `ISSUE`, `HEAD_SHA`, +`REVIEW_STATUS`, `MERGE_READY`, `BLOCKERS`, and `VALIDATION`. + +Special states: + +- `STATE: blocked` — `BLOCKERS` must name an explicit unblock condition. +- `STATE: superseded` — requires `CANONICAL_ITEM` and `SUPERSEDED_ITEM`. +- `STATE: ready-to-merge` — requires approval proof and head SHA in + `HEAD_SHA`, `REVIEW_STATUS`, `MERGE_READY`, or `VALIDATION`. + +Final reports must not claim a comment was posted when +`canonical_comment_validation.allowed` is false (#496 AC14). + ## Fail-closed behavior Before any mutating action the workflow verifies identity, active profile, diff --git a/final_report_validator.py b/final_report_validator.py index 52e5b65..2ba8910 100644 --- a/final_report_validator.py +++ b/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, ], } diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 261f2b2..dffba70 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -838,6 +838,7 @@ import master_parity_gate # noqa: E402 # Read-only operations are never blocked by staleness. _STARTUP_PARITY = master_parity_gate.capture_startup_parity(PROJECT_ROOT) 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 @@ -3492,6 +3493,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( @@ -5469,6 +5477,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", @@ -6732,6 +6746,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" @@ -8316,6 +8339,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, @@ -8325,8 +8376,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" diff --git a/tests/test_canonical_comment_validator.py b/tests/test_canonical_comment_validator.py new file mode 100644 index 0000000..47e75d8 --- /dev/null +++ b/tests/test_canonical_comment_validator.py @@ -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 = "\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() \ No newline at end of file diff --git a/tests/test_final_report_validator.py b/tests/test_final_report_validator.py index 4701458..0f14fe3 100644 --- a/tests/test_final_report_validator.py +++ b/tests/test_final_report_validator.py @@ -579,6 +579,31 @@ class TestReconcilerCloseProof(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") diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 664b244..80246bb 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -3193,6 +3193,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):