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.

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 = (
+    "<!-- mcp-review-lease:v1 -->",
+    "<!-- mcp-conflict-fix-lease:v1 -->",
+    "<!-- gitea-issue-claim-heartbeat:v1 -->",
+)
+
+_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<paste-ready prompt>\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<paste-ready prompt>\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 "",
+    }

docs/llm-workflow-runbooks.md

@@ -325,17 +325,6 @@ shared state and manual writes can clobber another session's live lease. Use
 3. Operator override only when explicitly authorized — record
    `External-state mutations` and `operator override proof` in the final report.
 
-**Adoption proof in the live lock response (#477):** when `gitea_lock_issue`
-adopts an existing own branch, the response carries an `adoption` block with
-citable fields — `adoption_decision` (`ADOPT`), `adopted` (`true`),
-`adopted_branch`, `adopted_branch_head`, `matcher_summary` (boundary-safe reason
-the branch qualified), `competing_branch_check`, and `safe_next_action`. A normal
-lock instead returns an `adoption_check` block with `adoption_decision`
-(`NO_MATCH`) and `adopted: false`, so a non-adoption response can never be misread
-as claiming adoption. Recovery reports should quote the live lock response
-`adoption`/`adoption_check` block directly instead of inferring adoption from
-separate offline checks.
-
 `gitea_create_pr` rejects lock files that lack sanctioned `lock_provenance`
 metadata. Final-report validation blocks handoffs that hide lock read/write/delete
 under `External-state mutations: none` or mix author PR creation with reviewer
@@ -492,25 +481,6 @@ Root-level matches are listed in `.gitignore` so they never get committed.
 `gitea_get_runtime_context` and `gitea_lock_issue` surface **warnings** (not
 hard blocks) when these artifacts are still present.
 
-## Capability preflight lifetime (#470)
-
-After `gitea_resolve_task_capability(task=…)` proves the mutation is allowed,
-interleaved **read-only** calls preserve that proof until a gated mutation
-consumes it:
-
-- Safe reads: `gitea_whoami`, `gitea_view_pr`, `gitea_view_issue`, `gitea_list_*`,
-  `gitea_get_runtime_context`, `gitea_check_pr_eligibility`, and related
-  read-only inventory/eligibility tools (see `preflight_contract.py`).
-- Each mutation consumes the proof once; call `gitea_resolve_task_capability`
-  again immediately before the next mutation on the same task.
-- Resolving capability for a **different** task replaces the prior task binding.
-- Workspace edits before resolve, profile switches, or a dirty whoami baseline
-  invalidate proof (fail closed).
-
-If a mutation fails with “capability has not been resolved” or “task mismatch”,
-re-run `gitea_resolve_task_capability(task="<mutation>")` immediately before
-retrying — do not guess or skip the resolve step.
-
 Implementation work and review work must use separate branch folders. For
 example, an implementation branch might live under
 `branches/fix-issue-123-example`, while a review branch for the resulting PR
@@ -779,6 +749,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,

final_report_validator.py

@@ -175,6 +175,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,
@@ -329,6 +346,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,
     *,
@@ -492,42 +546,6 @@ def _rule_reviewer_validation_failure_history(
     ]
 
 
-def _rule_reviewer_validation_cwd_proof(
-    report_text: str,
-    *,
-    validation_session: dict | None = None,
-) -> list[dict[str, str]]:
-    from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report
-
-    session = validation_session or {}
-    claims = (
-        session.get("validation_ran")
-        or session.get("command")
-        or session.get("baseline_validation_ran")
-    )
-    if not claims and "validation command:" not in (report_text or "").lower():
-        return []
-
-    result = assess_validation_cwd_proof_report(
-        report_text,
-        validation_session=session,
-    )
-    if result.get("proven") or not result.get("claims_validation"):
-        return []
-    severity = "block" if result.get("violations") else "downgrade"
-    return [
-        validator_finding(
-            "reviewer.validation_cwd_proof",
-            severity,
-            "Validation cwd/HEAD proof",
-            reason,
-            result.get("safe_next_action")
-            or "document pwd, HEAD SHA, and explicit cwd before validation",
-        )
-        for reason in (result.get("violations") or result.get("reasons") or ["incomplete"])
-    ]
-
-
 def _rule_reviewer_stale_head_proof(report_text: str) -> list[dict[str, str]]:
     from pr_work_lease import assess_reviewer_stale_head_final_report
 
@@ -1048,10 +1066,15 @@ _SHARED_ISSUE_LOCK_RULES = (
     _rule_shared_author_reviewer_same_run,
 )
 
+_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_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_legacy_workspace_mutations,
         _rule_reviewer_vague_mutations_none,
@@ -1059,7 +1082,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_git_fetch_readonly,
         _rule_reviewer_validation_command,
         _rule_reviewer_validation_failure_history,
-        _rule_reviewer_validation_cwd_proof,
         _rule_reviewer_validation_structured,
         _rule_reviewer_linked_issue,
         _rule_reviewer_baseline_on_failure,
@@ -1076,6 +1098,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "reconcile_already_landed": [
         _rule_reconcile_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reconcile_stale_author_fields,
         _rule_reconcile_eligible_reviewed,
@@ -1088,12 +1111,14 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "author_issue": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
     ],
     "work_issue": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
         _rule_conflict_fix_push_proof,
@@ -1101,17 +1126,20 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "issue_filing": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
     "inventory": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reconcile_pagination_proof,
     ],
     "issue_selection": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
 }

gitea_mcp_server.py

@@ -165,7 +165,6 @@ _preflight_capability_called = False
 _preflight_whoami_violation = False
 _preflight_capability_violation = False
 _preflight_resolved_role = None
-_preflight_resolved_task: str | None = None
 _process_start_porcelain: str | None = None
 _preflight_whoami_baseline_porcelain: str | None = None
 _preflight_capability_baseline_porcelain: str | None = None
@@ -380,30 +379,11 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
     }
 
 
-def _clear_preflight_capability_state() -> None:
+def record_preflight_check(type_name: str, resolved_role: str | None = None):
-    """Drop resolved capability proof (consumed by a mutation or fresh resolve)."""
-    global _preflight_capability_called, _preflight_capability_violation
-    global _preflight_resolved_task
-    global _preflight_capability_baseline_porcelain, _preflight_capability_violation_files
-    global _preflight_reviewer_violation_files
-
-    _preflight_capability_called = False
-    _preflight_capability_violation = False
-    _preflight_capability_violation_files = []
-    _preflight_capability_baseline_porcelain = None
-    _preflight_resolved_task = None
-    _preflight_reviewer_violation_files = []
-
-
-def record_preflight_check(
-    type_name: str,
-    resolved_role: str | None = None,
-    resolved_task: str | None = None,
-):
     """Record a pre-flight check (whoami or capability) with session-scoped deltas."""
     global _preflight_whoami_called, _preflight_capability_called
     global _preflight_whoami_violation, _preflight_capability_violation
-    global _preflight_resolved_role, _preflight_resolved_task
+    global _preflight_resolved_role
     global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
     global _preflight_whoami_violation_files, _preflight_capability_violation_files
     global _preflight_reviewer_violation_files
