final_report_validator.py

@@ -491,6 +491,45 @@ def _rule_reviewer_validation_failure_history(
     ]
 
 
+def _rule_reviewer_stale_head_proof(report_text: str) -> list[dict[str, str]]:
+    from pr_work_lease import assess_reviewer_stale_head_final_report
+
+    result = assess_reviewer_stale_head_final_report(report_text)
+    if result.get("proven"):
+        return []
+    return _findings_from_reasons(
+        "reviewer.stale_head_proof",
+        result.get("reasons") or [],
+        field="Stale-head proof",
+        severity="block",
+        safe_next_action=(
+            "state reviewed head SHA, live head before approval/merge, and "
+            "whether any push occurred during validation"
+        ),
+    )
+
+
+def _rule_conflict_fix_push_proof(report_text: str) -> list[dict[str, str]]:
+    from pr_work_lease import assess_conflict_fix_final_report
+
+    text = report_text or ""
+    if "conflict-fix" not in text.lower() and "conflict fix" not in text.lower():
+        return []
+    result = assess_conflict_fix_final_report(text)
+    if result.get("proven"):
+        return []
+    return _findings_from_reasons(
+        "author.conflict_fix_push_proof",
+        result.get("reasons") or [],
+        field="Conflict-fix push proof",
+        severity="block",
+        safe_next_action=(
+            "state branch head before/after push, reviewer lease status, "
+            "fast-forward status, and whether any reviewer was active"
+        ),
+    )
+
+
 def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
     text = report_text or ""
     if not _BARE_PYTEST_RE.search(text):
@@ -965,6 +1004,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_target_branch_freshness,
         _rule_reviewer_mutation_ledger,
         _rule_reviewer_review_mutation,
+        _rule_reviewer_stale_head_proof,
     ],
     "reconcile_already_landed": [
         _rule_reconcile_controller_handoff,
@@ -989,6 +1029,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
+        _rule_conflict_fix_push_proof,
     ],
     "issue_filing": [
         _rule_shared_controller_handoff,

gitea_mcp_server.py

@@ -548,6 +548,7 @@ import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
 import review_merge_state_machine  # noqa: E402
+import pr_work_lease  # noqa: E402
 import native_mcp_preference  # noqa: E402
 
 
@@ -2387,6 +2388,50 @@ def gitea_get_pr_review_feedback(
     }
 
 
+def _list_pr_lease_comments(
+    pr_number: int,
+    *,
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+    limit: int = 100,
+) -> list[dict]:
+    """Fetch PR/issue thread comments used for reviewer/conflict-fix leases."""
+    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) or []
+    return list(comments[:limit])
+
+
+def _pr_work_lease_reviewer_block(
+    *,
+    pr_number: int,
+    reviewed_head_sha: str | None,
+    live_head_sha: str | None,
+    mutation: str,
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+) -> dict:
+    comments = _list_pr_lease_comments(
+        pr_number,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+    return pr_work_lease.assess_reviewer_mutation_blocked(
+        pr_number=pr_number,
+        comments=comments,
+        reviewed_head_sha=reviewed_head_sha,
+        live_head_sha=live_head_sha,
+        mutation=mutation,
+    )
+
+
 def _evaluate_pr_review_submission(
     pr_number: int,
     action: str,
@@ -2485,6 +2530,11 @@ def _evaluate_pr_review_submission(
     lock = _load_review_decision_lock() or {}
     if live and lock.get("ready_expected_head_sha"):
         pinned_sha = lock.get("ready_expected_head_sha")
+    if live and not pinned_sha:
+        reasons.append(
+            "reviewed head SHA required before live review mutation (fail closed, #399)"
+        )
+        return result
     if pinned_sha and actual_sha and pinned_sha != actual_sha:
         reasons.append(
             "expected head SHA does not match current PR head (fail closed)"
@@ -2494,6 +2544,21 @@ def _evaluate_pr_review_submission(
         reasons.append("PR head SHA unavailable (fail closed)")
         return result
 
+    lease_block = _pr_work_lease_reviewer_block(
+        pr_number=pr_number,
+        reviewed_head_sha=pinned_sha,
+        live_head_sha=actual_sha,
+        mutation=action,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+    if lease_block.get("block"):
+        reasons.extend(lease_block.get("reasons") or [])
+        result["pr_work_lease"] = lease_block
+        return result
+
     result["would_perform"] = True
     if not live:
         reasons.append(
@@ -2636,6 +2701,39 @@ def gitea_mark_final_review_decision(
                 f"{sorted(_REVIEW_ACTIONS)}"
             ],
         }
+    if not (expected_head_sha or "").strip():
+        return {
+            "marked_ready": False,
+            "reasons": [
+                "expected_head_sha required before marking final review "
+                "decision (fail closed, #399)"
+            ],
+        }
+    elig = gitea_check_pr_eligibility(
+        pr_number=pr_number,
+        action="review",
+        remote=remote,
+        host=None,
+        org=org,
+        repo=repo,
+    )
+    live_head = elig.get("head_sha")
+    lease_block = _pr_work_lease_reviewer_block(
+        pr_number=pr_number,
+        reviewed_head_sha=expected_head_sha,
+        live_head_sha=live_head,
+        mutation="mark_ready",
+        remote=remote,
+        host=None,
+        org=org,
+        repo=repo,
+    )
+    if lease_block.get("block"):
+        return {
+            "marked_ready": False,
+            "reasons": lease_block.get("reasons") or [],
+            "pr_work_lease": lease_block,
+        }
     if action == "request_changes":
         # Duplicate request-changes suppression (#332): an unresolved
         # REQUEST_CHANGES at the current head must not be duplicated.
@@ -3286,8 +3384,27 @@ def gitea_merge_pr(
     if reasons:
         return result
 
-    # Gate 4 — head SHA must match if the caller pinned a reviewed SHA.
+    # Gate 4 — reviewed head SHA is mandatory and must match live PR head (#399).
     actual_sha = result["head_sha"]
+    if not (expected_head_sha or "").strip():
+        reasons.append(
+            "expected_head_sha required before merge (fail closed, #399)"
+        )
+        return result
+    lease_block = _pr_work_lease_reviewer_block(
+        pr_number=pr_number,
+        reviewed_head_sha=expected_head_sha,
+        live_head_sha=actual_sha,
+        mutation="merge",
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+    if lease_block.get("block"):
+        reasons.extend(lease_block.get("reasons") or [])
+        result["pr_work_lease"] = lease_block
+        return result
     if expected_head_sha and actual_sha and expected_head_sha != actual_sha:
         reasons.append(
             "expected head SHA does not match current PR head (fail closed)"
@@ -6570,6 +6687,110 @@ def gitea_post_heartbeat(
     )
 
 
+@mcp.tool()
+def gitea_acquire_conflict_fix_lease(
+    pr_number: int,
+    branch: str,
+    worktree_path: str,
+    head_before: str,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Acquire a conflict-fix lease on a PR branch before pushing (#399)."""
+    blocked = _profile_permission_block(
+        task_capability_map.required_permission("comment_issue"))
+    if blocked:
+        return blocked
+    verify_preflight_purity(remote, worktree_path=worktree_path)
+    comments = _list_pr_lease_comments(
+        pr_number,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+    reviewer_lease = pr_work_lease.find_active_reviewer_lease(
+        comments, pr_number=pr_number)
+    if reviewer_lease:
+        return {
+            "acquired": False,
+            "reasons": [
+                f"active reviewer lease on PR #{pr_number}; cannot acquire "
+                "conflict-fix lease (fail closed)"
+            ],
+            "active_reviewer_lease": reviewer_lease,
+        }
+    profile_name = get_profile().get("profile_name") or "unknown"
+    body = pr_work_lease.format_conflict_fix_lease_body(
+        pr_number=pr_number,
+        branch=branch,
+        worktree=worktree_path,
+        profile=profile_name,
+        head_before=head_before,
+        reviewer_active=bool(reviewer_lease),
+    )
+    posted = _post_structured_issue_comment(
+        issue_number=pr_number,
+        body=body,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+        audit_op="conflict_fix_lease_acquire",
+    )
+    return {
+        "acquired": posted.get("success", False),
+        "pr_number": pr_number,
+        "branch": branch,
+        "worktree_path": worktree_path,
+        "head_before": head_before,
+        "comment_id": posted.get("comment_id"),
+        "active_reviewer_lease": reviewer_lease,
+        "reasons": [] if posted.get("success") else ["lease comment post failed"],
+    }
+
+
+@mcp.tool()
+def gitea_assess_conflict_fix_push(
+    pr_number: int,
+    branch_head_before: str,
+    branch_head_after: str,
+    worktree_path: str,
+    push_cwd: str,
+    is_fast_forward: bool = True,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only pre-push gate for author conflict-fix sessions (#399)."""
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "push_allowed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    comments = _list_pr_lease_comments(
+        pr_number,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+    return pr_work_lease.assess_conflict_fix_push(
+        pr_number=pr_number,
+        comments=comments,
+        branch_head_before=branch_head_before,
+        branch_head_after=branch_head_after,
+        worktree_path=worktree_path,
+        push_cwd=push_cwd,
+        is_fast_forward=is_fast_forward,
+    )
+
+
 @mcp.tool()
 def gitea_reconcile_issue_claims(
     state: str = "open",

pr_work_lease.py

@@ -0,0 +1,482 @@
+"""Conflict-fix and reviewer PR work leases (#399, #407 reader).
+
+Structured PR/issue comments prove exclusive phases so author conflict-fix
+pushes cannot race reviewer validation/approval/merge on the same head.
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+REVIEWER_LEASE_MARKER = "<!-- mcp-review-lease:v1 -->"
+CONFLICT_FIX_LEASE_MARKER = "<!-- mcp-conflict-fix-lease:v1 -->"
+
+_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
+
+_FIELD_RE = re.compile(
+    r"^\s*([a-z_]+)\s*:\s*(.+?)\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+
+_TERMINAL_REVIEWER_PHASES = frozenset({"done", "released", "blocked"})
+_ACTIVE_REVIEWER_PHASES = frozenset({
+    "claimed",
+    "validating",
+    "approved",
+    "request-changes",
+    "merging",
+})
+_TERMINAL_CONFLICT_FIX_PHASES = frozenset({"released", "blocked", "done"})
+_ACTIVE_CONFLICT_FIX_PHASES = frozenset({"claimed", "pushing", "pushed"})
+
+DEFAULT_CONFLICT_FIX_TTL_MINUTES = 120
+DEFAULT_REVIEWER_LEASE_TTL_MINUTES = 120
+
+
+def _parse_timestamp(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    text = value.strip()
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def _normalize_sha(value: str | None) -> str | None:
+    text = (value or "").strip().lower()
+    if not text:
+        return None
+    return text if _FULL_SHA.match(text) else None
+
+
+def _parse_pr_ref(value: str | None) -> int | None:
+    digits = re.sub(r"[^\d]", "", value or "")
+    return int(digits) if digits.isdigit() else None
+
+
+def _parse_marker_comment(body: str, marker: str) -> dict[str, str] | None:
+    text = body or ""
+    if marker not in text:
+        return None
+    fields: dict[str, str] = {}
+    for match in _FIELD_RE.finditer(text):
+        fields[match.group(1).strip().lower()] = match.group(2).strip()
+    return fields or None
+
+
+def parse_reviewer_lease_comment(body: str) -> dict[str, Any] | None:
+    fields = _parse_marker_comment(body, REVIEWER_LEASE_MARKER)
+    if not fields:
+        return None
+    return {
+        "lease_kind": "reviewer",
+        "pr_number": _parse_pr_ref(fields.get("pr")),
+        "issue_number": _parse_pr_ref(fields.get("issue")),
+        "reviewer_identity": fields.get("reviewer_identity"),
+        "profile": fields.get("profile"),
+        "session_id": fields.get("session_id"),
+        "worktree": fields.get("worktree"),
+        "phase": (fields.get("phase") or "").strip().lower() or None,
+        "candidate_head": _normalize_sha(fields.get("candidate_head")),
+        "target_branch": fields.get("target_branch"),
+        "target_branch_sha": _normalize_sha(fields.get("target_branch_sha")),
+        "last_activity": fields.get("last_activity"),
+        "expires_at": fields.get("expires_at"),
+        "blocker": fields.get("blocker"),
+        "raw_fields": fields,
+    }
+
+
+def parse_conflict_fix_lease_comment(body: str) -> dict[str, Any] | None:
+    fields = _parse_marker_comment(body, CONFLICT_FIX_LEASE_MARKER)
+    if not fields:
+        return None
+    ff = (fields.get("fast_forward") or "").strip().lower()
+    reviewer_active = (fields.get("reviewer_active") or "").strip().lower()
+    return {
+        "lease_kind": "conflict_fix",
+        "pr_number": _parse_pr_ref(fields.get("pr")),
+        "branch": fields.get("branch"),
+        "worktree": fields.get("worktree"),
+        "profile": fields.get("profile"),
+        "session_id": fields.get("session_id"),
+        "phase": (fields.get("phase") or "").strip().lower() or None,
+        "head_before": _normalize_sha(fields.get("head_before")),
+        "head_after": _normalize_sha(fields.get("head_after")),
+        "expires_at": fields.get("expires_at"),
+        "reviewer_active": reviewer_active in {"yes", "true", "1"},
+        "fast_forward": ff in {"yes", "true", "1"},
+        "raw_fields": fields,
+    }
+
+
+def _comment_entries(comments: list[dict], *, pr_number: int | None) -> list[dict]:
+    entries: list[dict] = []
+    for comment in comments or []:
+        body = comment.get("body") or ""
+        for parser in (parse_reviewer_lease_comment, parse_conflict_fix_lease_comment):
+            parsed = parser(body)
+            if not parsed:
+                continue
+            if pr_number is not None and parsed.get("pr_number") not in (None, pr_number):
+                continue
+            entries.append({
+                **parsed,
+                "comment_id": comment.get("id"),
+                "author": (comment.get("user") or {}).get("login") or comment.get("author"),
+                "created_at": comment.get("created_at"),
+                "updated_at": comment.get("updated_at"),
+            })
+            break
+    return entries
+
+
+def _lease_expired(lease: dict, *, now: datetime) -> bool:
+    expires_at = _parse_timestamp(lease.get("expires_at"))
+    return bool(expires_at and expires_at <= now)
+
+
+def _lease_phase_active(lease: dict, *, active_phases: frozenset[str]) -> bool:
+    phase = (lease.get("phase") or "").strip().lower()
+    if phase in _TERMINAL_REVIEWER_PHASES or phase in _TERMINAL_CONFLICT_FIX_PHASES:
+        return False
+    return phase in active_phases or bool(phase and phase not in (
+        _TERMINAL_REVIEWER_PHASES | _TERMINAL_CONFLICT_FIX_PHASES
+    ))
+
+
+def find_active_reviewer_lease(
+    comments: list[dict],
+    *,
+    pr_number: int,
+    now: datetime | None = None,
+) -> dict[str, Any] | None:
+    """Return the newest unexpired reviewer lease for *pr_number*, if any."""
+    now = now or datetime.now(timezone.utc)
+    candidates = [
+        entry for entry in _comment_entries(comments, pr_number=pr_number)
+        if entry.get("lease_kind") == "reviewer"
+    ]
+    for lease in reversed(candidates):
+        if _lease_expired(lease, now=now):
+            continue
+        phase = (lease.get("phase") or "").strip().lower()
+        if phase in _TERMINAL_REVIEWER_PHASES:
+            continue
+        if phase in _ACTIVE_REVIEWER_PHASES or phase:
+            return lease
+    return None
+
+
+def find_active_conflict_fix_lease(
+    comments: list[dict],
+    *,
+    pr_number: int,
+    now: datetime | None = None,
+) -> dict[str, Any] | None:
+    """Return the newest unexpired conflict-fix lease for *pr_number*, if any."""
+    now = now or datetime.now(timezone.utc)
+    candidates = [
+        entry for entry in _comment_entries(comments, pr_number=pr_number)
+        if entry.get("lease_kind") == "conflict_fix"
+    ]
+    for lease in reversed(candidates):
+        if _lease_expired(lease, now=now):
+            continue
+        phase = (lease.get("phase") or "").strip().lower()
+        if phase in _TERMINAL_CONFLICT_FIX_PHASES:
+            continue
+        if phase in _ACTIVE_CONFLICT_FIX_PHASES or phase:
+            return lease
+    return None
+
+
+def format_conflict_fix_lease_body(
+    *,
+    pr_number: int,
+    branch: str,
+    worktree: str,
+    profile: str,
+    head_before: str,
+    phase: str = "claimed",
+    session_id: str = "unknown",
+    expires_at: datetime | None = None,
+    reviewer_active: bool = False,
+) -> str:
+    expires = expires_at or (
+        datetime.now(timezone.utc) + timedelta(minutes=DEFAULT_CONFLICT_FIX_TTL_MINUTES)
+    )
+    expires_text = expires.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
+        "+00:00", "Z"
+    )
+    lines = [
+        CONFLICT_FIX_LEASE_MARKER,
+        f"pr: #{pr_number}",
+        f"branch: {branch}",
+        f"worktree: {worktree}",
+        f"profile: {profile}",
+        f"session_id: {session_id}",
+        f"phase: {phase}",
+        f"head_before: {head_before}",
+        f"expires_at: {expires_text}",
+        f"reviewer_active: {'yes' if reviewer_active else 'no'}",
+    ]
+    return "\n".join(lines)
+
+
+def assess_head_sha_equality(
+    reviewed_head_sha: str | None,
+    live_head_sha: str | None,
+) -> dict[str, Any]:
+    """Fail closed when reviewed and live PR heads differ."""
+    reviewed = _normalize_sha(reviewed_head_sha)
+    live = _normalize_sha(live_head_sha)
+    reasons: list[str] = []
+    if not reviewed or not live:
+        reasons.append(
+            "reviewed/live head SHA missing or not full 40-hex; fail closed"
+        )
+    elif reviewed != live:
+        reasons.append(
+            "PR head changed after validation; re-pin and re-validate before "
+            "approval or merge"
+        )
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "reviewed_head_sha": reviewed,
+        "live_head_sha": live,
+        "head_changed": bool(reviewed and live and reviewed != live),
+    }
+
+
+def assess_conflict_fix_push(
+    *,
+    pr_number: int,
+    comments: list[dict],
+    branch_head_before: str | None,
+    branch_head_after: str | None,
+    worktree_path: str | None,
+    push_cwd: str | None,
+    is_fast_forward: bool | None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Author pre-push gate: block when a reviewer holds an active lease."""
+    now = now or datetime.now(timezone.utc)
+    reasons: list[str] = []
+    reviewer_lease = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
+    conflict_lease = find_active_conflict_fix_lease(comments, pr_number=pr_number, now=now)
+
+    if reviewer_lease:
+        reasons.append(
+            f"active reviewer lease on PR #{pr_number} "
+            f"(phase={reviewer_lease.get('phase')}); author push blocked"
+        )
+
+    head_before = _normalize_sha(branch_head_before)
+    head_after = _normalize_sha(branch_head_after)
+    if not head_before:
+        reasons.append("branch head before push missing or invalid SHA")
+    if head_after and head_before and head_before == head_after:
+        reasons.append("branch head unchanged; no push to perform")
+
+    worktree = (worktree_path or "").strip()
+    cwd = (push_cwd or "").strip()
+    if not worktree:
+        reasons.append("worktree path required for conflict-fix push proof")
+    elif cwd and worktree and not cwd.rstrip("/").endswith(worktree.rstrip("/").split("/")[-1]):
+        if worktree not in cwd:
+            reasons.append(
+                f"push cwd '{cwd}' does not match session worktree '{worktree}'"
+            )
+
+    if is_fast_forward is False:
+        reasons.append("non-fast-forward push rejected for conflict-fix (fail closed)")
+
+    if conflict_lease and conflict_lease.get("phase") == "pushing":
+        owner = conflict_lease.get("worktree")
+        if owner and worktree and owner != worktree:
+            reasons.append(
+                f"sibling conflict-fix lease active from worktree '{owner}'"
+            )
+
+    push_allowed = not reasons
+    return {
+        "push_allowed": push_allowed,
+        "block": not push_allowed,
+        "reasons": reasons,
+        "active_reviewer_lease": reviewer_lease,
+        "active_conflict_fix_lease": conflict_lease,
+        "branch_head_before": head_before,
+        "branch_head_after": head_after,
+        "reviewer_was_active": bool(reviewer_lease),
+        "fast_forward": is_fast_forward,
+    }
+
+
+def assess_reviewer_mutation_blocked(
+    *,
+    pr_number: int,
+    comments: list[dict],
+    reviewed_head_sha: str | None,
+    live_head_sha: str | None,
+    mutation: str,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Reviewer gate: block when conflict-fix lease active or head moved."""
+    now = now or datetime.now(timezone.utc)
+    reasons: list[str] = []
+    conflict_lease = find_active_conflict_fix_lease(comments, pr_number=pr_number, now=now)
+    if conflict_lease and (conflict_lease.get("phase") or "") in _ACTIVE_CONFLICT_FIX_PHASES:
+        reasons.append(
+            f"active conflict-fix lease on PR #{pr_number} "
+            f"(phase={conflict_lease.get('phase')}); reviewer {mutation} blocked"
+        )
+
+    head_check = assess_head_sha_equality(reviewed_head_sha, live_head_sha)
+    if head_check["block"]:
+        reasons.extend(head_check["reasons"])
+
+    if not _normalize_sha(reviewed_head_sha):
+        reasons.append(
+            f"reviewed head SHA required before reviewer {mutation} (fail closed)"
+        )
+
+    allowed = not reasons
+    return {
+        "mutation_allowed": allowed,
+        "block": not allowed,
+        "reasons": reasons,
+        "active_conflict_fix_lease": conflict_lease,
+        "head_check": head_check,
+        "reviewed_head_sha": head_check.get("reviewed_head_sha"),
+        "live_head_sha": head_check.get("live_head_sha"),
+        "push_during_validation": bool(
+            conflict_lease and conflict_lease.get("phase") in {"pushing", "pushed"}
+        ),
+    }
+
+
+_REVIEWED_HEAD_RE = re.compile(
+    r"reviewed head sha\s*:\s*([0-9a-f]{40})",
+    re.IGNORECASE,
+)
+_LIVE_HEAD_BEFORE_APPROVAL_RE = re.compile(
+    r"(?:live head sha before approval|final live head sha before approval)\s*:\s*([0-9a-f]{40})",
+    re.IGNORECASE,
+)
+_LIVE_HEAD_BEFORE_MERGE_RE = re.compile(
+    r"(?:live head sha before merge|final live head sha before merge)\s*:\s*([0-9a-f]{40})",
+    re.IGNORECASE,
+)
+_PUSH_DURING_VALIDATION_RE = re.compile(
+    r"push(?:es)? occurred during validation\s*:\s*(yes|no|true|false)",
+    re.IGNORECASE,
+)
+_CONFLICT_HEAD_BEFORE_RE = re.compile(
+    r"branch head before push\s*:\s*([0-9a-f]{40})",
+    re.IGNORECASE,
+)
+_CONFLICT_HEAD_AFTER_RE = re.compile(
+    r"branch head after push\s*:\s*([0-9a-f]{40})",
+    re.IGNORECASE,
+)
+_REVIEWER_LEASE_STATUS_RE = re.compile(
+    r"active reviewer lease status\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_FAST_FORWARD_RE = re.compile(
+    r"whether push was fast-forward\s*:\s*(yes|no|true|false)",
+    re.IGNORECASE,
+)
+_REVIEWER_ACTIVE_RE = re.compile(
+    r"whether any reviewer was active\s*:\s*(yes|no|true|false)",
+    re.IGNORECASE,
+)
+
+
+def assess_reviewer_stale_head_final_report(report_text: str) -> dict[str, Any]:
+    """Final-report proof for reviewed vs live head SHAs (#399 AC 6)."""
+    text = report_text or ""
+    reasons: list[str] = []
+    reviewed = _normalize_sha(_REVIEWED_HEAD_RE.search(text).group(1) if _REVIEWED_HEAD_RE.search(text) else None)
+    live_approval = _normalize_sha(
+        _LIVE_HEAD_BEFORE_APPROVAL_RE.search(text).group(1)
+        if _LIVE_HEAD_BEFORE_APPROVAL_RE.search(text)
+        else None
+    )
+    live_merge = _normalize_sha(
+        _LIVE_HEAD_BEFORE_MERGE_RE.search(text).group(1)
+        if _LIVE_HEAD_BEFORE_MERGE_RE.search(text)
+        else None
+    )
+    push_during = _PUSH_DURING_VALIDATION_RE.search(text)
+
+    if not reviewed:
+        reasons.append("reviewed head SHA not stated in final report")
+    if not live_approval:
+        reasons.append("final live head SHA before approval not stated")
+    if not live_merge:
+        reasons.append("final live head SHA before merge not stated")
+    if not push_during:
+        reasons.append("whether push occurred during validation not stated")
+    elif reviewed and live_approval and reviewed != live_approval:
+        reasons.append("live head before approval differs from reviewed head SHA")
+    elif reviewed and live_merge and reviewed != live_merge:
+        reasons.append("live head before merge differs from reviewed head SHA")
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "reviewed_head_sha": reviewed,
+        "live_head_sha_before_approval": live_approval,
+        "live_head_sha_before_merge": live_merge,
+        "push_during_validation": (push_during.group(1).lower() if push_during else None),
+    }
+
+
+def assess_conflict_fix_final_report(report_text: str) -> dict[str, Any]:
+    """Final-report proof for conflict-fix push sessions (#399 AC 7)."""
+    text = report_text or ""
+    reasons: list[str] = []
+    head_before = _normalize_sha(
+        _CONFLICT_HEAD_BEFORE_RE.search(text).group(1)
+        if _CONFLICT_HEAD_BEFORE_RE.search(text)
+        else None
+    )
+    head_after = _normalize_sha(
+        _CONFLICT_HEAD_AFTER_RE.search(text).group(1)
+        if _CONFLICT_HEAD_AFTER_RE.search(text)
+        else None
+    )
+    if not head_before:
+        reasons.append("branch head before push not stated")
+    if not head_after:
+        reasons.append("branch head after push not stated")
+    if not _REVIEWER_LEASE_STATUS_RE.search(text):
+        reasons.append("active reviewer lease status not stated")
+    if not _FAST_FORWARD_RE.search(text):
+        reasons.append("whether push was fast-forward not stated")
+    if not _REVIEWER_ACTIVE_RE.search(text):
+        reasons.append("whether any reviewer was active not stated")
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "branch_head_before": head_before,
+        "branch_head_after": head_after,
+    }

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

@@ -752,6 +752,24 @@ inventory before continuing.
 Final reports must include lease session id, acquisition proof, heartbeat
 status, and release/blocked status.
 
+## 26C. Conflict-fix lease and stale-head protection (#399)
+
+Before validating, approving, or merging a PR:
+
+1. Check for an active conflict-fix lease on the PR; stop if one is active.
+2. Pin `expected_head_sha` before validation and pass it to
+   `gitea_mark_final_review_decision`, `gitea_submit_pr_review`, and
+   `gitea_merge_pr`.
+3. Re-fetch live PR head immediately before approval and merge; refuse when
+   live head differs from the reviewed SHA.
+
+Final reports must state:
+
+* reviewed head SHA
+* final live head SHA before approval
+* final live head SHA before merge
+* whether any push occurred during validation
+
 ## 27. Merge rules
 
 Before merge, rerun fresh live checks:

skills/llm-project-workflow/workflows/work-issue.md

@@ -607,6 +607,29 @@ After push, report:
 
 If push fails, stop and produce a recovery handoff.
 
+## 20A. Conflict-fix lease and push gate (#399)
+
+When pushing to an existing PR branch to resolve merge conflicts:
+
+1. Call `gitea_acquire_conflict_fix_lease` before any push.
+2. Call `gitea_assess_conflict_fix_push` immediately before `git push` with:
+   * branch head before push
+   * branch head after push (local)
+   * session worktree path
+   * push cwd
+   * whether the push is fast-forward
+3. Do not push when a reviewer holds an active lease on the same PR.
+4. Do not force-push.
+5. Do not push from the main checkout or wrong cwd.
+
+Conflict-fix final reports must state:
+
+* branch head before push
+* branch head after push
+* active reviewer lease status
+* whether push was fast-forward
+* whether any reviewer was active
+
 ## 21. PR creation rules
 
 Create a PR only if implementation and validation pass, unless project policy explicitly allows draft PRs with documented validation failures.

tests/test_audit.py

@@ -284,21 +284,38 @@ class TestSimpleToolAudit(_AuditWiringBase):
         self.assertEqual(result["number"], 9)
 
 
+_NO_PR_WORK_LEASE_BLOCK = {"block": False, "reasons": [], "mutation_allowed": True}
+
+
 class TestGatedToolAudit(_AuditWiringBase):
 
     def setUp(self):
         super().setUp()
+        from mcp_server import init_review_decision_lock
         from tests.test_mcp_server import _install_owned_reviewer_lease
         import reviewer_pr_lease
 
+        # init_review_decision_lock clears any prior session lease (#407).
+        init_review_decision_lock("prgs", "review_pr")
         self._lease_patch = _install_owned_reviewer_lease(8)
         self._lease_patch.start()
         self._auth_identity_patch = patch(
             "mcp_server._authenticated_username", return_value="reviewer-bot"
         )
         self._auth_identity_patch.start()
+        self._pr_lease_comments_patch = patch(
+            "mcp_server._list_pr_lease_comments", return_value=[]
+        )
+        self._pr_lease_comments_patch.start()
+        self._pr_work_lease_patch = patch(
+            "mcp_server._pr_work_lease_reviewer_block",
+            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
+        )
+        self._pr_work_lease_patch.start()
         self.addCleanup(self._auth_identity_patch.stop)
         self.addCleanup(self._lease_patch.stop)
+        self.addCleanup(self._pr_lease_comments_patch.stop)
+        self.addCleanup(self._pr_work_lease_patch.stop)
         self.addCleanup(reviewer_pr_lease.clear_session_lease)
 
     def _pr(self, author, state="open", sha="abc123", mergeable=True):
@@ -349,7 +366,9 @@ class TestGatedToolAudit(_AuditWiringBase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_submit_review_success_audited(self, _auth, mock_api):
+        # mark_final_review_decision and submit each run eligibility (user + PR).
         mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
             {"login": "reviewer-bot"}, self._pr("author-bot"),
             {"id": 7, "state": "APPROVED"},
             [{"id": 7, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
@@ -358,23 +377,16 @@ class TestGatedToolAudit(_AuditWiringBase):
         env = self._env(GITEA_PROFILE_NAME="gitea-reviewer",
                         GITEA_ALLOWED_OPERATIONS="read,review,approve")
         with patch.dict(os.environ, env, clear=True):
-            from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
-            from tests.test_mcp_server import _install_owned_reviewer_lease
-            import reviewer_pr_lease
+            from mcp_server import gitea_mark_final_review_decision
 
-            init_review_decision_lock("prgs", "review_pr")
-            gitea_mark_final_review_decision(8, "approve", remote="prgs")
-            lease_patch = _install_owned_reviewer_lease(8)
-            lease_patch.start()
-            try:
-                r = gitea_submit_pr_review(
-                    pr_number=8, action="approve",
-                    body="LGTM", remote="prgs",
-                    final_review_decision_ready=True,
-                )
-            finally:
-                lease_patch.stop()
-                reviewer_pr_lease.clear_session_lease()
+            gitea_mark_final_review_decision(
+                8, "approve", expected_head_sha="abc123", remote="prgs",
+            )
+            r = gitea_submit_pr_review(
+                pr_number=8, action="approve",
+                body="LGTM", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertTrue(r["performed"])
         recs = self._records()
         self.assertEqual(len(recs), 1)

tests/test_final_report_validator.py

@@ -41,6 +41,10 @@ def _review_handoff(**overrides):
         "- Selected PR: #203",
         "- Reviewer eligibility: eligible",
         "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Final live head SHA before approval: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Final live head SHA before merge: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Push occurred during validation: no",
         "- Worktree path: branches/review-203",
         "- Worktree dirty: clean",
         "- Scratch worktree used: yes (branches/review-203)",

tests/test_llm_agent_sha.py

@@ -122,15 +122,19 @@ class TestShaCannotBypassSelfReview(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_review_tool_refuses_self_approval_despite_sha(self, _auth, mock_api, mock_get_all):
-        mock_get_all.return_value = [{"number": 9, "title": "PR 9", "state": "open", "head": {"ref": "branch9", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
+        from mcp_server import init_review_decision_lock
+        from tests.test_mcp_server import FULL_HEAD_SHA, _seed_ready_review_decision
+
+        head_sha = FULL_HEAD_SHA
+        mock_get_all.return_value = [{"number": 9, "title": "PR 9", "state": "open", "head": {"ref": "branch9", "sha": head_sha}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
         mock_api.side_effect = [
             {"login": "jcwalker3"}, # /user (inventory)
             {"login": "jcwalker3"}, # /user (submit eligibility)
-            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/9
+            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": head_sha}, "mergeable": True}, # /pulls/9
         ]
-        from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
         init_review_decision_lock("prgs", "review_pr")
-        gitea_mark_final_review_decision(9, "approve", remote="prgs")
+        with patch("mcp_server._list_pr_lease_comments", return_value=[]):
+            _seed_ready_review_decision(9, "approve", sha=head_sha, remote="prgs")
         env = self._env(SHA_WOULD_BE_REVIEWER, "reviewer")
         with patch.dict(os.environ, env, clear=True):
             r = gitea_review_pr(

tests/test_mcp_server.py

@@ -47,6 +47,7 @@ import gitea_config  # noqa: E402
 import mcp_server
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+FULL_HEAD_SHA = "a" * 40
 
 _NO_BLOCKER_FEEDBACK = {
     "success": True,
@@ -60,6 +61,7 @@ def _mark_request_changes_ready(pr_number=8, **kwargs):
     suppression feedback fetch stubbed to 'no existing blocker'."""
     with patch("mcp_server.gitea_get_pr_review_feedback",
                return_value=dict(_NO_BLOCKER_FEEDBACK)):
+        kwargs.setdefault("expected_head_sha", "abc123")
         return gitea_mark_final_review_decision(
             pr_number, "request_changes", **kwargs)
 
@@ -80,6 +82,7 @@ def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
 
 
 _DEFAULT_LEASE_SESSION = "mcp-test-reviewer-lease"
+_NO_PR_WORK_LEASE_BLOCK = {"block": False, "reasons": [], "mutation_allowed": True}
 
 
 def _reviewer_lease_comment(
@@ -137,6 +140,45 @@ def _install_owned_reviewer_lease(
     )
 
 
+def _seed_ready_review_decision(
+    pr_number,
+    action,
+    *,
+    sha=FULL_HEAD_SHA,
+    remote="prgs",
+    org=None,
+    repo=None,
+):
+    """Mark the review-decision lock ready without mark_final API calls (#399)."""
+    import mcp_server as _m
+
+    resolved_org, resolved_repo = org, repo
+    if remote in _m.REMOTES:
+        _, resolved_org, resolved_repo = _m._resolve(remote, None, org, repo)
+    profile_name = (_m.get_profile().get("profile_name") or "").strip()
+    session_lock = (
+        (os.environ.get(_m.SESSION_PROFILE_LOCK_ENV) or "").strip()
+        or profile_name
+    )
+    _m._save_review_decision_lock({
+        "task": "review_pr",
+        "remote": remote,
+        "session_pid": os.getpid(),
+        "session_profile": profile_name,
+        "session_profile_lock": session_lock,
+        "final_review_decision_ready": True,
+        "ready_pr_number": pr_number,
+        "ready_action": action,
+        "ready_expected_head_sha": sha,
+        "ready_remote": remote,
+        "ready_org": resolved_org,
+        "ready_repo": resolved_repo,
+        "live_mutations": [],
+        "correction_authorized": False,
+        "correction_reason": None,
+    })
+
+
 # Issue-write tools are profile-gated (#69).
 ISSUE_WRITE_ENV = {
     "GITEA_ALLOWED_OPERATIONS": (
@@ -609,6 +651,17 @@ class TestMergePR(unittest.TestCase):
         self.addCleanup(self._auth_identity_patch.stop)
         self.addCleanup(self._lease_patch.stop)
         self.addCleanup(reviewer_pr_lease.clear_session_lease)
+        self._pr_lease_comments_patch = patch(
+            "mcp_server._list_pr_lease_comments", return_value=[]
+        )
+        self._pr_lease_comments_patch.start()
+        self.addCleanup(self._pr_lease_comments_patch.stop)
+        self._pr_work_lease_patch = patch(
+            "mcp_server._pr_work_lease_reviewer_block",
+            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
+        )
+        self._pr_work_lease_patch.start()
+        self.addCleanup(self._pr_work_lease_patch.stop)
 
     def _pr(self, author, state="open", sha="abc123", mergeable=True):
         return {
@@ -682,7 +735,8 @@ class TestMergePR(unittest.TestCase):
         with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
-                expected_changed_files=["b.py", "a.py"], remote="prgs")
+                expected_changed_files=["b.py", "a.py"],
+                expected_head_sha="abc123", remote="prgs")
         self.assertTrue(r["performed"])
 
     # -- read-back / cleanup surfacing (#98) -----------------------------------
@@ -700,8 +754,10 @@ class TestMergePR(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_merge_pr(pr_number=8, confirmation=self._confirm(8),
-                               remote="prgs")
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8),
+                expected_head_sha="abc123", remote="prgs",
+            )
         # The merge itself is still reported performed/successful.
         self.assertTrue(r["performed"])
         self.assertEqual(r["merge_result"], "PR #8 merged via 'merge'.")
@@ -729,8 +785,10 @@ class TestMergePR(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_merge_pr(pr_number=8, confirmation=self._confirm(8),
-                               remote="prgs")
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8),
+                expected_head_sha="abc123", remote="prgs",
+            )
         self.assertTrue(r["performed"])
         self.assertEqual(r["merge_commit"], "c9")
         self.assertTrue(r["cleanup_status"].startswith("skipped (cleanup error:"))
@@ -744,7 +802,7 @@ class TestMergePR(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_merge_pr(pr_number=8, confirmation="", remote="prgs")
+            r = gitea_merge_pr(pr_number=8, confirmation="", expected_head_sha="abc123", remote="prgs")
         self.assertFalse(r["performed"])
         self.assertTrue(any("explicit confirmation required" in x for x in r["reasons"]))
         mock_api.assert_not_called()
@@ -755,7 +813,7 @@ class TestMergePR(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 9", remote="prgs")
+            r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 9", expected_head_sha="abc123", remote="prgs")
         self.assertFalse(r["performed"])
         mock_api.assert_not_called()
 
@@ -889,7 +947,8 @@ class TestMergePR(unittest.TestCase):
         with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
-                expected_changed_files=["a.py", "b.py"], remote="prgs")
+                expected_changed_files=["a.py", "b.py"],
+                expected_head_sha="abc123", remote="prgs")
         self.assertFalse(r["performed"])
         self.assertIn(
             "PR changed files do not match expected_changed_files (fail closed)",
@@ -902,7 +961,7 @@ class TestMergePR(unittest.TestCase):
     def test_invalid_merge_method_rejected(self, mock_api):
         with patch.dict(os.environ, {}, clear=True):
             r = gitea_merge_pr(
-                pr_number=8, confirmation="MERGE PR 8", do="octopus", remote="prgs")
+                pr_number=8, confirmation="MERGE PR 8", do="octopus", expected_head_sha="abc123", remote="prgs")
         self.assertFalse(r["performed"])
         self.assertTrue(any("unknown merge method" in x for x in r["reasons"]))
         mock_api.assert_not_called()
@@ -920,7 +979,9 @@ class TestMergePR(unittest.TestCase):
                "GITEA_TOKEN": "super-secret-token"}
         with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
-                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+                pr_number=8, confirmation=self._confirm(8),
+                expected_head_sha="abc123", remote="prgs",
+            )
         blob = repr(r).lower()
         for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
             self.assertNotIn(secret, blob)
@@ -937,7 +998,9 @@ class TestMergePR(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
         with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
-                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+                pr_number=8, confirmation=self._confirm(8),
+                expected_head_sha="abc123", remote="prgs",
+            )
         self.assertFalse(r["performed"])
         blob = repr(r)
         self.assertIn("[REDACTED]", blob)
@@ -955,7 +1018,9 @@ class TestMergePR(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
         with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
-                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+                pr_number=8, confirmation=self._confirm(8),
+                expected_head_sha="abc123", remote="prgs",
+            )
         self.assertFalse(r["performed"])
         self.assertFalse(r.get("approval_visible"))
         self.assertTrue(any("no visible APPROVED review" in x for x in r["reasons"]))
@@ -976,7 +1041,9 @@ class TestMergePR(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
         with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
-                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+                pr_number=8, confirmation=self._confirm(8),
+                expected_head_sha="abc123", remote="prgs",
+            )
         self.assertFalse(r["performed"])
         self.assertTrue(r.get("has_blocking_change_requests"))
         self.assertTrue(any("REQUEST_CHANGES" in x for x in r["reasons"]))
@@ -1077,16 +1144,18 @@ class TestReviewPR(unittest.TestCase):
             "forbidden_operations": [],
             "base_url": None,
         }
-        mock_get_all.return_value = [{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
+        head_sha = FULL_HEAD_SHA
+        mock_get_all.return_value = [{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": head_sha}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
         # mock_api responses: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility)
         mock_api.side_effect = [
             {"login": "jcwalker3"}, # /api/v1/user (inventory)
             {"login": "jcwalker3"}, # /api/v1/user (submit eligibility)
-            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/1
+            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": head_sha}, "mergeable": True}, # /pulls/1
         ]
         from mcp_server import init_review_decision_lock
         init_review_decision_lock("prgs", "review_pr")
-        gitea_mark_final_review_decision(1, "approve", remote="prgs")
+        with patch("mcp_server._list_pr_lease_comments", return_value=[]):
+            _seed_ready_review_decision(1, "approve", sha=head_sha, remote="prgs")
         result = gitea_review_pr(
             pr_number=1,
             event="APPROVE",
@@ -1742,7 +1811,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
     """Block incidental live review mutations during validation."""
 
     PR = 203
-    SHA = "abc123"
+    SHA = FULL_HEAD_SHA
 
     def _pr(self, author, sha=SHA):
         return {
@@ -1767,6 +1836,18 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
         self.addCleanup(self._auth_identity_patch.stop)
         self.addCleanup(self._lease_patch.stop)
         self.addCleanup(reviewer_pr_lease.clear_session_lease)
+        self._pr_lease_comments_patch = patch(
+            "mcp_server._list_pr_lease_comments", return_value=[]
+        )
+        self._pr_lease_comments_patch.start()
+        self._pr_work_lease_patch = patch(
+            "mcp_server._pr_work_lease_reviewer_block",
+            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
+        )
+        self._pr_work_lease_patch.start()
+        self.addCleanup(self._pr_lease_comments_patch.stop)
+        self.addCleanup(self._pr_work_lease_patch.stop)
+
 
     def _env(self):
         return patch.dict(os.environ, {
@@ -1864,15 +1945,28 @@ class TestSubmitPrReview(unittest.TestCase):
         import reviewer_pr_lease
 
         init_review_decision_lock("prgs", "review_pr")
-        gitea_mark_final_review_decision(8, "approve", remote="prgs")
         self._lease_patch = _install_owned_reviewer_lease(8)
         self._lease_patch.start()
         self._auth_identity_patch = patch(
             "mcp_server._authenticated_username", return_value="reviewer-bot"
         )
         self._auth_identity_patch.start()
+        self._pr_lease_comments_patch = patch(
+            "mcp_server._list_pr_lease_comments", return_value=[]
+        )
+        self._pr_lease_comments_patch.start()
+        self._pr_work_lease_patch = patch(
+            "mcp_server._pr_work_lease_reviewer_block",
+            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
+        )
+        self._pr_work_lease_patch.start()
+        gitea_mark_final_review_decision(
+            8, "approve", remote="prgs", expected_head_sha="abc123",
+        )
         self.addCleanup(self._auth_identity_patch.stop)
         self.addCleanup(self._lease_patch.stop)
+        self.addCleanup(self._pr_lease_comments_patch.stop)
+        self.addCleanup(self._pr_work_lease_patch.stop)
         self.addCleanup(reviewer_pr_lease.clear_session_lease)
 
     def _pr(self, author, state="open", sha="abc123", mergeable=True):
@@ -2033,10 +2127,11 @@ class TestSubmitPrReview(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_comment_succeeds_when_review_eligible(self, _auth, mock_api):
-        gitea_mark_final_review_decision(8, "comment", remote="prgs")
         mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
             {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 3},
         ]
+        gitea_mark_final_review_decision(8, "comment", remote="prgs", expected_head_sha="abc123")
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review"}
         with patch.dict(os.environ, env, clear=True):
@@ -2052,12 +2147,15 @@ class TestSubmitPrReview(unittest.TestCase):
         with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
                 patch("mcp_server.api_request") as mock_api:
             mock_api.side_effect = [
+                {"login": "jcwalker3"}, self._pr("jcwalker3"),
                 {"login": "jcwalker3"}, self._pr("jcwalker3"), {"id": 4},
             ]
             env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                    "GITEA_ALLOWED_OPERATIONS": "read,review"}
             with patch.dict(os.environ, env, clear=True):
-                gitea_mark_final_review_decision(8, "comment", remote="prgs")
+                gitea_mark_final_review_decision(
+                    8, "comment", remote="prgs", expected_head_sha="abc123",
+                )
                 r = gitea_submit_pr_review(
                     pr_number=8, action="comment", body="note", remote="prgs",
                     final_review_decision_ready=True,
@@ -2100,14 +2198,14 @@ class TestSubmitPrReview(unittest.TestCase):
         with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
                 patch("mcp_server.api_request") as mock_api:
             mock_api.side_effect = [
-                {"login": "reviewer-bot"}, self._pr("author-bot", sha="abc123"),
+                {"login": "reviewer-bot"}, self._pr("author-bot", sha="deadbeef"),
             ]
             env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                    "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
             with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
                     pr_number=8, action="approve",
-                    expected_head_sha="deadbeef", remote="prgs",
+                    expected_head_sha="abc123", remote="prgs",
                     final_review_decision_ready=True,
                 )
             self.assertFalse(r["performed"])
@@ -2158,7 +2256,7 @@ class TestSubmitPrReview(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve",
                "GITEA_TOKEN": "super-secret-token"}
         with patch.dict(os.environ, env, clear=True):
-            gitea_mark_final_review_decision(5, "approve", remote="prgs")
+            gitea_mark_final_review_decision(5, "approve", remote="prgs", expected_head_sha="abc123")
             r = gitea_submit_pr_review(
                 pr_number=5, action="approve", remote="prgs",
                 final_review_decision_ready=True,
@@ -2178,7 +2276,6 @@ class TestSubmitPrReview(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
-            gitea_mark_final_review_decision(8, "approve", remote="prgs")
             r = gitea_submit_pr_review(
                 pr_number=8, action="approve", remote="prgs",
                 final_review_decision_ready=True,
@@ -2268,7 +2365,7 @@ class TestSubmitPrReview(unittest.TestCase):
 
     def test_mark_final_decision_rejects_remote_mismatch(self):
         init_review_decision_lock("prgs", "review_pr")
-        r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools")
+        r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools", expected_head_sha="abc123")
         self.assertFalse(r["marked_ready"])
         self.assertTrue(any("does not match locked remote" in x for x in r["reasons"]))
 
@@ -2280,28 +2377,30 @@ class TestSubmitPrReview(unittest.TestCase):
             {"id": 42, "state": "APPROVED"},
             [_formal_review("reviewer-bot", "APPROVED", review_id=42)],
             {"login": "reviewer-bot"}, self._pr("author-bot"),
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
             {"id": 43, "state": "REQUEST_CHANGES"},
             [_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=43)],
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
-        gitea_mark_final_review_decision(8, "approve", remote="prgs")
-        with patch.dict(os.environ, env, clear=True):
-            first = gitea_submit_pr_review(
-                pr_number=8, action="approve", remote="prgs",
-                final_review_decision_ready=True,
-            )
-            auth = gitea_authorize_review_correction(
-                prior_review_id=42,
-                prior_review_state="approve",
-                reason="operator approved correcting mistaken approve",
-                operator_authorized=True,
-            )
-            _mark_request_changes_ready(remote="prgs")
-            second = gitea_submit_pr_review(
-                pr_number=8, action="request_changes", remote="prgs",
-                final_review_decision_ready=True,
-            )
+        with patch("mcp_server.gitea_get_pr_review_feedback",
+                   return_value=dict(_NO_BLOCKER_FEEDBACK)):
+            with patch.dict(os.environ, env, clear=True):
+                first = gitea_submit_pr_review(
+                    pr_number=8, action="approve", remote="prgs",
+                    final_review_decision_ready=True,
+                )
+                auth = gitea_authorize_review_correction(
+                    prior_review_id=42,
+                    prior_review_state="approve",
+                    reason="operator approved correcting mistaken approve",
+                    operator_authorized=True,
+                )
+                _mark_request_changes_ready(remote="prgs")
+                second = gitea_submit_pr_review(
+                    pr_number=8, action="request_changes", remote="prgs",
+                    final_review_decision_ready=True,
+                )
         self.assertTrue(first["performed"])
         self.assertTrue(auth["authorized"])
         self.assertTrue(second["performed"])
@@ -2313,7 +2412,7 @@ class TestSubmitPrReview(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
             # Mark decision for specific remote, org, repo
-            gitea_mark_final_review_decision(8, "approve", remote="prgs", org="MyOrg", repo="MyRepo")
+            gitea_mark_final_review_decision(8, "approve", remote="prgs", expected_head_sha="abc123", org="MyOrg", repo="MyRepo")
 
             # Mismatched remote
             r = gitea_submit_pr_review(
@@ -2355,6 +2454,11 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         self.mock_audit = patch("gitea_audit.write_event").start()
         # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
         patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["read", "merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
+        patch("mcp_server._list_pr_lease_comments", return_value=[]).start()
+        patch(
+            "mcp_server._pr_work_lease_reviewer_block",
+            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
+        ).start()
 
     def tearDown(self):
         patch.stopall()
@@ -2401,7 +2505,8 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
     def test_merge_pr_with_closes_removes_label(self):
         import reviewer_pr_lease
 
-        lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
+        head_sha = FULL_HEAD_SHA
+        lease_patch = _install_owned_reviewer_lease(1, head_sha=head_sha)
         lease_patch.start()
         self.addCleanup(lease_patch.stop)
         self.addCleanup(reviewer_pr_lease.clear_session_lease)
@@ -2410,12 +2515,12 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
             if method == "GET" and "/user" in url:
                 return {"login": "merger"}
             if method == "GET" and url.endswith("/reviews"):
-                return [_formal_review("reviewer", "APPROVED", sha="sha123")]
+                return [_formal_review("reviewer", "APPROVED", sha=head_sha)]
             if method == "GET" and "pulls/1" in url and "/files" not in url:
                 return {
                     "user": {"login": "author"},
                     "state": "open",
-                    "head": {"sha": "sha123", "ref": "feat/my-branch"},
+                    "head": {"sha": head_sha, "ref": "feat/my-branch"},
                     "base": {"ref": "main"},
                     "mergeable": True,
                     "merged_commit_sha": "merge123",
@@ -2435,14 +2540,18 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
             return {}
         self.mock_api.side_effect = api_side_effect
 
-        res = gitea_merge_pr(pr_number=1, confirmation="MERGE PR 1", do="merge")
+        res = gitea_merge_pr(
+            pr_number=1, confirmation="MERGE PR 1", do="merge",
+            expected_head_sha=head_sha,
+        )
         self.assertTrue(res["performed"])
         self.assertEqual(res["cleanup_status"].get(123), "released")
 
     def test_merge_pr_with_branch_name_removes_label(self):
         import reviewer_pr_lease
 
-        lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
+        head_sha = FULL_HEAD_SHA
+        lease_patch = _install_owned_reviewer_lease(1, head_sha=head_sha)
         lease_patch.start()
         self.addCleanup(lease_patch.stop)
         self.addCleanup(reviewer_pr_lease.clear_session_lease)
@@ -2451,12 +2560,12 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
             if method == "GET" and "/user" in url:
                 return {"login": "merger"}
             if method == "GET" and url.endswith("/reviews"):
-                return [_formal_review("reviewer", "APPROVED", sha="sha123")]
+                return [_formal_review("reviewer", "APPROVED", sha=head_sha)]
             if method == "GET" and "pulls/1" in url and "/files" not in url:
                 return {
                     "user": {"login": "author"},
                     "state": "open",
-                    "head": {"sha": "sha123", "ref": "fix/issue-123-slug"},
+                    "head": {"sha": head_sha, "ref": "fix/issue-123-slug"},
                     "base": {"ref": "main"},
                     "mergeable": True,
                     "merged_commit_sha": "merge123",
@@ -2476,7 +2585,10 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
             return {}
         self.mock_api.side_effect = api_side_effect
 
-        res = gitea_merge_pr(pr_number=1, confirmation="MERGE PR 1", do="merge")
+        res = gitea_merge_pr(
+            pr_number=1, confirmation="MERGE PR 1", do="merge",
+            expected_head_sha=head_sha,
+        )
         self.assertTrue(res["performed"])
         self.assertEqual(res["cleanup_status"].get(123), "released")
 

tests/test_permission_reports.py

@@ -230,7 +230,9 @@ class TestEligibilityDenialReport(PermissionReportBase):
             return PR_PAYLOAD
         mock_api.side_effect = fake_api
         mcp_server.init_review_decision_lock("prgs", "review_pr")
-        mcp_server.gitea_mark_final_review_decision(42, "approve", remote="prgs")
+        from tests.test_mcp_server import _seed_ready_review_decision
+
+        _seed_ready_review_decision(42, "approve", remote="prgs")
         with patch.dict(os.environ, self._env("author-profile")):
             res = mcp_server.gitea_submit_pr_review(
                 pr_number=42, action="approve", body="lgtm", remote="prgs",
@@ -272,7 +274,9 @@ class TestReviewCommentPathUsesCanonicalOp(PermissionReportBase):
             return PR_PAYLOAD
         mock_api.side_effect = fake_api
         mcp_server.init_review_decision_lock("prgs", "review_pr")
-        mcp_server.gitea_mark_final_review_decision(42, "comment", remote="prgs")
+        from tests.test_mcp_server import _seed_ready_review_decision
+
+        _seed_ready_review_decision(42, "comment", remote="prgs")
         with patch.dict(os.environ, self._env("author-profile")):
             res = mcp_server.gitea_submit_pr_review(
                 pr_number=42, action="comment", body="finding", remote="prgs",

tests/test_pr_queue_inventory.py

@@ -128,18 +128,29 @@ class TestPRQueueInventory(unittest.TestCase):
             "forbidden_operations": [],
             "base_url": None,
         }
+
+        from mcp_server import init_review_decision_lock
+        from tests.test_mcp_server import (
+            FULL_HEAD_SHA,
+            _install_owned_reviewer_lease,
+            _seed_ready_review_decision,
+        )
+        import reviewer_pr_lease
+
+        head_sha = FULL_HEAD_SHA
         mock_fetch.return_value = _final_page_fetch([
-            {"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other_user"}}
+            {"number": 1, "title": "PR 1", "state": "open",
+             "head": {"ref": "branch1", "sha": head_sha},
+             "base": {"ref": "master"}, "mergeable": True,
+             "user": {"login": "other_user"}}
         ])
-        # mock_api: inventory whoami, eligibility whoami, eligibility PR,
-        # POST review (#244: state + visible-verdict GET reviews).
         mock_api.side_effect = [
             {"login": "reviewer1"},
             {"login": "reviewer1"},
             {
                 "user": {"login": "other_user"},
                 "state": "open",
-                "head": {"sha": "abc1"},
+                "head": {"sha": head_sha},
                 "mergeable": True,
             },
             {"id": 100, "state": "APPROVED"},
@@ -148,22 +159,19 @@ class TestPRQueueInventory(unittest.TestCase):
                     "id": 100,
                     "user": {"login": "reviewer1"},
                     "state": "APPROVED",
-                    "commit_id": "abc1",
+                    "commit_id": head_sha,
                     "submitted_at": "2026-07-06T10:00:00Z",
                     "dismissed": False,
                 }
             ],
         ]
 
-        from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
-        from tests.test_mcp_server import _install_owned_reviewer_lease
-        import reviewer_pr_lease
-
-        with patch("mcp_server._authenticated_username", return_value="reviewer1"):
+        with patch("mcp_server._authenticated_username", return_value="reviewer1"), \
+                patch("mcp_server._list_pr_lease_comments", return_value=[]):
             init_review_decision_lock("prgs", "review_pr")
-            gitea_mark_final_review_decision(1, "approve", remote="prgs")
+            _seed_ready_review_decision(1, "approve", sha=head_sha, remote="prgs")
             lease_patch = _install_owned_reviewer_lease(
-                1, head_sha="abc1", session_id="inventory-review-lease",
+                1, head_sha=head_sha, session_id="inventory-review-lease",
             )
             lease_patch.start()
             self.addCleanup(lease_patch.stop)

tests/test_pr_work_lease.py

@@ -0,0 +1,207 @@
+#!/usr/bin/env python3
+"""Regression tests for conflict-fix and reviewer PR work leases (#399)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from pr_work_lease import (  # noqa: E402
+    CONFLICT_FIX_LEASE_MARKER,
+    REVIEWER_LEASE_MARKER,
+    assess_conflict_fix_final_report,
+    assess_conflict_fix_push,
+    assess_head_sha_equality,
+    assess_reviewer_mutation_blocked,
+    assess_reviewer_stale_head_final_report,
+    format_conflict_fix_lease_body,
+    parse_conflict_fix_lease_comment,
+    parse_reviewer_lease_comment,
+)
+
+HEAD_A = "a" * 40
+HEAD_B = "b" * 40
+NOW = datetime(2026, 7, 7, 15, 0, tzinfo=timezone.utc)
+
+
+def _reviewer_lease_body(*, phase: str = "validating", expires_minutes: int = 60) -> str:
+    expires = (NOW + timedelta(minutes=expires_minutes)).isoformat().replace("+00:00", "Z")
+    return "\n".join([
+        REVIEWER_LEASE_MARKER,
+        "pr: #376",
+        "phase: " + phase,
+        f"candidate_head: {HEAD_A}",
+        f"expires_at: {expires}",
+        "profile: prgs-reviewer",
+    ])
+
+
+def _conflict_fix_body(*, phase: str = "claimed", worktree: str = "branches/fix-376") -> str:
+    expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
+    return "\n".join([
+        CONFLICT_FIX_LEASE_MARKER,
+        "pr: #376",
+        f"phase: {phase}",
+        f"worktree: {worktree}",
+        f"head_before: {HEAD_A}",
+        f"expires_at: {expires}",
+        "profile: prgs-author",
+    ])
+
+
+class TestLeaseParsing(unittest.TestCase):
+    def test_parse_reviewer_lease(self):
+        parsed = parse_reviewer_lease_comment(_reviewer_lease_body())
+        self.assertEqual(parsed["pr_number"], 376)
+        self.assertEqual(parsed["phase"], "validating")
+        self.assertEqual(parsed["candidate_head"], HEAD_A)
+
+    def test_parse_conflict_fix_lease(self):
+        parsed = parse_conflict_fix_lease_comment(_conflict_fix_body())
+        self.assertEqual(parsed["pr_number"], 376)
+        self.assertEqual(parsed["phase"], "claimed")
+
+
+class TestConflictFixPushGate(unittest.TestCase):
+    def test_blocks_push_during_active_reviewer_lease(self):
+        comments = [{"body": _reviewer_lease_body()}]
+        result = assess_conflict_fix_push(
+            pr_number=376,
+            comments=comments,
+            branch_head_before=HEAD_A,
+            branch_head_after=HEAD_B,
+            worktree_path="branches/fix-376",
+            push_cwd="/proj/branches/fix-376",
+            is_fast_forward=True,
+            now=NOW,
+        )
+        self.assertFalse(result["push_allowed"])
+        self.assertTrue(any("reviewer lease" in r for r in result["reasons"]))
+
+    def test_rejects_non_fast_forward(self):
+        result = assess_conflict_fix_push(
+            pr_number=376,
+            comments=[],
+            branch_head_before=HEAD_A,
+            branch_head_after=HEAD_B,
+            worktree_path="branches/fix-376",
+            push_cwd="/proj/branches/fix-376",
+            is_fast_forward=False,
+            now=NOW,
+        )
+        self.assertFalse(result["push_allowed"])
+        self.assertTrue(any("non-fast-forward" in r for r in result["reasons"]))
+
+    def test_wrong_cwd_push_attempt(self):
+        result = assess_conflict_fix_push(
+            pr_number=376,
+            comments=[],
+            branch_head_before=HEAD_A,
+            branch_head_after=HEAD_B,
+            worktree_path="branches/fix-376",
+            push_cwd="/proj/master",
+            is_fast_forward=True,
+            now=NOW,
+        )
+        self.assertFalse(result["push_allowed"])
+        self.assertTrue(any("cwd" in r.lower() for r in result["reasons"]))
+
+    def test_sibling_conflict_fix_collision(self):
+        comments = [{"body": _conflict_fix_body(phase="pushing", worktree="branches/other")}]
+        result = assess_conflict_fix_push(
+            pr_number=376,
+            comments=comments,
+            branch_head_before=HEAD_A,
+            branch_head_after=HEAD_B,
+            worktree_path="branches/fix-376",
+            push_cwd="/proj/branches/fix-376",
+            is_fast_forward=True,
+            now=NOW,
+        )
+        self.assertFalse(result["push_allowed"])
+        self.assertTrue(any("sibling conflict-fix" in r for r in result["reasons"]))
+
+
+class TestReviewerMutationGate(unittest.TestCase):
+    def test_blocks_review_during_conflict_fix(self):
+        comments = [{"body": _conflict_fix_body(phase="pushing")}]
+        result = assess_reviewer_mutation_blocked(
+            pr_number=376,
+            comments=comments,
+            reviewed_head_sha=HEAD_A,
+            live_head_sha=HEAD_A,
+            mutation="approve",
+            now=NOW,
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("conflict-fix lease" in r for r in result["reasons"]))
+
+    def test_stale_head_blocks_approval(self):
+        result = assess_reviewer_mutation_blocked(
+            pr_number=376,
+            comments=[],
+            reviewed_head_sha=HEAD_A,
+            live_head_sha=HEAD_B,
+            mutation="merge",
+            now=NOW,
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(result["head_check"]["head_changed"])
+
+    def test_head_equality_required_fields(self):
+        result = assess_head_sha_equality(HEAD_A, HEAD_B)
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["head_changed"])
+
+
+class TestFinalReportProof(unittest.TestCase):
+    def test_reviewer_stale_head_report_requires_fields(self):
+        result = assess_reviewer_stale_head_final_report("no head proof here")
+        self.assertFalse(result["proven"])
+
+    def test_reviewer_stale_head_report_passes(self):
+        report = "\n".join([
+            f"Reviewed head SHA: {HEAD_A}",
+            f"Final live head SHA before approval: {HEAD_A}",
+            f"Final live head SHA before merge: {HEAD_A}",
+            "Push occurred during validation: no",
+        ])
+        result = assess_reviewer_stale_head_final_report(report)
+        self.assertTrue(result["proven"])
+
+    def test_conflict_fix_report_requires_fields(self):
+        result = assess_conflict_fix_final_report("incomplete")
+        self.assertFalse(result["proven"])
+
+    def test_conflict_fix_report_passes(self):
+        report = "\n".join([
+            f"Branch head before push: {HEAD_A}",
+            f"Branch head after push: {HEAD_B}",
+            "Active reviewer lease status: none",
+            "Whether push was fast-forward: yes",
+            "Whether any reviewer was active: no",
+        ])
+        result = assess_conflict_fix_final_report(report)
+        self.assertTrue(result["proven"])
+
+
+class TestFormatLease(unittest.TestCase):
+    def test_format_conflict_fix_lease_includes_marker(self):
+        body = format_conflict_fix_lease_body(
+            pr_number=376,
+            branch="feat/x",
+            worktree="branches/fix-376",
+            profile="prgs-author",
+            head_before=HEAD_A,
+        )
+        self.assertIn(CONFLICT_FIX_LEASE_MARKER, body)
+        parsed = parse_conflict_fix_lease_comment(body)
+        self.assertEqual(parsed["pr_number"], 376)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_review_final_report_schema.py

@@ -19,7 +19,11 @@ def _minimal_review_report(**overrides):
         "- Issue/PR: #182 / PR #203",
         "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
         "- Files changed: review_proofs.py",
-        "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
+        "- Validation: pass: pytest tests/test_review_proofs.py -q in branches/review-203",
+        "- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Final live head SHA before approval: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Final live head SHA before merge: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Push occurred during validation: no",
         "- Mutations: review only",
         "- File edits by reviewer: none",
         "- Worktree/index mutations: none",
@@ -80,7 +84,7 @@ class TestReviewFinalReportSchema(unittest.TestCase):
 
     def test_reviewed_head_without_validation_blocks(self):
         report = _minimal_review_report().replace(
-            "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
+            "- Validation: pass: pytest tests/test_review_proofs.py -q in branches/review-203",
             "- Validation: not run",
         )
         report += "\n- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"

tests/test_terminal_review_hard_stop.py

@@ -12,6 +12,8 @@ from unittest.mock import patch
 
 import mcp_server
 
+HEAD_SHA = "a" * 40
+
 
 def _lock(mutations=None, correction=False):
     return {
@@ -127,10 +129,20 @@ def _feedback(blocking, stale=False, success=True):
         "success": success,
         "has_blocking_change_requests": blocking,
         "review_feedback_stale": stale,
-        "current_head_sha": "abc123",
+        "current_head_sha": HEAD_SHA,
     }
 
 
+def _mark(action, pr_number=6, **kwargs):
+    kwargs.setdefault("expected_head_sha", HEAD_SHA)
+    no_lease_block = {"block": False, "reasons": [], "mutation_allowed": True}
+    with patch("mcp_server._list_pr_lease_comments", return_value=[]), \
+            patch("mcp_server._pr_work_lease_reviewer_block", return_value=no_lease_block):
+        return mcp_server.gitea_mark_final_review_decision(
+            pr_number=pr_number, action=action, remote="prgs", **kwargs
+        )
+
+
 class TestDuplicateRequestChangesSuppression(unittest.TestCase):
     def tearDown(self):
         mcp_server._save_review_decision_lock(None)
@@ -139,8 +151,7 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
         _seed()
         with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                           return_value=_feedback(blocking=True, stale=False)):
-            result = mcp_server.gitea_mark_final_review_decision(
-                pr_number=6, action="request_changes", remote="prgs")
+            result = _mark("request_changes")
         self.assertFalse(result["marked_ready"])
         self.assertTrue(
             any("duplicate" in r for r in result["reasons"]),
@@ -150,16 +161,14 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
         _seed()
         with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                           return_value=_feedback(blocking=True, stale=True)):
-            result = mcp_server.gitea_mark_final_review_decision(
-                pr_number=6, action="request_changes", remote="prgs")
+            result = _mark("request_changes")
         self.assertTrue(result["marked_ready"], result.get("reasons"))
 
     def test_request_changes_allowed_when_no_blocker(self):
         _seed()
         with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                           return_value=_feedback(blocking=False)):
-            result = mcp_server.gitea_mark_final_review_decision(
-                pr_number=6, action="request_changes", remote="prgs")
+            result = _mark("request_changes")
         self.assertTrue(result["marked_ready"], result.get("reasons"))
 
     def test_request_changes_fails_closed_when_feedback_unavailable(self):
@@ -167,8 +176,7 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
         with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                           return_value=_feedback(blocking=False,
                                                  success=False)):
-            result = mcp_server.gitea_mark_final_review_decision(
-                pr_number=6, action="request_changes", remote="prgs")
+            result = _mark("request_changes")
         self.assertFalse(result["marked_ready"])
         self.assertTrue(
             any("could not verify" in r for r in result["reasons"]),
@@ -178,8 +186,7 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
         _seed()
         with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                           side_effect=AssertionError("must not be called")):
-            result = mcp_server.gitea_mark_final_review_decision(
-                pr_number=6, action="approve", remote="prgs")
+            result = _mark("approve")
         self.assertTrue(result["marked_ready"], result.get("reasons"))