@@ -411,24 +391,13 @@ def record_preflight_check(
     current = _get_workspace_porcelain()
 
     if type_name == "whoami":
-        # Re-evaluate whoami violations instead of replaying sticky state (#252).
+        # Fresh whoami restarts the capability step and re-evaluates violations
-        # Interleaved read-only whoami must not clear a valid capability proof (#469).
+        # instead of replaying a sticky record (#252).
-        saved_capability = None
+        _preflight_capability_called = False
-        preserve_capability = (
+        _preflight_capability_violation = False
-            _preflight_capability_called
+        _preflight_capability_violation_files = []
-            and _preflight_whoami_called
+        _preflight_capability_baseline_porcelain = None
-            and not _preflight_whoami_violation
+        _preflight_reviewer_violation_files = []
-        )
-        if preserve_capability:
-            saved_capability = (
-                _preflight_capability_violation,
-                list(_preflight_capability_violation_files),
-                _preflight_capability_baseline_porcelain,
-                _preflight_resolved_role,
-                _preflight_resolved_task,
-            )
-        else:
-            _clear_preflight_capability_state()
 
         process_start = _ensure_process_start_porcelain()
         whoami_delta = _new_tracked_changes_since(process_start, current)
@@ -436,20 +405,6 @@ def record_preflight_check(
         _preflight_whoami_violation_files = whoami_delta
         _preflight_whoami_baseline_porcelain = current
         _preflight_whoami_called = True
-
-        if (
-            preserve_capability
-            and saved_capability is not None
-            and not whoami_delta
-        ):
-            (
-                _preflight_capability_violation,
-                _preflight_capability_violation_files,
-                _preflight_capability_baseline_porcelain,
-                _preflight_resolved_role,
-                _preflight_resolved_task,
-            ) = saved_capability
-            _preflight_capability_called = True
     elif type_name == "capability":
         baseline = _preflight_whoami_baseline_porcelain or ""
         capability_delta = _new_tracked_changes_since(baseline, current)
@@ -459,18 +414,11 @@ def record_preflight_check(
         _preflight_capability_called = True
         if resolved_role:
             _preflight_resolved_role = resolved_role
-        if resolved_task:
-            _preflight_resolved_task = resolved_task
 
 
 def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
-    """#274: author file/branch mutations must run from a branches/ worktree.
+    """#274: author mutations must run from a branches/ session worktree."""
-
+    if _preflight_resolved_role == "reviewer":
-    Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a
-    Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
-    (#468).
-    """
-    if _preflight_resolved_role in ("reviewer", "reconciler"):
         return
     ctx = _resolve_author_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
@@ -486,11 +434,7 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
         )
 
 
-def verify_preflight_purity(
+def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
-    remote: str | None = None,
-    worktree_path: str | None = None,
-    task: str | None = None,
-):
     """Verify that identity and capability were verified prior to session edits."""
     global _preflight_reviewer_violation_files
 
@@ -507,27 +451,7 @@ def verify_preflight_purity(
         )
     if not _preflight_capability_called:
         raise RuntimeError(
-            preflight_contract.format_missing_capability_error(task)
+            "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
-        )
-    if (
-        task is not None
-        and _preflight_resolved_task is not None
-        and task != _preflight_resolved_task
-    ):
-        raise RuntimeError(
-            preflight_contract.format_task_mismatch_error(
-                _preflight_resolved_task, task
-            )
-        )
-    if (
-        task is not None
-        and _preflight_resolved_task is not None
-        and task != _preflight_resolved_task
-    ):
-        raise RuntimeError(
-            "Pre-flight task mismatch: "
-            f"resolved '{_preflight_resolved_task}' but mutation requires "
-            f"'{task}' (fail closed)"
         )
 
     ctx = _resolve_author_mutation_context(worktree_path)
@@ -584,7 +508,6 @@ def verify_preflight_purity(
                 )
 
     _enforce_branches_only_author_mutation(worktree_path)
-    _clear_preflight_capability_state()
 
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
@@ -616,7 +539,6 @@ import stacked_pr_support  # noqa: E402
 import merge_approval_gate  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
-import preflight_contract  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
@@ -626,6 +548,7 @@ import reconciliation_workflow  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import pr_work_lease  # noqa: E402
 import native_mcp_preference  # noqa: E402
+import canonical_comment_validator as ccv  # noqa: E402
 
 
 # Keyed issue-lock storage (#443): per remote/org/repo/issue files under
@@ -1301,7 +1224,7 @@ def gitea_create_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     base = repo_api_url(h, o, r)
     open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
     closed_issues = api_get_all(
@@ -1434,7 +1357,7 @@ def gitea_lock_issue(
         )
     else:
         git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
-    verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
+    verify_preflight_purity(remote, worktree_path=resolved_worktree)
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -1562,14 +1485,6 @@ def gitea_lock_issue(
             f"Adopted existing branch '{branch_name}' and locked issue "
             f"#{issue_number} for recovery (fail-closed check complete)."
         )
-    else:
-        # #477 AC2: normal (no-adoption) lock responses carry explicit,
-        # adoption-free proof metadata so they stay clear and cannot be
-        # misread as claiming a branch was adopted.
-        result["adoption_check"] = issue_lock_adoption.build_non_adoption_lock_proof(
-            issue_number=issue_number,
-            branch_name=branch_name,
-        )
     if agent_artifacts:
         result["warnings"] = [
             "Agent temp artifacts at repo root (delete before implementation): "
@@ -1658,7 +1573,7 @@ def gitea_create_pr(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
@@ -2674,12 +2589,7 @@ def _list_pr_lease_comments(
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
-    comments = api_request("GET", api, auth)
+    comments = api_request("GET", api, auth) or []
-    # Fail safe to no lease comments when the API returns a non-list payload
-    # (e.g. an error object such as an HTTP 401 body): lease state can only be
-    # proven from real comment entries, never inferred from an error shape (#485).
-    if not isinstance(comments, list):
-        return []
     return list(comments[:limit])
 
 
@@ -2724,7 +2634,7 @@ def _evaluate_pr_review_submission(
     final_review_decision_ready: bool = False,
 ) -> dict:
     """Shared gate chain for live submit and dry-run review tools."""
-    verify_preflight_purity(remote, task="review_pr")
+    verify_preflight_purity(remote)
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
@@ -2837,6 +2747,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(
@@ -3231,12 +3148,12 @@ def gitea_edit_pr(
     if not payload:
         raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
 
-    closing = payload.get("state") == "closed"
+    verify_preflight_purity(remote)
-    verify_preflight_purity(remote, task="close_pr" if closing else None)
 
     # PR closure is a first-class capability, distinct from retitling or
     # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
     # never touches the network.
+    closing = payload.get("state") == "closed"
     if closing:
         gate_reasons = _profile_operation_gate("gitea.pr.close")
         if gate_reasons:
@@ -3481,7 +3398,7 @@ def gitea_commit_files(
             branch="",
         )
 
-    verify_preflight_purity(remote, task="commit_files")
+    verify_preflight_purity(remote)
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
     h, o, r = _resolve(remote, host, org, repo)
@@ -3582,7 +3499,7 @@ def gitea_merge_pr(
         reasons/gates passed or blocked, and merge result / merge commit if
         available. Never secrets.
     """
-    verify_preflight_purity(remote, task="merge_pr")
+    verify_preflight_purity(remote)
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -4111,7 +4028,7 @@ def gitea_delete_branch(
             "permission_report": _permission_block_report("gitea.branch.delete"),
         }
 
-    verify_preflight_purity(remote, task="delete_branch")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     import urllib.parse
@@ -4220,7 +4137,7 @@ def gitea_reconcile_merged_cleanups(
         report["executed"] = False
         return {"success": True, "performed": False, **report}
 
-    verify_preflight_purity(remote, task="reconcile_merged_cleanups")
+    verify_preflight_purity(remote)
     actions: list[dict] = []
     for entry in report.get("entries") or []:
         head_branch = entry.get("head_branch") or ""
@@ -4534,7 +4451,7 @@ def gitea_reconcile_already_landed_pr(
         )
         return result
 
-    verify_preflight_purity(remote, task="reconcile_already_landed_pr")
+    verify_preflight_purity(remote)
 
     if post_comment and comment_body.strip():
         comment_block = _profile_operation_gate("gitea.pr.comment")
@@ -4545,6 +4462,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",
@@ -4637,7 +4560,7 @@ def gitea_close_issue(
         task_capability_map.required_permission("close_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="close_issue")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -5065,7 +4988,7 @@ def gitea_acquire_reviewer_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.comment"),
         }
 
-    verify_preflight_purity(remote, task="review_pr")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -5166,7 +5089,7 @@ def gitea_heartbeat_reviewer_pr_lease(
             ],
         }
 
-    verify_preflight_purity(remote, task="review_pr")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     body = reviewer_pr_lease.format_lease_body(
@@ -5283,12 +5206,7 @@ def gitea_list_issue_comments(
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"
-    comments = api_request("GET", api, auth)
+    comments = api_request("GET", api, auth) or []
-    # Fail safe to no comments when the API returns a non-list payload (e.g. an
-    # error object such as an HTTP 401 body) so listing never crashes on a
-    # malformed response (#485).
-    if not isinstance(comments, list):
-        comments = []
     reveal = _reveal_endpoints()
     out = []
     for c in comments[:limit]:
@@ -5342,7 +5260,7 @@ def gitea_create_issue_comment(
         (permission blocks also carry a structured 'permission_report',
         #142).
     """
-    verify_preflight_purity(remote, task="comment_issue")
+    verify_preflight_purity(remote)
     gate_reasons = _profile_operation_gate("gitea.issue.comment")
     reasons = list(gate_reasons)
     if not (body or "").strip():
@@ -5355,6 +5273,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"
@@ -6806,6 +6733,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,
@@ -6815,8 +6770,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"
@@ -6876,7 +6841,7 @@ def gitea_mark_issue(
         task_capability_map.required_permission("mark_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -6954,7 +6919,7 @@ def gitea_post_heartbeat(
         task_capability_map.required_permission("post_heartbeat"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="post_heartbeat")
+    verify_preflight_purity(remote)
     active_profile = profile or get_profile().get("profile_name")
     body = issue_claim_heartbeat.format_heartbeat_body(
         kind="progress",
@@ -6993,9 +6958,7 @@ def gitea_acquire_conflict_fix_lease(
         task_capability_map.required_permission("comment_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(
+    verify_preflight_purity(remote, worktree_path=worktree_path)
-        remote, worktree_path=worktree_path, task="comment_issue"
-    )
     comments = _list_pr_lease_comments(
         pr_number,
         remote=remote,
@@ -7199,7 +7162,7 @@ def gitea_cleanup_stale_claims(
         task_capability_map.required_permission("cleanup_stale_claims"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="cleanup_stale_claims")
+    verify_preflight_purity(remote)
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -7353,7 +7316,7 @@ def gitea_set_issue_labels(
         task_capability_map.required_permission("set_issue_labels"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="set_issue_labels")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -7590,7 +7553,7 @@ def gitea_resolve_task_capability(
                 "exact_safe_next_action": next_safe_action,
             }
 
-    record_preflight_check("capability", required_role, resolved_task=task)
+    record_preflight_check("capability", required_role)
 
     # Try automatic dispatch switching
     _ensure_matching_profile(required_permission, required_role, remote, host)

issue_lock_adoption.py

@@ -20,43 +20,6 @@ ADOPT = "adopt_existing_branch"
 BLOCK_COMPETING = "block_competing_branch"
 NO_MATCH = "no_matching_branch"
 
-# Citable decision labels aligned with the ``assess_own_branch_adoption``
-# outcomes, surfaced verbatim in the live ``gitea_lock_issue`` response so
-# recovery reports (#473-style) can quote the lock tool output directly
-# instead of inferring adoption from separate offline checks (#477).
-DECISION_LABELS = {
-    ADOPT: "ADOPT",
-    BLOCK_COMPETING: "BLOCK_COMPETING",
-    NO_MATCH: "NO_MATCH",
-}
-
-_SAFE_NEXT_ACTIONS = {
-    ADOPT: (
-        "Own existing branch adopted for lock recovery; proceed to "
-        "gitea_create_pr for this issue and cite this adoption proof."
-    ),
-    BLOCK_COMPETING: (
-        "Competing same-issue branch(es) exist; resolve branch ownership "
-        "before locking. No adoption performed (fail closed)."
-    ),
-    NO_MATCH: (
-        "No existing branch carries this issue marker; normal lock path "
-        "applied. No adoption performed."
-    ),
-}
-
-
-def decision_label(outcome: str) -> str:
-    """Map an ``assess_own_branch_adoption`` outcome to its citable label."""
-    return DECISION_LABELS.get(outcome, "UNKNOWN")
-
-
-def safe_next_action(outcome: str) -> str:
-    """Return the safe next action string for an adoption *outcome*."""
-    return _SAFE_NEXT_ACTIONS.get(
-        outcome, "Unknown adoption outcome; treat as fail closed."
-    )
-
 
 def _branch_name(entry) -> str:
     if isinstance(entry, dict):
@@ -165,42 +128,6 @@ def assess_own_branch_adoption(
     }
 
 
-def _matcher_summary(issue_number: int, assessment: dict) -> str:
-    """Explain, citably, why the assessed branch did or did not qualify.
-
-    Names the numeric word-boundary rule so reports can show that
-    ``issue-42`` was not matched inside ``issue-420`` (#440 / #477 AC3).
-    """
-    outcome = assessment.get("outcome")
-    matched = assessment.get("matched_branch")
-    competing = assessment.get("competing_branches") or []
-    if outcome == ADOPT and matched:
-        return (
-            f"branch '{matched}' exactly matches the issue-{int(issue_number)} "
-            f"marker (numeric word-boundary; 'issue-{int(issue_number)}' is not "
-            f"matched inside 'issue-{int(issue_number)}0')"
-        )
-    if outcome == BLOCK_COMPETING:
-        return (
-            f"competing same-issue branch(es) {competing} carry the "
-            f"issue-{int(issue_number)} marker but are not the requested "
-            f"branch; ownership is ambiguous (fail closed)"
-        )
-    return (
-        f"no existing branch carries the issue-{int(issue_number)} marker "
-        f"under the numeric word-boundary rule"
-    )
-
-
-def _competing_branch_check(assessment: dict) -> dict:
-    """Structured competing-branch verdict for the proof block."""
-    competing = list(assessment.get("competing_branches") or [])
-    return {
-        "result": "blocked" if competing else "clear",
-        "competing_branches": competing,
-    }
-
-
 def build_adoption_proof(
     *,
     issue_number: int,
@@ -216,18 +143,7 @@ def build_adoption_proof(
     Requirement #4: adoption results must carry issue number, branch name,
     branch head commit, adoption reason, no-existing-PR proof, no-competing-
     live-lock proof, and lock file path/status.
-
-    #477: additionally surface explicit, citable adoption-proof fields tied to
-    the ``assess_own_branch_adoption`` outcome (``adoption_decision``,
-    ``adopted``, ``adopted_branch``, ``adopted_branch_head``,
-    ``matcher_summary``, ``competing_branch_check``, ``safe_next_action``) so a
-    recovery session can quote the live lock response directly. The explicit
-    fields are populated for any outcome; ``adopted_branch`` /
-    ``adopted_branch_head`` are set only when the outcome is ADOPT so a
-    non-adoption proof can never be misread as claiming adoption.
     """
-    outcome = assessment.get("outcome")
-    adopted = outcome == ADOPT
     return {
         "issue_number": issue_number,
         "branch_name": branch_name,
@@ -237,36 +153,4 @@ def build_adoption_proof(
         "no_competing_live_lock_proof": bool(competing_lock_checked),
         "lock_file_path": lock_file_path,
         "lock_file_status": lock_file_status,
-        # Explicit citable fields (#477).
-        "adoption_decision": decision_label(outcome),
-        "adopted": adopted,
-        "adopted_branch": branch_name if adopted else None,
-        "adopted_branch_head": assessment.get("matched_head_sha") if adopted else None,
-        "matcher_summary": _matcher_summary(issue_number, assessment),
-        "competing_branch_check": _competing_branch_check(assessment),
-        "safe_next_action": safe_next_action(outcome),
-    }
-
-
-def build_non_adoption_lock_proof(*, issue_number: int, branch_name: str) -> dict:
-    """Safe, adoption-free proof metadata for a normal (NO_MATCH) lock.
-
-    Requirement #477 AC2: non-adoption lock responses must stay clear and must
-    not imply adoption. This returns explicit ``adopted: False`` metadata with
-    the ``NO_MATCH`` decision so a normal lock response can carry citable proof
-    without ever asserting a branch was adopted.
-    """
-    return {
-        "issue_number": issue_number,
-        "branch_name": branch_name,
-        "adoption_decision": DECISION_LABELS[NO_MATCH],
-        "adopted": False,
-        "adopted_branch": None,
-        "adopted_branch_head": None,
-        "matcher_summary": (
-            f"no existing branch carries the issue-{int(issue_number)} marker; "
-            f"normal lock path (no adoption)"
-        ),
-        "competing_branch_check": {"result": "clear", "competing_branches": []},
-        "safe_next_action": safe_next_action(NO_MATCH),
     }

preflight_contract.py

@@ -1,66 +0,0 @@
-"""Capability preflight lifetime contract (#470).
-
-Defines which read-only MCP tools preserve an existing capability proof and the
-canonical sequencing operators must follow before mutations.
-"""
-
-from __future__ import annotations
-
-# Read-only tools that must not invalidate capability proof (#470).
-# gitea_whoami is read-only but re-pins identity; capability is preserved when
-# identity was already verified clean (#469).
-READ_ONLY_PREFLIGHT_TOOLS = frozenset({
-    "gitea_whoami",
-    "gitea_get_authenticated_user",
-    "gitea_get_current_user",
-    "gitea_view_pr",
-    "gitea_view_issue",
-    "gitea_list_prs",
-    "gitea_list_issues",
-    "gitea_list_issue_comments",
-    "gitea_get_runtime_context",
-    "gitea_resolve_task_capability",
-    "gitea_check_pr_eligibility",
-    "gitea_get_pr_review_feedback",
-    "gitea_assess_work_issue_duplicate",
-    "gitea_route_task_session",
-    "gitea_audit_config",
-    "gitea_list_labels",
-    "gitea_get_profile",
-    "gitea_list_profiles",
-})
-
-PREFLIGHT_CONTRACT_SUMMARY = (
-    "Capability preflight is session-scoped per MCP process, profile, and task. "
-    "After gitea_resolve_task_capability(task=...), interleaved read-only calls "
-    "(whoami, view_*, list_*, get_runtime_context, eligibility checks) preserve "
-    "the proof until a gated mutation consumes it. Each mutation consumes the proof "
-    "once; re-resolve immediately before the next mutation. A new resolve for a "
-    "different task replaces the prior task binding. Profile/session changes or "
-    "workspace edits before resolve invalidate proof."
-)
-
-
-def format_missing_capability_error(task: str | None = None) -> str:
-    base = (
-        "Pre-flight order violation: Task capability "
-        "(gitea_resolve_task_capability) has not been resolved (fail closed)"
-    )
-    if task:
-        return (
-            f"{base}. Re-run gitea_resolve_task_capability(task=\"{task}\") "
-            "immediately before this mutation."
-        )
-    return (
-        f"{base}. Re-run gitea_resolve_task_capability for the mutation task "
-        "immediately before acting."
-    )
-
-
-def format_task_mismatch_error(resolved: str, required: str) -> str:
-    return (
-        "Pre-flight task mismatch: "
-        f"resolved '{resolved}' but mutation requires '{required}' (fail closed). "
-        f"Re-run gitea_resolve_task_capability(task=\"{required}\") "
-        "immediately before this mutation."
-    )

review_proofs.py

@@ -5515,13 +5515,6 @@ def assess_validation_failure_history_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
-def assess_validation_cwd_proof_report(report_text, **kwargs):
-    """#398: validation commands require explicit worktree cwd and HEAD proof."""
-    from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report as _assess
-
-    return _assess(report_text, **kwargs)
-
-
 def assess_already_landed_classification_report(report_text, **kwargs):
     """#295: already-landed PRs are reconciliation-only, not review eligible."""
     from reviewer_already_landed_classification import (

reviewer_validation_cwd_proof.py

@@ -1,233 +0,0 @@
-"""Explicit worktree and cwd proof for PR review validation (#398)."""
-
-from __future__ import annotations
-
-import re
-from typing import Any
-
-_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
-
-_PWD_RE = re.compile(
-    r"(?:^|\n)\s*(?:pwd|working\s+directory|cwd)\s*:\s*(\S+)",
-    re.IGNORECASE,
-)
-_HEAD_RE = re.compile(
-    r"(?:git\s+rev-parse\s+head|observed\s+head\s+sha)\s*:\s*([0-9a-f]{7,40})",
-    re.IGNORECASE | re.MULTILINE,
-)
-_EXPECTED_HEAD_RE = re.compile(
-    r"(?:expected\s+(?:pr\s+)?head\s+sha|candidate\s+head\s+sha|pinned\s+head)\s*:\s*([0-9a-f]{7,40})",
-    re.IGNORECASE,
-)
-_STATUS_RE = re.compile(
-    r"git\s+status\s+(?:--short\s+--branch|--short|-sb)\s*:\s*(.+)",
-    re.IGNORECASE,
-)
-_VALIDATION_CMD_RE = re.compile(
-    r"validation\s+command\s*:\s*(.+)",
-    re.IGNORECASE,
-)
-_GIT_C_CMD_RE = re.compile(r"git\s+-C\s+\S+", re.IGNORECASE)
-_CD_CMD_RE = re.compile(r"(?:^|&&\s*)cd\s+\S+", re.IGNORECASE)
-_BASELINE_CWD_RE = re.compile(
-    r"baseline\s+(?:worktree|working\s+directory|cwd)\s*:\s*(\S+)",
-    re.IGNORECASE,
-)
-_BASELINE_SHA_RE = re.compile(
-    r"baseline\s+(?:target\s+)?sha\s*:\s*([0-9a-f]{7,40})",
-    re.IGNORECASE,
-)
-_BASELINE_CMD_RE = re.compile(
-    r"baseline\s+validation\s+command\s*:\s*(.+)",
-    re.IGNORECASE,
-)
-
-
-def _normalize_path(path: str) -> str:
-    return (path or "").replace("\\", "/").rstrip("/")
-
-
-def _path_under_branches(path: str, project_root: str | None = None) -> bool:
-    normalized = _normalize_path(path)
-    if not normalized:
-        return False
-    if "/branches/" in f"{normalized}/":
-        return True
-    if normalized.endswith("/branches"):
-        return True
-    if project_root:
-        root = _normalize_path(project_root)
-        if normalized.startswith(f"{root}/"):
-            rel = normalized[len(root) + 1 :]
-            return rel == "branches" or rel.startswith("branches/")
-    return False
-
-
-def _expand_sha(sha: str) -> str:
-    return (sha or "").strip().lower()
-
-
-def _sha_matches(expected: str, observed: str) -> bool:
-    exp = _expand_sha(expected)
-    obs = _expand_sha(observed)
-    if not exp or not obs:
-        return False
-    if len(exp) == 40 and len(obs) == 40:
-        return exp == obs
-    return obs.startswith(exp) or exp.startswith(obs)
-
-
-def _command_has_explicit_cwd(command: str, cwd: str) -> bool:
-    text = (command or "").strip()
-    if not text:
-        return False
-    if _GIT_C_CMD_RE.search(text):
-        return True
-    if _CD_CMD_RE.search(text):
-        return True
-    if cwd and cwd in text:
-        return True
-    return False
-
-
-def assess_validation_cwd_proof_report(
-    report_text: str,
-    *,
-    validation_session: dict | None = None,
-    project_root: str | None = None,
-) -> dict[str, Any]:
-    """Require cwd/HEAD proof before reviewer validation claims (#398)."""
-    text = report_text or ""
-    session = dict(validation_session or {})
-    reasons: list[str] = []
-    violations: list[str] = []
-
-    claims_validation = bool(
-        session.get("validation_ran")
-        or _VALIDATION_CMD_RE.search(text)
-        or session.get("command")
-    )
-    if not claims_validation:
-        return {
-            "proven": True,
-            "block": False,
-            "claims_validation": False,
-            "reasons": [],
-            "violations": [],
-            "safe_next_action": "proceed",
-        }
-
-    expected_head = (
-        session.get("expected_head_sha")
-        or session.get("candidate_head_sha")
-        or ""
-    ).strip()
-    if not expected_head:
-        match = _EXPECTED_HEAD_RE.search(text)
-        expected_head = (match.group(1) if match else "").strip()
-
-    observed_head = (session.get("observed_head_sha") or "").strip()
-    if not observed_head:
-        match = _HEAD_RE.search(text)
-        observed_head = (match.group(1) if match else "").strip()
-
-    cwd = (
-        session.get("working_directory")
-        or session.get("cwd")
-        or session.get("pwd")
-        or ""
-    ).strip()
-    if not cwd:
-        match = _PWD_RE.search(text)
-        cwd = (match.group(1) if match else "").strip().rstrip(",.;")
-
-    command = (session.get("command") or "").strip()
-    if not command:
-        match = _VALIDATION_CMD_RE.search(text)
-        command = (match.group(1) if match else "").strip().rstrip(".;")
-
-    if not cwd:
-        reasons.append(
-            "validation claimed without pwd/working-directory proof (#398)"
-        )
-    elif not _path_under_branches(cwd, project_root):
-        violations.append(
-            f"validation cwd {cwd!r} is not under branches/ (#398)"
-        )
-        reasons.append(
-            "reviewer validation must run from a branches/ worktree, "
-            "not the main checkout (#398)"
-        )
-
-    if not observed_head:
-        reasons.append(
-            "validation claimed without git rev-parse HEAD / observed HEAD SHA "
-            "proof (#398)"
-        )
-    elif expected_head and not _sha_matches(expected_head, observed_head):
-        violations.append(
-            f"observed HEAD {observed_head} does not match expected "
-            f"PR head {expected_head} (#398)"
-        )
-        reasons.append("validation HEAD SHA must match pinned PR head (#398)")
-
-    if not _STATUS_RE.search(text) and session.get("git_status") is None:
-        reasons.append(
-            "validation claimed without git status --short --branch proof (#398)"
-        )
-
-    if command and cwd and not _command_has_explicit_cwd(command, cwd):
-        if session.get("tool_working_directory") is not True:
-            reasons.append(
-                "validation command must use git -C <worktree>, "
-                "cd <worktree> && ..., or tool-provided cwd metadata (#398)"
-            )
-
-    baseline_ran = bool(
-        session.get("baseline_validation_ran")
-        or _BASELINE_CMD_RE.search(text)
-    )
-    if baseline_ran:
-        baseline_cwd = (session.get("baseline_worktree_path") or "").strip()
-        if not baseline_cwd:
-            match = _BASELINE_CWD_RE.search(text)
-            baseline_cwd = (match.group(1) if match else "").strip().rstrip(",.;")
-        if not baseline_cwd or not _path_under_branches(baseline_cwd, project_root):
-            reasons.append(
-                "baseline validation claimed without baseline worktree cwd "
-                "under branches/ (#398)"
-            )
-        baseline_sha = (session.get("baseline_target_sha") or "").strip()
-        if not baseline_sha:
-            match = _BASELINE_SHA_RE.search(text)
-            baseline_sha = (match.group(1) if match else "").strip()
-        if not baseline_sha:
-            reasons.append(
-                "baseline validation claimed without baseline target SHA (#398)"
-            )
-        baseline_cmd = (session.get("baseline_command") or "").strip()
-        if not baseline_cmd:
-            match = _BASELINE_CMD_RE.search(text)
-            baseline_cmd = (match.group(1) if match else "").strip()
-        if not baseline_cmd:
-            reasons.append(
-                "baseline validation claimed without exact baseline command (#398)"
-            )
-
-    proven = not reasons and not violations
-    return {
-        "proven": proven,
-        "block": bool(violations) or not proven,
-        "claims_validation": True,
-        "expected_head_sha": expected_head or None,
-        "observed_head_sha": observed_head or None,
-        "working_directory": cwd or None,
-        "reasons": reasons,
-        "violations": violations,
-        "safe_next_action": (
-            "before validation record pwd, git rev-parse HEAD, git status, "
-            "expected PR head SHA; run commands with git -C or cd in the same line"
-            if not proven
-            else "proceed"
-        ),
-    }

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -626,28 +626,6 @@ Do not claim “full-suite failures are pre-existing” unless baseline proof is
 
 ## 23. Validation command proof rule
 
-Before any diff, test, or compile validation, record in the same command
-transcript or final report:
-
-* `pwd` or explicit working directory
-* `git rev-parse HEAD`
-* `git status --short --branch`
-* expected PR head SHA (candidate head SHA)
-
-Validation commands must use one of:
-
-* `git -C <review_worktree> ...`
-* `cd <review_worktree> && ...` in the same command
-* tool-provided explicit working-directory metadata
-
-Do not rely on inferred shell cwd from a prior command in a different block.
-
-`gitea_validate_review_final_report` rejects validation claims without
-cwd/HEAD proof when `validation_session` is supplied.
-
-Baseline validation must document baseline worktree path, baseline target SHA,
-cwd proof, exact baseline command, and baseline result using the same rules.
-
 Report the exact validation command as executed.
 
 Report the working directory where validation ran.
@@ -1211,7 +1189,6 @@ Controller Handoff:
 * Files reviewed:
 * Validation:
 * Validation failure history:
-* Validation cwd/HEAD proof:
 * Official validation integrity status:
 * Terminal review mutation:
 * Review decision:

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 = "<!-- 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()

tests/test_final_report_validator.py

@@ -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")

tests/test_issue_lock_adoption.py

@@ -11,7 +11,6 @@ from issue_lock_adoption import (  # noqa: E402
     NO_MATCH,
     assess_own_branch_adoption,
     build_adoption_proof,
-    build_non_adoption_lock_proof,
 )
 
 REQ = "feat/issue-420-server-code-parity"
@@ -135,93 +134,5 @@ class TestBuildAdoptionProof(unittest.TestCase):
         self.assertTrue(proof["no_competing_live_lock_proof"])
 
 
-class TestExplicitAdoptionProofFields(unittest.TestCase):
-    """#477: explicit, citable adoption-proof fields for all outcomes."""
-
-    def _proof(self, assessment, branch):
-        return build_adoption_proof(
-            issue_number=420,
-            branch_name=branch,
-            assessment=assessment,
-            open_pr_checked=True,
-            competing_lock_checked=True,
-            lock_file_path="/tmp/example-lock.json",
-            lock_file_status="written",
-        )
-
-    def test_adopt_proof_exposes_explicit_fields(self):
-        assessment = assess_own_branch_adoption(
-            issue_number=420,
-            requested_branch=REQ,
-            existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
-        )
-        proof = self._proof(assessment, REQ)
-        self.assertEqual(proof["adoption_decision"], "ADOPT")
-        self.assertTrue(proof["adopted"])
-        self.assertEqual(proof["adopted_branch"], REQ)
-        self.assertEqual(proof["adopted_branch_head"], "934688a")
-        self.assertEqual(proof["competing_branch_check"]["result"], "clear")
-        self.assertEqual(proof["competing_branch_check"]["competing_branches"], [])
-        self.assertIn("gitea_create_pr", proof["safe_next_action"])
-        self.assertIn("exactly matches", proof["matcher_summary"])
-
-    def test_block_proof_reports_competing_and_does_not_claim_adoption(self):
-        assessment = assess_own_branch_adoption(
-            issue_number=420,
-            requested_branch=REQ,
-            existing_branches=[{"name": "feat/issue-420-rogue"}],
-        )
-        proof = self._proof(assessment, REQ)
-        self.assertEqual(proof["adoption_decision"], "BLOCK_COMPETING")
-        self.assertFalse(proof["adopted"])
-        self.assertIsNone(proof["adopted_branch"])
-        self.assertIsNone(proof["adopted_branch_head"])
-        self.assertEqual(proof["competing_branch_check"]["result"], "blocked")
-        self.assertIn(
-            "feat/issue-420-rogue",
-            proof["competing_branch_check"]["competing_branches"],
-        )
-        self.assertIn("fail closed", proof["safe_next_action"])
-
-    def test_no_match_proof_does_not_claim_adoption(self):
-        assessment = assess_own_branch_adoption(
-            issue_number=420,
-            requested_branch=REQ,
-            existing_branches=[{"name": "feat/issue-999-unrelated"}],
-        )
-        proof = self._proof(assessment, REQ)
-        self.assertEqual(proof["adoption_decision"], "NO_MATCH")
-        self.assertFalse(proof["adopted"])
-        self.assertIsNone(proof["adopted_branch"])
-        self.assertEqual(proof["competing_branch_check"]["result"], "clear")
-
-    def test_substring_collision_stays_boundary_safe(self):
-        # issue-42 must not adopt/claim against an issue-420 branch (#440/#477).
-        own = "feat/issue-42-widget"
-        assessment = assess_own_branch_adoption(
-            issue_number=42,
-            requested_branch=own,
-            existing_branches=[
-                {"name": own, "commit_sha": "abc1234"},
-                {"name": "feat/issue-420-server-code-parity"},
-            ],
-        )
-        proof = self._proof(assessment, own)
-        self.assertEqual(proof["adoption_decision"], "ADOPT")
-        self.assertEqual(proof["adopted_branch"], own)
-        self.assertEqual(proof["competing_branch_check"]["competing_branches"], [])
-
-    def test_non_adoption_lock_proof_is_adoption_free(self):
-        proof = build_non_adoption_lock_proof(
-            issue_number=196, branch_name="feat/issue-196-mutations"
-        )
-        self.assertEqual(proof["adoption_decision"], "NO_MATCH")
-        self.assertFalse(proof["adopted"])
-        self.assertIsNone(proof["adopted_branch"])
-        self.assertIsNone(proof["adopted_branch_head"])
-        self.assertEqual(proof["competing_branch_check"]["result"], "clear")
-        self.assertIn("no adoption", proof["safe_next_action"].lower())
-
-
 if __name__ == "__main__":
     unittest.main()

tests/test_llm_workflow_split.py

@@ -213,12 +213,6 @@ def test_validation_failure_history_verifier_exported():
     assert callable(assess_validation_failure_history_report)
 
 
-def test_validation_cwd_proof_verifier_exported():
-    from review_proofs import assess_validation_cwd_proof_report
-
-    assert callable(assess_validation_cwd_proof_report)
-
-
 def test_prior_blocker_skip_verifier_exported():
     from review_proofs import assess_prior_blocker_skip_proof
 

tests/test_mcp_server.py

@@ -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):
@@ -3460,45 +3490,6 @@ class TestIssueLocking(unittest.TestCase):
         self.assertIn("adoption", res)
         self.assertEqual(res["adoption"]["branch_head_commit"], "abc123")
 
-    @patch(
-        "mcp_server.issue_lock_worktree.read_worktree_git_state",
-        return_value=_clean_master_git_state_for_lock(),
-    )
-    @patch("mcp_server.api_get_all")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_adoption_response_has_explicit_proof(self, _auth, mock_api, _git_state):
-        # #477 AC1: the live lock response must carry citable adoption proof.
-        branch = "feat/issue-196-mutations"
-        self.mock_dup_fetcher.return_value = ([], [branch], {"status": "not_claimed"})
-        mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}]
-        res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs")
-        proof = res["adoption"]
-        self.assertEqual(proof["adoption_decision"], "ADOPT")
-        self.assertTrue(proof["adopted"])
-        self.assertEqual(proof["adopted_branch"], branch)
-        self.assertEqual(proof["adopted_branch_head"], "abc123")
-        self.assertEqual(proof["competing_branch_check"]["result"], "clear")
-        self.assertIn("gitea_create_pr", proof["safe_next_action"])
-        self.assertIn("196", proof["matcher_summary"])
-
-    @patch(
-        "mcp_server.issue_lock_worktree.read_worktree_git_state",
-        return_value=_clean_master_git_state_for_lock(),
-    )
-    @patch("mcp_server.api_get_all", return_value=[])
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_no_match_response_does_not_claim_adoption(self, _auth, _api, _git_state):
-        # #477 AC2/AC3: a normal (NO_MATCH) lock must carry adoption-free proof
-        # and must NOT expose an ``adoption`` block.
-        res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertTrue(res["success"])
-        self.assertNotIn("adoption", res)
-        check = res["adoption_check"]
-        self.assertEqual(check["adoption_decision"], "NO_MATCH")
-        self.assertFalse(check["adopted"])
-        self.assertIsNone(check["adopted_branch"])
-        self.assertEqual(check["competing_branch_check"]["result"], "clear")
-
     @patch(
         "mcp_server.issue_lock_worktree.read_worktree_git_state",
         return_value=_clean_master_git_state_for_lock(),

tests/test_pr_lease_comments_non_list_guard.py

@@ -1,76 +0,0 @@
-"""Regression tests for non-list API payloads on PR/issue comment listing (#485)."""
-import os
-import sys
-import unittest
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-from mcp_server import (  # noqa: E402
-    _list_pr_lease_comments,
-    gitea_list_issue_comments,
-)
-
-FAKE_AUTH = "Basic dGVzdDp0ZXN0"
-AUTHOR_ENV = {
-    "GITEA_PROFILE_NAME": "gitea-author",
-    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment",
-}
-
-
-class TestPrLeaseCommentsNonListGuard(unittest.TestCase):
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_list_pr_lease_comments_non_list_payload_returns_empty(self, _auth, mock_api):
-        mock_api.return_value = {"message": "Unauthorized"}
-        result = _list_pr_lease_comments(
-            12, remote="prgs", host=None, org=None, repo=None)
-        self.assertEqual(result, [])
-
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_list_pr_lease_comments_none_returns_empty(self, _auth, mock_api):
-        mock_api.return_value = None
-        result = _list_pr_lease_comments(
-            12, remote="prgs", host=None, org=None, repo=None)
-        self.assertEqual(result, [])
-
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_list_pr_lease_comments_list_payload_unchanged(self, _auth, mock_api):
-        comment = {"id": 7, "body": "<!-- mcp-review-lease:v1 -->"}
-        mock_api.return_value = [comment]
-        result = _list_pr_lease_comments(
-            12, remote="prgs", host=None, org=None, repo=None, limit=5)
-        self.assertEqual(result, [comment])
-
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_list_issue_comments_non_list_payload_returns_empty(self, _auth, mock_api):
-        mock_api.return_value = {"message": "Unauthorized"}
-        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
-            result = gitea_list_issue_comments(issue_number=9, remote="prgs")
-        self.assertTrue(result["success"])
-        self.assertEqual(result["comments"], [])
-
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_list_issue_comments_list_payload_unchanged(self, _auth, mock_api):
-        mock_api.return_value = [
-            {
-                "id": 101,
-                "user": {"login": "alice"},
-                "body": "hello",
-                "created_at": "2026-07-03T00:00:00Z",
-                "updated_at": "2026-07-03T01:00:00Z",
-            }
-        ]
-        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
-            result = gitea_list_issue_comments(issue_number=9, remote="prgs")
-        self.assertTrue(result["success"])
-        self.assertEqual(len(result["comments"]), 1)
-        self.assertEqual(result["comments"][0]["author"], "alice")
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_preflight_read_survival.py

@@ -1,159 +0,0 @@
-"""#469/#470: capability preflight lifetime across safe read-only calls."""
-
-import os
-import unittest
-
-import gitea_mcp_server as mcp_server
-
-
-class TestPreflightReadSurvival(unittest.TestCase):
-    def setUp(self):
-        self.orig_whoami = mcp_server._preflight_whoami_called
-        self.orig_capability = mcp_server._preflight_capability_called
-        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
-        self.orig_capability_violation = mcp_server._preflight_capability_violation
-        self.orig_resolved_role = mcp_server._preflight_resolved_role
-        self.orig_resolved_task = mcp_server._preflight_resolved_task
-        self.orig_process_start = mcp_server._process_start_porcelain
-        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
-        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
-        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
-            if key in os.environ:
-                del os.environ[key]
-        os.environ["GITEA_TEST_PORCELAIN"] = ""
-        mcp_server._preflight_whoami_called = False
-        mcp_server._preflight_capability_called = False
-        mcp_server._preflight_whoami_violation = False
-        mcp_server._preflight_capability_violation = False
-        mcp_server._preflight_resolved_role = None
-        mcp_server._preflight_resolved_task = None
-        mcp_server._process_start_porcelain = ""
-        mcp_server._preflight_whoami_baseline_porcelain = None
-        mcp_server._preflight_capability_baseline_porcelain = None
-
-    def tearDown(self):
-        mcp_server._preflight_whoami_called = self.orig_whoami
-        mcp_server._preflight_capability_called = self.orig_capability
-        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
-        mcp_server._preflight_capability_violation = self.orig_capability_violation
-        mcp_server._preflight_resolved_role = self.orig_resolved_role
-        mcp_server._preflight_resolved_task = self.orig_resolved_task
-        mcp_server._process_start_porcelain = self.orig_process_start
-        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
-        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
-        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
-            if key in os.environ:
-                del os.environ[key]
-
-    def test_interleaved_whoami_preserves_capability(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        self.assertTrue(mcp_server._preflight_capability_called)
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-
-        mcp_server.record_preflight_check("whoami")
-        self.assertTrue(mcp_server._preflight_capability_called)
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-
-        mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertFalse(mcp_server._preflight_capability_called)
-
-    def test_missing_capability_still_fails_closed(self):
-        mcp_server.record_preflight_check("whoami")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertIn("has not been resolved", str(ctx.exception))
-
-    def test_task_mismatch_fails_closed(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertIn("task mismatch", str(ctx.exception))
-
-    def test_capability_consumed_after_mutation_gate(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        mcp_server.verify_preflight_purity(task="create_issue")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="create_issue")
-        self.assertIn("has not been resolved", str(ctx.exception))
-
-    def test_whoami_recovery_after_violation_clears_capability(self):
-        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
-        mcp_server.record_preflight_check("whoami")
-        self.assertTrue(mcp_server._preflight_whoami_violation)
-
-        del os.environ["GITEA_TEST_FORCE_DIRTY"]
-        os.environ["GITEA_TEST_PORCELAIN"] = ""
-        mcp_server.record_preflight_check("whoami")
-        self.assertFalse(mcp_server._preflight_whoami_violation)
-        self.assertFalse(mcp_server._preflight_capability_called)
-
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reviewer", resolved_task="review_pr"
-        )
-        mcp_server.verify_preflight_purity(task="review_pr")
-
-    def test_close_pr_sequence_with_interleaved_reads(self):
-        """resolve(close_pr) → whoami/view reads → close_pr gate (#470)."""
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        # Simulate live-state revalidation between resolve and mutation.
-        mcp_server.record_preflight_check("whoami")
-        self.assertTrue(mcp_server._preflight_capability_called)
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-        mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertFalse(mcp_server._preflight_capability_called)
-
-    def test_fresh_capability_resolve_replaces_prior_task(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-        mcp_server.verify_preflight_purity(task="close_pr")
-
-    def test_missing_capability_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        msg = str(ctx.exception)
-        self.assertIn("gitea_resolve_task_capability", msg)
-        self.assertIn('task="close_pr"', msg)
-
-    def test_task_mismatch_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        msg = str(ctx.exception)
-        self.assertIn("task mismatch", msg)
-        self.assertIn('task="close_pr"', msg)
-
-    def test_consumed_capability_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        mcp_server.verify_preflight_purity(task="close_pr")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertIn('task="close_pr"', str(ctx.exception))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_reconciler_close_workspace_guard.py

@@ -1,89 +0,0 @@
-"""Reconciler close_pr must not require author branches/ worktree (#468)."""
-import os
-import sys
-import unittest
-from pathlib import Path
-from unittest.mock import patch
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-import gitea_mcp_server as srv
-
-FAKE_AUTH = "token test"
-CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
-
-RECONCILER_PROFILE = {
-    "profile_name": "prgs-reconciler",
-    "allowed_operations": ["gitea.read", "gitea.pr.close", "gitea.pr.comment"],
-    "forbidden_operations": [
-        "gitea.pr.approve",
-        "gitea.pr.merge",
-        "gitea.pr.review",
-        "gitea.pr.create",
-        "gitea.branch.push",
-        "gitea.repo.commit",
-    ],
-    "audit_label": "prgs-reconciler",
-}
-
-
-class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
-    def setUp(self):
-        srv._preflight_whoami_called = True
-        srv._preflight_capability_called = True
-        srv._preflight_whoami_violation = False
-        srv._preflight_capability_violation = False
-        self._orig_in_test = srv._preflight_in_test_mode
-        srv._preflight_in_test_mode = lambda: False
-
-    def tearDown(self):
-        srv._preflight_in_test_mode = self._orig_in_test
-
-    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
-    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
-    @patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE)
-    @patch("gitea_mcp_server.api_request")
-    def test_reconciler_close_pr_from_control_checkout_succeeds(
-        self, mock_api, _profile, _ns, _auth
-    ):
-        srv._preflight_resolved_role = "reconciler"
-        mock_api.return_value = {
-            "number": 414,
-            "title": "old",
-            "body": "",
-            "state": "closed",
-            "html_url": "https://gitea.example.com/pulls/414",
-        }
-        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
-            with patch.dict(os.environ, {}, clear=False):
-                os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
-                os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
-                result = srv.gitea_edit_pr(414, state="closed", remote="prgs")
-        self.assertTrue(result["success"])
-        self.assertEqual(result["state"], "closed")
-        mock_api.assert_called_once()
-
-    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
-    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
-    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
-    @patch(
-        "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
-        return_value=(True, []),
-    )
-    @patch("gitea_mcp_server.api_get_all", return_value=[])
-    @patch(
-        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
-        return_value={"current_branch": "master"},
-    )
-    def test_author_create_issue_still_blocked_on_control_checkout(
-        self, _git, _get_all, _role, _ns, _prof, _auth
-    ):
-        srv._preflight_resolved_role = "author"
-        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
-            with self.assertRaises(RuntimeError) as ctx:
-                srv.gitea_create_issue(title="Test", body="body")
-        self.assertIn("stable control checkout", str(ctx.exception))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_reviewer_validation_cwd_proof.py

@@ -1,135 +0,0 @@
-"""Tests for validation cwd/HEAD proof verifier (#398)."""
-import sys
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from final_report_validator import assess_final_report_validator  # noqa: E402
-from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report  # noqa: E402
-
-ROOT = "/Users/jasonwalker/Development/Gitea-Tools"
-WORKTREE = f"{ROOT}/branches/review-feat-issue-398"
-HEAD = "f5953549aad5e822f14f52d3ea3c6d7990109384"
-
-
-def _proof_backed_report() -> str:
-    return "\n".join([
-        f"Candidate head SHA: {HEAD}",
-        f"pwd: {WORKTREE}",
-        f"git rev-parse HEAD: {HEAD}",
-        "git status --short --branch: ## feat/issue-398...prgs/master",
-        f"Validation command: cd {WORKTREE} && venv/bin/python -m pytest tests/ -q",
-        "Result: 1497 passed, 6 skipped",
-    ])
-
-
-class TestValidationCwdProof(unittest.TestCase):
-    def test_no_validation_claim_passes(self):
-        result = assess_validation_cwd_proof_report("Review decision: approve")
-        self.assertTrue(result["proven"])
-
-    def test_missing_cwd_proof_fails(self):
-        result = assess_validation_cwd_proof_report(
-            f"Validation command: pytest tests/\nCandidate head SHA: {HEAD}",
-            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
-        )
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["block"])
-
-    def test_main_checkout_cwd_blocks(self):
-        result = assess_validation_cwd_proof_report(
-            "\n".join([
-                f"Candidate head SHA: {HEAD}",
-                f"pwd: {ROOT}",
-                f"git rev-parse HEAD: {HEAD}",
-                "git status --short --branch: ## master",
-                "Validation command: pytest tests/ -q",
-            ]),
-            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
-            project_root=ROOT,
-        )
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["violations"])
-
-    def test_wrong_head_blocks(self):
-        wrong = "a" * 40
-        result = assess_validation_cwd_proof_report(
-            "\n".join([
-                f"Candidate head SHA: {HEAD}",
-                f"pwd: {WORKTREE}",
-                f"git rev-parse HEAD: {wrong}",
-                "git status --short --branch: clean",
-                f"Validation command: cd {WORKTREE} && pytest -q",
-            ]),
-            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
-            project_root=ROOT,
-        )
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["violations"])
-
-    def test_fully_proof_backed_passes(self):
-        result = assess_validation_cwd_proof_report(
-            _proof_backed_report(),
-            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
-            project_root=ROOT,
-        )
-        self.assertTrue(result["proven"], result["reasons"])
-
-    def test_baseline_without_cwd_fails(self):
-        result = assess_validation_cwd_proof_report(
-            "\n".join([
-                _proof_backed_report(),
-                "Baseline validation command: pytest tests/ -q",
-            ]),
-            validation_session={
-                "validation_ran": True,
-                "expected_head_sha": HEAD,
-                "baseline_validation_ran": True,
-            },
-            project_root=ROOT,
-        )
-        self.assertFalse(result["proven"])
-        self.assertTrue(
-            any("baseline" in r.lower() for r in result["reasons"])
-        )
-
-    def test_baseline_with_full_proof_passes(self):
-        result = assess_validation_cwd_proof_report(
-            "\n".join([
-                _proof_backed_report(),
-                f"Baseline worktree: {ROOT}/branches/baseline-master-pr376",
-                f"Baseline target SHA: {HEAD}",
-                f"Baseline validation command: cd {ROOT}/branches/baseline-master-pr376 && pytest -q",
-            ]),
-            validation_session={
-                "validation_ran": True,
-                "expected_head_sha": HEAD,
-                "baseline_validation_ran": True,
-            },
-            project_root=ROOT,
-        )
-        self.assertTrue(result["proven"], result["reasons"])
-
-    def test_final_report_validator_integration(self):
-        result = assess_final_report_validator(
-            "Validation command: pytest tests/ -q",
-            "review_pr",
-            validation_session={"validation_ran": True},
-        )
-        self.assertTrue(result["blocked"] or result["downgraded"])
-        self.assertTrue(
-            any(
-                f.get("rule_id") == "reviewer.validation_cwd_proof"
-                for f in result.get("findings") or []
-            )
-        )
-
-    def test_exported_from_review_proofs(self):
-        from review_proofs import assess_validation_cwd_proof_report as exported
-
-        self.assertTrue(callable(exported))
-
-
-if __name__ == "__main__":
-    unittest.main()