fix: re-resolve conflicts for PR #516 after master advance

2026-07-08 22:33:21 -04:00
parent 09d8db5b9e 9a2e585a9e
commit d57ca94ec5
22 changed files with 2560 additions and 40 deletions
@@ -0,0 +1,52 @@
+# Controller Issue-Acceptance Gate
+
+A merged PR does not automatically prove an issue is fully satisfied. After
+merge, a controller must audit the linked issue against its acceptance criteria
+and post a durable handoff before the issue is treated as complete.
+
+## Workflow position
+
+1. Author implements the issue and opens a PR.
+2. Reviewer reviews the PR.
+3. Merger merges the approved PR.
+4. **Controller performs issue-acceptance audit.**
+5. Controller posts a `## Controller Issue Acceptance` comment with either:
+   - `STATE: accepted` and checked criteria, or
+   - a rejection path (`more-work-required`, `needs-tests`, `needs-docs`, etc.)
+     with `MISSING_WORK` and a paste-ready `NEXT_PROMPT`.
+
+Gitea may auto-close an issue via `Closes #N` in the PR body. That closure is
+merge mechanics only. Controller acceptance is still required before any final
+report or queue controller treats the issue as complete.
+
+## Template
+
+Use `issue_acceptance_gate.render_controller_acceptance_template()` or the
+copy in
+[`skills/llm-project-workflow/templates/controller-issue-acceptance.md`](../skills/llm-project-workflow/templates/controller-issue-acceptance.md).
+
+## Final-report rules
+
+Final reports must not claim `issue complete` solely because a PR merged.
+Either:
+
+- include a valid `## Controller Issue Acceptance` block with
+  `STATE: accepted`, or
+- explicitly state `controller acceptance pending` and identify the controller
+  as the next actor.
+
+`final_report_validator` enforces this through
+`issue_acceptance_gate.validate_final_report_issue_acceptance()`.
+
+## Role boundaries
+
+- Authors must not mark their own issues accepted.
+- Reviewers must not mark issue acceptance unless acting under controller
+  capability.
+- Mergers merge PRs; they do not substitute for controller acceptance.
+
+## Related
+
+- #495 — canonical next-action comment templates
+- #496 — fail-closed canonical comment validation before posting
+- #303 — controller handoff schema for reconciliation workflows
@@ -929,6 +929,7 @@ When posting a Canonical Thread Handoff after a binding blocker:

 ## Related documents

+- [`issue-acceptance-gate.md`](issue-acceptance-gate.md) — controller issue-acceptance audit after PR merge (#500).
 - [`../skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md) — portable cross-project LLM workflow skill.
 - [`gitea-execution-profiles.md`](gitea-execution-profiles.md) — the profile model.
 - [`gitea-dual-namespace-deployment.md`](gitea-dual-namespace-deployment.md) — static author/reviewer namespace deployment (#139 decision).
@@ -46,7 +46,8 @@ Optional environment variables:
 | `/runtime` | Stub — MCP runtime health (#430) |
 | `/audit` | Stub — report audit paste (#431) |
 | `/worktrees` | Stub — hygiene dashboard (#432) |
-| `/leases` | Stub — lease visibility (#433) |
+| `/leases` | Lease and collision visibility (#433) |
+| `/api/leases` | JSON lease/collision export |

 All routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
 `read-only-mvp`.
@@ -82,8 +83,16 @@ credentials. The UI surfaces pagination proof (returned count, pages fetched,
 If credentials are missing or the fetch fails, the page shows an explicit error
 instead of an empty queue (fail closed).

+## Lease visibility (#433)
+
+`/leases` surfaces read-only lease and collision state: local issue lock file,
+in-progress claim inventory (#268), reviewer PR lease comments when present
+(`<!-- mcp-review-lease:v1 -->`, #407), duplicate open PRs per issue (#400),
+and duplicate local branches per issue. Links to collision-history backend
+issues (#267, #268, #400, #407) are included. No lease acquire/release from UI.
+
 ## Tests

 ```bash
-pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py -q
+pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_lease_visibility.py -q
 ```
@@ -12,6 +12,7 @@ import re
 from typing import Any, Callable

 import branch_cleanup_guard
+import issue_acceptance_gate
 import issue_lock_provenance
 from mcp_native_cleanup_proof import assess_mcp_native_cleanup_proof
 from post_merge_cleanup_proof import assess_post_merge_cleanup_proof
@@ -961,6 +962,69 @@ def _rule_reconcile_linked_issue_live(
    return []


+_PR_CLOSE_NEGATIVE = frozenset({"", "none", "n/a", "na", "0", "no", "not closed", "not performed"})
+_ANCESTOR_AFFIRMATIVE_RE = re.compile(
+    r"ancestor|passed|true|verified|confirmed", re.IGNORECASE
+)
+
+
+def _reconciler_pr_close_performed(fields: dict[str, str], lock: dict) -> bool:
+    """True when the report/session indicates a reconciler PR close happened."""
+    if lock.get("pr_closed") is True:
+        return True
+    value = (fields.get("prs closed", "") or "").strip().lower()
+    if value in _PR_CLOSE_NEGATIVE:
+        return False
+    # A closed PR is reported by number (e.g. "#99") or an affirmative result.
+    return bool(re.search(r"#\s*\d+|\b(?:closed|success|done)\b", value))
+
+
+def _rule_reconcile_close_proof(
+    report_text: str,
+    *,
+    reconciler_close_lock: dict | None = None,
+) -> list[dict[str, str]]:
+    """#306: a reconciler PR close must carry exact proof fields.
+
+    Read-only/comment-only reconciliations are untouched. Once a PR close is
+    reported (via the ``PRs closed`` field or a session close lock), the
+    handoff must prove the close capability, ancestor landing, PR close
+    result, and the linked-issue result — narrative alone fails closed.
+    """
+    fields = _handoff_fields(report_text)
+    lock = reconciler_close_lock or {}
+    if not _reconciler_pr_close_performed(fields, lock):
+        return []
+
+    missing: list[str] = []
+    capabilities = fields.get("capabilities proven", "")
+    if "gitea.pr.close" not in capabilities.lower():
+        missing.append("close capability proof (gitea.pr.close)")
+    ancestor = fields.get("ancestor proof", "")
+    if not _ANCESTOR_AFFIRMATIVE_RE.search(ancestor):
+        missing.append("ancestor proof")
+    prs_closed = (fields.get("prs closed", "") or "").strip().lower()
+    if prs_closed in _PR_CLOSE_NEGATIVE and lock.get("pr_closed") is True:
+        missing.append("PR close result")
+    linked = fields.get("linked issue live status", "") or fields.get("issues closed", "")
+    if not linked.strip():
+        missing.append("linked issue result")
+
+    if not missing:
+        return []
+    return [
+        validator_finding(
+            "reconcile.close_proof_fields",
+            "block",
+            "Reconciler close proof",
+            "reconciler PR close reported without required proof field(s): "
+            + ", ".join(missing),
+            "include close capability proof, ancestor proof, PR close result, "
+            "and linked issue result in the handoff",
+        )
+    ]
+
+
 def _rule_reconcile_pagination_proof(report_text: str) -> list[dict[str, str]]:
    text = report_text or ""
    if not _INVENTORY_COMPLETE_RE.search(text):
@@ -1047,6 +1111,22 @@ def _rule_shared_manual_lock_pr_override(report_text: str) -> list[dict[str, str
    )


+def _rule_shared_issue_acceptance_gate(report_text: str) -> list[dict[str, str]]:
+    result = issue_acceptance_gate.validate_final_report_issue_acceptance(report_text)
+    if not result.get("applicable") or result.get("valid"):
+        return []
+    return _findings_from_reasons(
+        "shared.issue_acceptance_gate",
+        result.get("reasons") or [],
+        field="Controller acceptance",
+        severity="block",
+        safe_next_action=(
+            "add Controller Issue Acceptance proof or state that controller "
+            "acceptance is pending; do not claim issue complete from PR merge alone"
+        ),
+    )
+
+
 def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, str]]:
    result = issue_lock_provenance.assess_author_reviewer_same_run_report(report_text)
    if result.get("proven"):
@@ -1249,6 +1329,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
        _rule_reconcile_stale_author_fields,
        _rule_reconcile_eligible_reviewed,
        _rule_reconcile_linked_issue_live,
+        _rule_reconcile_close_proof,
        _rule_reconcile_pagination_proof,
        _rule_reviewer_git_fetch_readonly,
        _rule_reviewer_legacy_workspace_mutations,
@@ -1265,6 +1346,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
+        _rule_shared_issue_acceptance_gate,
        _rule_reviewer_vague_mutations_none,
        _rule_conflict_fix_push_proof,
        _rule_worktree_cleanup_audit_proof,
@@ -1347,6 +1429,7 @@ def assess_final_report_validator(
    issue_filing_lock: dict | None = None,
    session_pr_opened: bool = False,
    validation_session: dict | None = None,
+    reconciler_close_lock: dict | None = None,
 ) -> dict[str, Any]:
    """Validate final-report text against task-specific proof rules (#327).

@@ -1403,6 +1486,7 @@ def assess_final_report_validator(
        "local_edits": local_edits,
        "session_pr_opened": session_pr_opened,
        "validation_session": validation_session,
+        "reconciler_close_lock": reconciler_close_lock,
    }

    for rule in _RULES_BY_TASK.get(normalized_kind, ()):
@@ -696,6 +696,39 @@ def _verify_role_mutation_workspace(
    task: str | None = None,
 ) -> str:
    """Bind reviewer/merger mutations to the active namespace workspace (#510)."""
+    # Check running runtimes to prevent stale mutations
+    try:
+        if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
+            config = gitea_config.load_config()
+            required_permission = task_capability_map.required_permission(task) if task else None
+            matching_profiles = []
+            if required_permission and config and "profiles" in config:
+                for p_name, p_data in config["profiles"].items():
+                    p_allowed = p_data.get("allowed_operations") or []
+                    p_forbidden = p_data.get("forbidden_operations") or []
+                    p_allowed_n = []
+                    for op in p_allowed:
+                        try:
+                            p_allowed_n.append(gitea_config.normalize_operation(op))
+                        except Exception:
+                            pass
+                    p_forbidden_n = []
+                    for op in p_forbidden:
+                        try:
+                            p_forbidden_n.append(gitea_config.normalize_operation(op))
+                        except Exception:
+                            pass
+                    ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n)
+                    if ok:
+                        matching_profiles.append(p_name)
+            runtime_reasons = _check_mcp_runtimes_diagnostics(task or "unknown", matching_profiles)
+            if runtime_reasons:
+                raise RuntimeError("; ".join(runtime_reasons))
+    except Exception as exc:
+        if "stale-runtime:" in str(exc):
+            raise RuntimeError(str(exc))
+        pass
+
    role = _effective_workspace_role()
    git_state = issue_lock_worktree.read_worktree_git_state(
        _resolve_preflight_workspace_path(worktree_path)
@@ -771,6 +804,7 @@ import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
 import review_workflow_boundary  # noqa: E402
 import review_workflow_load  # noqa: E402
+import mcp_session_state  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
 import issue_lock_provenance  # noqa: E402
@@ -2448,37 +2482,110 @@ _REVIEW_ACTIONS = {

 _TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})

-# In-process only (#211): never persist to /tmp — host-global files are
-# spoofable by any local process and go stale across sessions.
+# Durable across MCP daemon process pools (#559), but never under /tmp (#211):
+# host-global temp files are spoofable. Persistence uses the user-private
+# cache directory from mcp_session_state (mode 0o700 / files 0o600), keyed by
+# remote + profile identity with TTL.
 _REVIEW_DECISION_LOCK: dict | None = None


+def _decision_lock_binding(lock: dict | None = None) -> dict:
+    """Resolve key fields for durable decision-lock storage."""
+    profile = get_profile()
+    profile_name = (profile.get("profile_name") or "").strip()
+    env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
+    stored_lock = ((lock or {}).get("session_profile_lock") or "").strip()
+    session_lock = env_lock or stored_lock or profile_name
+    remote = ((lock or {}).get("remote") or "").strip() or None
+    org = ((lock or {}).get("org") or (lock or {}).get("ready_org") or "").strip() or None
+    repo = ((lock or {}).get("repo") or (lock or {}).get("ready_repo") or "").strip() or None
+    return {
+        "session_profile": profile_name,
+        "session_profile_lock": session_lock,
+        "profile_identity": mcp_session_state.current_profile_identity(
+            profile_name=profile_name,
+            session_profile_lock=session_lock,
+        ),
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+    }
+
+
 def _load_review_decision_lock():
+    """Load decision lock from memory, falling back to durable session state."""
    global _REVIEW_DECISION_LOCK
+    if _REVIEW_DECISION_LOCK is not None:
+        return _REVIEW_DECISION_LOCK
+    binding = _decision_lock_binding()
+    # Profile-keyed durable file; remote/org/repo validated from payload (#559).
+    durable = mcp_session_state.load_state(
+        kind=mcp_session_state.KIND_DECISION_LOCK,
+        profile_identity=binding.get("profile_identity"),
+    )
+    if durable is not None:
+        _REVIEW_DECISION_LOCK = dict(durable)
    return _REVIEW_DECISION_LOCK


 def _save_review_decision_lock(data):
+    """Persist decision lock to memory + durable shared state (#559)."""
    global _REVIEW_DECISION_LOCK
-    _REVIEW_DECISION_LOCK = data
+    if data is None:
+        binding = _decision_lock_binding(_REVIEW_DECISION_LOCK)
+        mcp_session_state.clear_state(
+            kind=mcp_session_state.KIND_DECISION_LOCK,
+            profile_identity=binding.get("profile_identity"),
+        )
+        _REVIEW_DECISION_LOCK = None
+        return
+    payload = dict(data)
+    binding = _decision_lock_binding(payload)
+    payload.setdefault("session_pid", os.getpid())
+    payload["writer_pid"] = os.getpid()
+    payload["session_profile"] = binding["session_profile"] or payload.get(
+        "session_profile"
+    )
+    payload["session_profile_lock"] = binding["session_profile_lock"]
+    payload["profile_identity"] = binding["profile_identity"]
+    if binding.get("remote") and not payload.get("remote"):
+        payload["remote"] = binding["remote"]
+    persisted = mcp_session_state.save_state(
+        kind=mcp_session_state.KIND_DECISION_LOCK,
+        payload=payload,
+        remote=payload.get("remote"),
+        org=payload.get("org") or payload.get("ready_org"),
+        repo=payload.get("repo") or payload.get("ready_repo"),
+        profile_identity=payload.get("profile_identity"),
+    )
+    _REVIEW_DECISION_LOCK = dict(persisted or payload)


 def _review_decision_session_reasons(lock: dict | None) -> list[str]:
-    """Reject locks that do not belong to this MCP process/session."""
+    """Reject locks that do not belong to this MCP session identity (#559).
+
+    Different daemon PIDs in the same IDE session pool are allowed when the
+    profile identity and remote match. Spoofed/stale locks still fail closed.
+    """
    if lock is None:
        return []
    reasons = []
-    if lock.get("session_pid") != os.getpid():
-        reasons.append(
-            "review decision lock was created in a different process "
-            "(fail closed)"
-        )
    env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
-    stored_lock = (lock.get("session_profile_lock") or "").strip()
+    stored_lock = (lock.get("session_profile_lock") or lock.get("profile_identity") or "").strip()
    if env_lock and stored_lock and env_lock != stored_lock:
        reasons.append(
            "review decision lock session profile lock mismatch (fail closed)"
        )
+    # TTL / identity checks from durable envelope fields.
+    for reason in mcp_session_state.identity_match_reasons(
+        lock,
+        remote=lock.get("remote"),
+        org=lock.get("org") or lock.get("ready_org"),
+        repo=lock.get("repo") or lock.get("ready_repo"),
+        profile_identity=env_lock or stored_lock,
+    ):
+        if "profile identity mismatch" in reason or "expired" in reason or "future" in reason or "missing recorded_at" in reason:
+            reasons.append(reason)
    return reasons


@@ -2489,11 +2596,12 @@ def init_review_decision_lock(remote: str | None, task: str | None, force: bool
    if not force:
        lock = _load_review_decision_lock()
        if lock is not None:
-            if lock.get("remote") == remote and lock.get("session_pid") == os.getpid():
-                env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
-                stored_lock = (lock.get("session_profile_lock") or "").strip()
-                if not env_lock or not stored_lock or env_lock == stored_lock:
-                    return
+            env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
+            stored_lock = (lock.get("session_profile_lock") or "").strip()
+            same_remote = lock.get("remote") == remote
+            same_profile = (not env_lock or not stored_lock or env_lock == stored_lock)
+            if same_remote and same_profile and not _review_decision_session_reasons(lock):
+                return
    review_workflow_load.clear_review_workflow_load()
    profile = get_profile()
    profile_name = (profile.get("profile_name") or "").strip()
@@ -2508,6 +2616,10 @@ def init_review_decision_lock(remote: str | None, task: str | None, force: bool
        "session_pid": os.getpid(),
        "session_profile": profile_name,
        "session_profile_lock": session_lock,
+        "profile_identity": mcp_session_state.current_profile_identity(
+            profile_name=profile_name,
+            session_profile_lock=session_lock,
+        ),
        "final_review_decision_ready": False,
        "ready_pr_number": None,
        "ready_action": None,
@@ -8745,6 +8857,102 @@ def gitea_route_task_session(
    )


+def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> list[str]:
+    """Check running runtimes and return errors if they are missing or stale."""
+    import subprocess
+    import re
+    from datetime import datetime
+
+    reasons = []
+    code_path = os.path.join(PROJECT_ROOT, "gitea_mcp_server.py")
+    if not os.path.exists(code_path):
+        return reasons
+
+    code_mtime = datetime.fromtimestamp(os.path.getmtime(code_path))
+
+    try:
+        proc = subprocess.run(
+            ["ps", "-o", "pid,lstart,command", "-ax"],
+            capture_output=True, text=True, check=True
+        )
+    except Exception as exc:
+        return [f"stale-runtime: failed to list running processes: {exc}"]
+
+    self_pid = os.getpid()
+    self_stale = False
+
+    running_profiles = {}
+    for line in proc.stdout.splitlines()[1:]:
+        line = line.strip()
+        if not line or "mcp_server.py" not in line:
+            continue
+
+        parts = line.split(None, 6)
+        if len(parts) < 7:
+            continue
+        pid_str = parts[0]
+        lstart_str = " ".join(parts[1:6])
+
+        try:
+            pid = int(pid_str)
+            start_time = datetime.strptime(lstart_str, "%a %b %d %H:%M:%S %Y")
+        except Exception:
+            continue
+
+        try:
+            env_proc = subprocess.run(
+                ["ps", "eww", str(pid)],
+                capture_output=True, text=True, check=True
+            )
+            env_out = env_proc.stdout
+        except Exception:
+            continue
+
+        profile = "gitea-default"
+        match = re.search(r'\bGITEA_MCP_PROFILE=([^\s]+)', env_out)
+        if match:
+            profile = match.group(1)
+
+        is_stale = start_time < code_mtime
+        if pid == self_pid and is_stale:
+            self_stale = True
+
+        if profile not in running_profiles or start_time > running_profiles[profile]["start_time"]:
+            running_profiles[profile] = {
+                "pid": pid,
+                "start_time": start_time,
+                "is_stale": is_stale
+            }
+
+    if self_stale:
+        reasons.append(
+            "stale-runtime: The active Gitea MCP server process is stale (running code from before changes were merged). "
+            "Please fully restart the Gitea MCP server (e.g. touch /Users/jasonwalker/.gemini/config/mcp_config.json) and retry."
+        )
+
+    if matching_profiles:
+        any_running = False
+        any_fresh = False
+        for mp in matching_profiles:
+            if mp in running_profiles:
+                any_running = True
+                if not running_profiles[mp]["is_stale"]:
+                    any_fresh = True
+                    break
+        if not any_running:
+            reasons.append(
+                f"stale-runtime: None of the matching profiles for task '{task}' ({matching_profiles}) are running in the OS. "
+                "Please restart the MCP server to ensure they are spawned."
+            )
+        elif not any_fresh:
+            reasons.append(
+                f"stale-runtime: All matching profiles for task '{task}' ({matching_profiles}) are running but stale. "
+                "Please fully restart the Gitea MCP server and retry."
+            )
+
+    return reasons
+
+
@mcp.tool()
 def gitea_resolve_task_capability(
    task: str,
@@ -8855,6 +9063,16 @@ def gitea_resolve_task_capability(
    configured = len(matching_profiles) > 0
    available_in_session = allowed_in_current_session

+    if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
+        runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
+        if runtime_reasons:
+            restart_required = True
+            reason_msg = "; ".join(runtime_reasons)
+            next_safe_action = (
+                "stale-runtime: Gitea MCP runtime conflict or missing process detected. "
+                "Please fully restart the Gitea MCP server and retry."
+            )
+
    if not allowed_in_current_session:
        if configured and switching:
            restart_required = True
@@ -0,0 +1,300 @@
+"""Controller issue-acceptance gate helpers (#500).
+
+Pure validation for controller acceptance comments and final-report claims
+that an issue is complete. Does not post comments or close issues.
+"""
+
+from __future__ import annotations
+
+import re
+
+ACCEPTANCE_HEADING = "controller issue acceptance"
+
+REQUIRED_FIELDS = (
+    "STATE",
+    "WHO_IS_NEXT",
+    "NEXT_ACTION",
+    "NEXT_PROMPT",
+    "ISSUE",
+    "MERGED_PR",
+    "MERGE_COMMIT",
+    "ACCEPTANCE_CRITERIA_CHECKED",
+    "VALIDATION_REVIEWED",
+    "CONTROLLER_DECISION",
+    "WHY",
+)
+
+ACCEPTED_STATES = frozenset({"accepted"})
+REJECTION_STATES = frozenset({
+    "more-work-required",
+    "more_work_required",
+    "needs-tests",
+    "needs_tests",
+    "needs-docs",
+    "needs_docs",
+    "needs-feature-enhancement",
+    "needs_feature_enhancement",
+    "needs-follow-up-issue",
+    "needs_follow_up_issue",
+    "blocked",
+})
+
+ALLOWED_NEXT_ACTORS = frozenset({
+    "controller",
+    "author",
+    "reviewer",
+    "merger",
+    "reconciler",
+    "user",
+})
+
+_FIELD_RE = re.compile(r"^\s*(?:[-*]\s*)?([A-Z][A-Z0-9_ ]+)\s*:\s*(.*)$")
+_FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
+_ISSUE_REF_RE = re.compile(r"#\d+")
+_PR_REF_RE = re.compile(r"#\d+")
+_CHECKED_ITEM_RE = re.compile(r"\[[xX]\]")
+_UNCHECKED_ITEM_RE = re.compile(r"\[[\s]\]")
+
+_CLAIMS_ISSUE_COMPLETE_RE = re.compile(
+    r"\bissue\s+(?:is\s+)?(?:complete|completed|accepted|closed\s+as\s+complete|fully\s+satisfied)\b|"
+    r"\bissue\s+acceptance\s*:\s*accepted\b|"
+    r"\bcontroller\s+acceptance\s*:\s*(?:accepted|complete)\b",
+    re.IGNORECASE,
+)
+_MERGE_ONLY_COMPLETE_RE = re.compile(
+    r"(?:pr\s+merged|merged\s+pr|merge\s+result\s*:\s*merged).{0,120}"
+    r"(?:issue\s+(?:is\s+)?(?:complete|closed|accepted)|issue\s+complete)",
+    re.IGNORECASE | re.DOTALL,
+)
+_CLAIMS_CONTROLLER_ACCEPTANCE_RE = re.compile(
+    r"controller\s+issue\s+acceptance|controller\s+acceptance\s+(?:posted|complete|pending)",
+    re.IGNORECASE,
+)
+_PENDING_ACCEPTANCE_RE = re.compile(
+    r"controller\s+acceptance\s+(?:pending|required|not\s+(?:yet\s+)?(?:performed|complete))",
+    re.IGNORECASE,
+)
+
+
+def render_controller_acceptance_template() -> str:
+    """Return the canonical controller issue-acceptance comment template."""
+    return """## Controller Issue Acceptance
+
+STATE:
+<accepted | more-work-required | needs-tests | needs-docs | needs-feature-enhancement | needs-follow-up-issue | blocked>
+
+WHO_IS_NEXT:
+<author | reviewer | merger | reconciler | controller | user>
+
+NEXT_ACTION:
+<one sentence>
+
+NEXT_PROMPT:
+<paste-ready prompt for the next LLM>
+
+ISSUE:
+#...
+
+MERGED_PR:
+#...
+
+MERGE_COMMIT:
+<40-character SHA>
+
+ACCEPTANCE_CRITERIA_CHECKED:
+- [x] ...
+- [ ] ...
+
+VALIDATION_REVIEWED:
+<tests/proofs reviewed>
+
+CONTROLLER_DECISION:
+<accepted or rejected>
+
+WHY:
+<reasoning>
+
+MISSING_WORK:
+<none, or exact missing work>
+
+FOLLOW_UP_ISSUES:
+<none, or issue list to create>
+
+BLOCKERS:
+<none, or exact blockers>
+
+LAST_UPDATED_BY:
+<identity/profile/date>
+"""
+
+
+def extract_acceptance_fields(text: str | None) -> dict[str, str]:
+    """Return upper-case labeled fields from a controller acceptance block."""
+    fields: dict[str, str] = {}
+    current_key: str | None = None
+    for line in (text or "").splitlines():
+        match = _FIELD_RE.match(line)
+        if match:
+            current_key = match.group(1).strip().upper().replace(" ", "_")
+            fields[current_key] = match.group(2).strip()
+            continue
+        stripped = line.strip()
+        if current_key and stripped:
+            existing = fields.get(current_key, "")
+            fields[current_key] = (
+                f"{existing}\n{stripped}" if existing else stripped
+            )
+    return fields
+
+
+def contains_acceptance_block(text: str | None) -> bool:
+    return ACCEPTANCE_HEADING in (text or "").lower()
+
+
+def _empty_or_placeholder(value: str | None) -> bool:
+    value = (value or "").strip().lower()
+    return not value or value in {"none", "n/a", "unknown", "tbd", "<...>", "..."}
+
+
+def _normalize_state(value: str | None) -> str:
+    return (value or "").strip().lower().replace(" ", "_").replace("-", "_")
+
+
+def validate_controller_acceptance_comment(text: str | None) -> dict:
+    """Validate a controller issue-acceptance comment."""
+    body = text or ""
+    if not contains_acceptance_block(body):
+        return {
+            "valid": False,
+            "fields": {},
+            "reasons": ["missing Controller Issue Acceptance heading"],
+        }
+
+    fields = extract_acceptance_fields(body)
+    reasons: list[str] = []
+
+    for field in REQUIRED_FIELDS:
+        if _empty_or_placeholder(fields.get(field)):
+            reasons.append(f"missing required controller acceptance field: {field}")
+
+    state = _normalize_state(fields.get("STATE"))
+    if state and state not in ACCEPTED_STATES and state not in REJECTION_STATES:
+        reasons.append(
+            "STATE must be accepted or a rejection path "
+            "(more-work-required, needs-tests, needs-docs, "
+            "needs-feature-enhancement, needs-follow-up-issue, blocked)"
+        )
+
+    actor = (fields.get("WHO_IS_NEXT") or "").strip().lower()
+    if actor and actor not in ALLOWED_NEXT_ACTORS:
+        reasons.append(
+            "WHO_IS_NEXT must be one of: "
+            + ", ".join(sorted(ALLOWED_NEXT_ACTORS))
+        )
+
+    if not _ISSUE_REF_RE.search(fields.get("ISSUE") or ""):
+        reasons.append("ISSUE must cite an issue number (#N)")
+    if not _PR_REF_RE.search(fields.get("MERGED_PR") or ""):
+        reasons.append("MERGED_PR must cite a merged PR number (#N)")
+    if not _FULL_SHA_RE.search(fields.get("MERGE_COMMIT") or ""):
+        reasons.append("MERGE_COMMIT must include a full 40-character SHA")
+
+    criteria = fields.get("ACCEPTANCE_CRITERIA_CHECKED") or ""
+    if not _CHECKED_ITEM_RE.search(criteria) and not _UNCHECKED_ITEM_RE.search(criteria):
+        reasons.append(
+            "ACCEPTANCE_CRITERIA_CHECKED must list checked/unchecked criteria items"
+        )
+
+    decision = (fields.get("CONTROLLER_DECISION") or "").strip().lower()
+    if state in ACCEPTED_STATES:
+        if decision not in {"accepted", "accept"}:
+            reasons.append("accepted STATE requires CONTROLLER_DECISION: accepted")
+        if not _CHECKED_ITEM_RE.search(criteria):
+            reasons.append(
+                "accepted STATE requires at least one checked acceptance criterion"
+            )
+        if _empty_or_placeholder(fields.get("WHY")):
+            reasons.append("accepted STATE requires WHY with acceptance rationale")
+    elif state in REJECTION_STATES:
+        if decision not in {"rejected", "reject", "more_work_required"}:
+            reasons.append(
+                "rejection STATE requires CONTROLLER_DECISION: rejected"
+            )
+        if _empty_or_placeholder(fields.get("NEXT_PROMPT")):
+            reasons.append(
+                "rejection STATE requires a paste-ready NEXT_PROMPT for the next actor"
+            )
+        missing = fields.get("MISSING_WORK") or ""
+        if _empty_or_placeholder(missing):
+            reasons.append(
+                "rejection STATE requires MISSING_WORK describing what is still needed"
+            )
+
+    return {
+        "valid": not reasons,
+        "fields": fields,
+        "reasons": reasons,
+    }
+
+
+def claims_issue_complete(text: str | None) -> bool:
+    """Return True when text claims an issue is complete/accepted."""
+    return bool(_CLAIMS_ISSUE_COMPLETE_RE.search(text or ""))
+
+
+def claims_merge_only_issue_complete(text: str | None) -> bool:
+    """Return True when text treats PR merge as issue completion."""
+    return bool(_MERGE_ONLY_COMPLETE_RE.search(text or ""))
+
+
+def claims_controller_acceptance_update(text: str | None) -> bool:
+    return bool(_CLAIMS_CONTROLLER_ACCEPTANCE_RE.search(text or ""))
+
+
+def notes_controller_acceptance_pending(text: str | None) -> bool:
+    return bool(_PENDING_ACCEPTANCE_RE.search(text or ""))
+
+
+def validate_final_report_issue_acceptance(report_text: str | None) -> dict:
+    """Validate issue-completion and controller-acceptance claims in final reports."""
+    text = report_text or ""
+    reasons: list[str] = []
+
+    complete_claim = claims_issue_complete(text)
+    merge_only = claims_merge_only_issue_complete(text)
+    acceptance_claim = claims_controller_acceptance_update(text)
+    pending_noted = notes_controller_acceptance_pending(text)
+    has_block = contains_acceptance_block(text)
+
+    if merge_only and not (has_block and validate_controller_acceptance_comment(text)["valid"]):
+        reasons.append(
+            "final report treats PR merge as issue completion without controller acceptance proof"
+        )
+
+    if complete_claim and not pending_noted:
+        if not has_block:
+            reasons.append(
+                "final report claims issue complete but includes no Controller Issue Acceptance block"
+            )
+        else:
+            result = validate_controller_acceptance_comment(text)
+            if not result["valid"]:
+                reasons.extend(result["reasons"])
+            else:
+                state = _normalize_state(result["fields"].get("STATE"))
+                if state not in ACCEPTED_STATES:
+                    reasons.append(
+                        "final report claims issue complete but controller STATE is not accepted"
+                    )
+
+    if acceptance_claim and has_block:
+        result = validate_controller_acceptance_comment(text)
+        if not result["valid"]:
+            reasons.extend(result["reasons"])
+
+    applicable = complete_claim or merge_only or acceptance_claim or pending_noted
+    return {
+        "applicable": applicable,
+        "valid": not reasons,
+        "reasons": reasons,
+    }
@@ -0,0 +1,391 @@
+"""Durable MCP session validation state shared across daemon process pools (#559).
+
+The IDE often routes sequential MCP tool calls to different daemon processes.
+Session-scoped proofs (workflow load, review decision lock) must therefore
+survive process boundaries while remaining fail-closed against spoofing.
+
+Security notes (extends #211):
+- Never store under host-global ``/tmp`` (world-writable, spoofable).
+- Default root is ``~/.cache/gitea-tools/session-state`` (mode ``0o700``).
+- Files are written atomically with mode ``0o600``.
+- Records are keyed by remote + org + repo + profile identity, not by PID.
+- TTL prevents indefinitely stale reuse across unrelated sessions.
+"""
+
+from __future__ import annotations
+
+import fcntl
+import json
+import os
+import re
+import tempfile
+from contextlib import contextmanager
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+STATE_DIR_ENV = "GITEA_MCP_SESSION_STATE_DIR"
+DEFAULT_STATE_DIR = os.path.expanduser("~/.cache/gitea-tools/session-state")
+TTL_HOURS_ENV = "GITEA_MCP_SESSION_STATE_TTL_HOURS"
+DEFAULT_TTL_HOURS = 4.0
+
+KIND_WORKFLOW_LOAD = "review_workflow_load"
+KIND_DECISION_LOCK = "review_decision_lock"
+
+_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
+SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"
+
+
+def default_state_dir() -> str:
+    raw = (os.environ.get(STATE_DIR_ENV) or DEFAULT_STATE_DIR).strip()
+    return raw or DEFAULT_STATE_DIR
+
+
+def ttl_hours() -> float:
+    raw = (os.environ.get(TTL_HOURS_ENV) or "").strip()
+    if not raw:
+        return DEFAULT_TTL_HOURS
+    try:
+        value = float(raw)
+    except ValueError:
+        return DEFAULT_TTL_HOURS
+    return value if value > 0 else DEFAULT_TTL_HOURS
+
+
+def _sanitize_segment(value: str) -> str:
+    text = (value or "").strip()
+    if not text:
+        return "_"
+    return _SAFE_SEGMENT_RE.sub("_", text)
+
+
+def current_profile_identity(
+    profile_name: str | None = None,
+    session_profile_lock: str | None = None,
+    profile_identity: str | None = None,
+) -> str:
+    """Resolve the session profile identity used as the durable key."""
+    env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
+    explicit = (profile_identity or session_profile_lock or "").strip()
+    lock = (explicit or env_lock or "").strip()
+    name = (profile_name or "").strip()
+    return lock or name or "unknown-profile"
+
+
+def state_key(
+    *,
+    kind: str,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    profile_identity: str | None = None,
+) -> str:
+    """Build durable filename key.
+
+    Session proofs are one-active-per-profile (workflow load / decision lock),
+    so the primary key is kind + profile identity. Remote/org/repo are stored
+    inside the payload and validated on load (#559), which lets a later daemon
+    process recover state without already knowing the remote argument.
+    """
+    # Keep remote/org/repo parameters for API stability / future kinds; they are
+    # intentionally not part of the filename for session-scoped proofs.
+    _ = (remote, org, repo)
+    return "-".join(
+        _sanitize_segment(part)
+        for part in (
+            kind,
+            profile_identity or "unknown-profile",
+        )
+    )
+
+
+def state_file_path(
+    *,
+    kind: str,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    profile_identity: str | None = None,
+    state_dir: str | None = None,
+) -> str:
+    root = (state_dir or default_state_dir()).strip()
+    name = state_key(
+        kind=kind,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile_identity,
+    )
+    return os.path.join(root, f"{name}.json")
+
+
+def _ensure_state_dir(state_dir: str | None = None) -> str:
+    root = (state_dir or default_state_dir()).strip()
+    os.makedirs(root, mode=0o700, exist_ok=True)
+    return root
+
+
+def _now_utc() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _parse_iso(value: str | None) -> datetime | None:
+    text = (value or "").strip()
+    if not text:
+        return None
+    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)
+
+
+@contextmanager
+def _exclusive_file_lock(lock_path: str):
+    os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True)
+    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
+    try:
+        fcntl.flock(fd, fcntl.LOCK_EX)
+        yield fd
+    finally:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+        finally:
+            os.close(fd)
+
+
+def _read_json(path: str) -> dict[str, Any] | None:
+    if not path or not os.path.exists(path):
+        return None
+    try:
+        with open(path, encoding="utf-8") as handle:
+            data = json.load(handle)
+    except (OSError, json.JSONDecodeError):
+        return None
+    return data if isinstance(data, dict) else None
+
+
+def _write_json(path: str, data: dict[str, Any]) -> None:
+    parent = os.path.dirname(path) or "."
+    os.makedirs(parent, mode=0o700, exist_ok=True)
+    payload = json.dumps(data, indent=2, sort_keys=True) + "\n"
+    fd, temp_path = tempfile.mkstemp(prefix=".session-", suffix=".json", dir=parent)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(payload)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.chmod(temp_path, 0o600)
+        os.replace(temp_path, path)
+        try:
+            os.chmod(path, 0o600)
+        except OSError:
+            pass
+    finally:
+        if os.path.exists(temp_path):
+            try:
+                os.remove(temp_path)
+            except OSError:
+                pass
+
+
+def identity_match_reasons(
+    record: dict[str, Any] | None,
+    *,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    profile_identity: str | None = None,
+) -> list[str]:
+    """Return fail-closed reasons when durable record identity does not match."""
+    if record is None:
+        return []
+    reasons: list[str] = []
+    expected_profile = current_profile_identity(profile_identity=profile_identity)
+    stored_profile = (
+        (record.get("session_profile_lock") or record.get("profile_identity") or "")
+        .strip()
+    )
+    if stored_profile and expected_profile and stored_profile != expected_profile:
+        if expected_profile != "unknown-profile":
+            reasons.append(
+                "session state profile identity mismatch "
+                f"(stored={stored_profile!r}, active={expected_profile!r}; fail closed)"
+            )
+
+    for field, expected in (
+        ("remote", remote),
+        ("org", org),
+        ("repo", repo),
+    ):
+        want = (expected or "").strip()
+        have = (str(record.get(field) or "")).strip()
+        if want and have and want != have:
+            reasons.append(
+                f"session state {field} mismatch "
+                f"(stored={have!r}, expected={want!r}; fail closed)"
+            )
+
+    recorded_at = _parse_iso(record.get("recorded_at") or record.get("updated_at"))
+    if recorded_at is None:
+        reasons.append("session state missing recorded_at timestamp (fail closed)")
+    else:
+        age = _now_utc() - recorded_at
+        if age > timedelta(hours=ttl_hours()):
+            reasons.append(
+                f"session state expired after {ttl_hours():g}h (fail closed)"
+            )
+        if age < timedelta(0):
+            reasons.append("session state recorded_at is in the future (fail closed)")
+    return reasons
+
+
+def load_state(
+    *,
+    kind: str,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    profile_identity: str | None = None,
+    state_dir: str | None = None,
+) -> dict[str, Any] | None:
+    """Load durable state payload when identity checks pass."""
+    profile = current_profile_identity(profile_identity=profile_identity)
+    path = state_file_path(
+        kind=kind,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile,
+        state_dir=state_dir,
+    )
+    lock_path = f"{path}.lock"
+    with _exclusive_file_lock(lock_path):
+        envelope = _read_json(path)
+    if not envelope:
+        return None
+    payload = envelope.get("payload")
+    if not isinstance(payload, dict):
+        return None
+    # Identity fields live on both envelope and payload for convenience.
+    merged = dict(payload)
+    for key in (
+        "kind",
+        "remote",
+        "org",
+        "repo",
+        "profile_identity",
+        "session_profile_lock",
+        "recorded_at",
+        "updated_at",
+        "writer_pid",
+    ):
+        if key in envelope and key not in merged:
+            merged[key] = envelope[key]
+    reasons = identity_match_reasons(
+        merged,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile,
+    )
+    if reasons:
+        return None
+    return merged
+
+
+def save_state(
+    *,
+    kind: str,
+    payload: dict[str, Any] | None,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    profile_identity: str | None = None,
+    state_dir: str | None = None,
+) -> dict[str, Any] | None:
+    """Persist or clear durable state for the given session identity key."""
+    profile = current_profile_identity(
+        profile_name=payload.get("session_profile") if payload else None,
+        session_profile_lock=(
+            (payload or {}).get("session_profile_lock") or profile_identity
+        ),
+    )
+    # Prefer explicit args over payload fields for key location.
+    key_remote = remote if remote is not None else (payload or {}).get("remote")
+    key_org = org if org is not None else (payload or {}).get("org")
+    key_repo = repo if repo is not None else (payload or {}).get("repo")
+
+    root = _ensure_state_dir(state_dir)
+    path = state_file_path(
+        kind=kind,
+        remote=key_remote,
+        org=key_org,
+        repo=key_repo,
+        profile_identity=profile,
+        state_dir=root,
+    )
+    lock_path = f"{path}.lock"
+    with _exclusive_file_lock(lock_path):
+        if payload is None:
+            for candidate in (path, lock_path):
+                if os.path.exists(candidate):
+                    try:
+                        os.remove(candidate)
+                    except OSError:
+                        pass
+            return None
+
+        now = _now_utc().isoformat().replace("+00:00", "Z")
+        body = dict(payload)
+        body.setdefault("session_pid", os.getpid())
+        body["writer_pid"] = os.getpid()
+        body["profile_identity"] = profile
+        if not (body.get("session_profile_lock") or "").strip():
+            body["session_profile_lock"] = profile
+        body["recorded_at"] = body.get("recorded_at") or now
+        body["updated_at"] = now
+        if key_remote is not None:
+            body["remote"] = key_remote
+        if key_org is not None:
+            body["org"] = key_org
+        if key_repo is not None:
+            body["repo"] = key_repo
+
+        envelope = {
+            "kind": kind,
+            "remote": key_remote,
+            "org": key_org,
+            "repo": key_repo,
+            "profile_identity": profile,
+            "session_profile_lock": body.get("session_profile_lock"),
+            "recorded_at": body["recorded_at"],
+            "updated_at": body["updated_at"],
+            "writer_pid": body["writer_pid"],
+            "payload": body,
+        }
+        _write_json(path, envelope)
+        return dict(body)
+
+
+def clear_state(
+    *,
+    kind: str,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    profile_identity: str | None = None,
+    state_dir: str | None = None,
+) -> None:
+    save_state(
+        kind=kind,
+        payload=None,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile_identity,
+        state_dir=state_dir,
+    )
@@ -1,4 +1,4 @@
-"""Canonical review-merge workflow load proof for reviewer mutations (#389, #403)."""
+"""Canonical review-merge workflow load proof for reviewer mutations (#389, #403, #559)."""

 from __future__ import annotations

@@ -7,6 +7,7 @@ import os
 import re
 from pathlib import Path

+import mcp_session_state
 import review_workflow_boundary as boundary

 WORKFLOW_REL_PATH = (
@@ -88,17 +89,79 @@ def assess_prompt_conflict(prompt_text: str | None) -> tuple[bool, list[str]]:
    return bool(reasons), reasons


+def _session_binding_fields() -> dict:
+    """Capture profile identity used to share state across daemon processes."""
+    env_lock = (os.environ.get(mcp_session_state.SESSION_PROFILE_LOCK_ENV) or "").strip()
+    profile_name = (os.environ.get("GITEA_MCP_PROFILE") or "").strip()
+    remote = (os.environ.get("GITEA_MCP_REMOTE") or "").strip() or None
+    identity = mcp_session_state.current_profile_identity(
+        profile_name=profile_name,
+        session_profile_lock=env_lock,
+    )
+    return {
+        "session_profile": profile_name or identity,
+        "session_profile_lock": env_lock or identity,
+        "profile_identity": identity,
+        "remote": remote,
+    }
+
+
+def _persist_workflow_load(record: dict | None) -> dict | None:
+    """Write durable workflow-load proof (or clear it)."""
+    binding = _session_binding_fields()
+    if record is None:
+        mcp_session_state.clear_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            remote=binding.get("remote"),
+            profile_identity=binding.get("profile_identity"),
+        )
+        return None
+    payload = dict(record)
+    payload.update({
+        k: v for k, v in binding.items() if v is not None
+    })
+    return mcp_session_state.save_state(
+        kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+        payload=payload,
+        remote=payload.get("remote"),
+        org=payload.get("org"),
+        repo=payload.get("repo"),
+        profile_identity=payload.get("profile_identity"),
+    )
+
+
+def _load_durable_workflow_load() -> dict | None:
+    binding = _session_binding_fields()
+    return mcp_session_state.load_state(
+        kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+        remote=binding.get("remote"),
+        profile_identity=binding.get("profile_identity"),
+    )
+
+
+def _active_workflow_load() -> dict | None:
+    """Prefer in-process cache; fall back to durable shared state (#559)."""
+    global _REVIEW_WORKFLOW_LOAD
+    if _REVIEW_WORKFLOW_LOAD is not None:
+        return _REVIEW_WORKFLOW_LOAD
+    durable = _load_durable_workflow_load()
+    if durable is not None:
+        _REVIEW_WORKFLOW_LOAD = dict(durable)
+    return _REVIEW_WORKFLOW_LOAD
+
+
 def record_review_workflow_load(
    project_root: str,
    *,
    prompt_text: str | None = None,
 ) -> dict:
-    """Record in-process workflow load proof for the current MCP session."""
+    """Record workflow load proof for the current MCP session (durable + memory)."""
    global _REVIEW_WORKFLOW_LOAD
    meta = build_canonical_workflow_metadata(
        project_root, prompt_text=prompt_text)
    boundary_state = boundary.assess_boundary_status(project_root)
-    _REVIEW_WORKFLOW_LOAD = {
+    binding = _session_binding_fields()
+    record = {
        **meta,
        "session_pid": os.getpid(),
        "loaded": True,
@@ -107,7 +170,10 @@ def record_review_workflow_load(
        "pre_review_command_count": boundary_state.get("pre_review_command_count"),
        "boundary_violation_count": boundary_state.get("boundary_violation_count"),
        "boundary_reasons": list(boundary_state.get("reasons") or []),
+        **{k: v for k, v in binding.items() if v is not None},
    }
+    persisted = _persist_workflow_load(record)
+    _REVIEW_WORKFLOW_LOAD = dict(persisted or record)
    return dict(_REVIEW_WORKFLOW_LOAD)


@@ -115,12 +181,13 @@ def clear_review_workflow_load() -> None:
    """Test helper and review_pr session reset."""
    global _REVIEW_WORKFLOW_LOAD
    _REVIEW_WORKFLOW_LOAD = None
+    _persist_workflow_load(None)
    boundary.clear_pre_review_commands()


 def workflow_load_status(project_root: str | None = None) -> dict:
    """Non-throwing status for capability/runtime reports."""
-    load = _REVIEW_WORKFLOW_LOAD
+    load = _active_workflow_load()
    if load is None:
        return {
            "workflow_load_proof_present": False,
@@ -148,6 +215,8 @@ def workflow_load_status(project_root: str | None = None) -> dict:
        "prompt_conflicts_with_workflow": load.get(
            "prompt_conflicts_with_workflow"),
        "session_pid": load.get("session_pid"),
+        "writer_pid": load.get("writer_pid"),
+        "profile_identity": load.get("profile_identity"),
        "boundary_status": load.get("boundary_status"),
        "boundary_clean": load.get("boundary_clean"),
        "workflow_load_helper_result": boundary.workflow_load_helper_result(
@@ -160,13 +229,47 @@ def _session_validation_reasons(
    load: dict,
    project_root: str | None,
 ) -> list[str]:
+    """Validate durable/in-memory load proof for this session identity (#559)."""
    reasons: list[str] = []
-    if load.get("session_pid") != os.getpid():
+    # Cross-process daemon pools are allowed when profile identity matches.
+    # Reject only when the stored profile identity conflicts with this process.
+    binding = _session_binding_fields()
+    stored_identity = (
+        load.get("session_profile_lock")
+        or load.get("profile_identity")
+        or ""
+    ).strip()
+    active_identity = (binding.get("profile_identity") or "").strip()
+    if (
+        stored_identity
+        and active_identity
+        and stored_identity != active_identity
+        and active_identity != "unknown-profile"
+        and stored_identity != "unknown-profile"
+    ):
        reasons.append(
-            "workflow load proof was recorded in a different process "
-            "(fail closed)"
+            "workflow load proof profile identity mismatch "
+            f"(stored={stored_identity!r}, active={active_identity!r}; fail closed)"
        )
        return reasons
+
+    # Expired durable records are treated as absent.
+    identity_reasons = mcp_session_state.identity_match_reasons(
+        load,
+        remote=binding.get("remote") or load.get("remote"),
+        org=load.get("org"),
+        repo=load.get("repo"),
+        profile_identity=active_identity or stored_identity,
+    )
+    # Filter out remote-mismatch noise when remote was not bound at load time.
+    for reason in identity_reasons:
+        if "missing recorded_at" in reason or "expired" in reason or "future" in reason:
+            reasons.append(reason)
+        elif "profile identity mismatch" in reason:
+            reasons.append(reason)
+    if reasons:
+        return reasons
+
    if load.get("prompt_conflicts_with_workflow"):
        reasons.extend(load.get("prompt_conflict_reasons") or [
            "active prompt conflicts with loaded review-merge workflow"
@@ -196,7 +299,8 @@ def review_workflow_load_blockers(
 ) -> list[str]:
    """Reasons reviewer mutations must fail closed."""
    boundary_reasons = boundary.boundary_blockers(project_root)
-    if boundary_reasons and _REVIEW_WORKFLOW_LOAD is None:
+    load = _active_workflow_load()
+    if boundary_reasons and load is None:
        return boundary_reasons
    status = workflow_load_status(project_root)
    if not status.get("workflow_load_proof_present"):
@@ -214,4 +318,4 @@ def recovery_handoff_without_replay() -> list[str]:
        "Do not call gitea_submit_pr_review, gitea_mark_final_review_decision, "
        "or gitea_merge_pr until workflow-load proof is present.",
        "Do not include approve/merge replay commands in the recovery handoff.",
-    ]
+    ]
@@ -0,0 +1,68 @@
+# Controller issue-acceptance prompt
+
+Use after a PR merges when auditing whether the linked issue is truly complete.
+
+```text
+Audit issue #<N> against its acceptance criteria after merged PR #<PR>.
+Post a Controller Issue Acceptance comment with checked criteria, validation
+reviewed, controller decision, next actor, and paste-ready next prompt.
+Do not mark the issue accepted unless every required criterion is satisfied.
+```
+
+## Comment template
+
+```text
+## Controller Issue Acceptance
+
+STATE:
+<accepted | more-work-required | needs-tests | needs-docs | needs-feature-enhancement | needs-follow-up-issue | blocked>
+
+WHO_IS_NEXT:
+<author | reviewer | merger | reconciler | controller | user>
+
+NEXT_ACTION:
+<one sentence>
+
+NEXT_PROMPT:
+<paste-ready prompt for the next LLM>
+
+ISSUE:
+#...
+
+MERGED_PR:
+#...
+
+MERGE_COMMIT:
+<40-character SHA>
+
+ACCEPTANCE_CRITERIA_CHECKED:
+- [x] ...
+- [ ] ...
+
+VALIDATION_REVIEWED:
+<tests/proofs reviewed>
+
+CONTROLLER_DECISION:
+<accepted or rejected>
+
+WHY:
+<reasoning>
+
+MISSING_WORK:
+<none, or exact missing work>
+
+FOLLOW_UP_ISSUES:
+<none, or issue list to create>
+
+BLOCKERS:
+<none, or exact blockers>
+
+LAST_UPDATED_BY:
+<identity/profile/date>
+```
+
+## Rejection paths
+
+When rejecting completion, `STATE` must name the gap (`needs-tests`,
+`needs-docs`, `more-work-required`, etc.), `MISSING_WORK` must be explicit,
+and `NEXT_PROMPT` must be ready for the next author session.
@@ -375,6 +375,24 @@ Include:
 * confirmation that no normal review, approval, request-changes, or merge was
  performed

+## 18A. Reconciler close proof is enforced (#306)
+
+When a reconciler run closes a PR, the final-report validator
+(`final_report_validator` rule `reconcile.close_proof_fields`) **blocks** the
+handoff unless it carries all four close proofs. The prompt is guidance; the
+MCP validator is the authority.
+
+A close is detected from `PRs closed: #<n>` (or a session close lock). Once a
+close is reported, the handoff must include:
+
+* `Capabilities proven:` naming `gitea.pr.close` — the exact close capability
+* `Ancestor proof:` — the landed/ancestor proof for the closed PR
+* `PRs closed:` — the PR close result (the closed PR number)
+* `Linked issue live status:` (or `Issues closed:`) — the linked-issue result
+
+Comment-only and blocked reconciliations (no PR close) are unaffected: the rule
+returns no finding when nothing was closed.
+
 ## 19. Local artifact and report consistency rule

 Do not create local walkthrough, notes, markdown, JSON, or report artifacts
@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""Live health-check script to verify Gitea MCP namespace connections.
+
+Spawns the MCP server processes as defined in the IDE's global config,
+performs the JSON-RPC handshake, and queries the tools list to verify
+that the connection is fully operational and doesn't return EOF.
+"""
+import json
+import os
+import subprocess
+import sys
+
+def test_connection(name, config):
+    print(f"Testing MCP connection for '{name}'...")
+    command = config.get("command")
+    args = config.get("args", [])
+    env = config.get("env", {})
+    
+    # Merge current environment
+    run_env = os.environ.copy()
+    run_env.update(env)
+    
+    # Spawn subprocess
+    try:
+        proc = subprocess.Popen(
+            [command] + args,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            bufsize=1,
+            env=run_env
+        )
+    except Exception as e:
+        print(f"  [FAIL] Failed to spawn process: {e}")
+        return False
+
+    # Send initialize request
+    init_req = {
+        "jsonrpc": "2.0",
+        "method": "initialize",
+        "params": {
+            "protocolVersion": "2024-11-05",
+            "capabilities": {},
+            "clientInfo": {"name": "healthcheck", "version": "1.0"}
+        },
+        "id": 1
+    }
+    
+    try:
+        proc.stdin.write(json.dumps(init_req) + "\n")
+        proc.stdin.flush()
+        
+        # Read response
+        line = proc.stdout.readline()
+        if not line:
+            stderr_content = proc.stderr.read()
+            print(f"  [FAIL] Received EOF from process. Stderr:\n{stderr_content}")
+            proc.terminate()
+            return False
+            
+        print(f"  [OK] Received initialize response: {line.strip()[:150]}...")
+        
+        # Send initialized notification
+        init_notif = {
+            "jsonrpc": "2.0",
+            "method": "notifications/initialized"
+        }
+        proc.stdin.write(json.dumps(init_notif) + "\n")
+        proc.stdin.flush()
+        
+        # Send tools/list request
+        list_req = {
+            "jsonrpc": "2.0",
+            "method": "tools/list",
+            "params": {},
+            "id": 2
+        }
+        proc.stdin.write(json.dumps(list_req) + "\n")
+        proc.stdin.flush()
+        
+        line = proc.stdout.readline()
+        if not line:
+            print("  [FAIL] Received EOF on tools/list request.")
+            proc.terminate()
+            return False
+            
+        res = json.loads(line)
+        if "error" in res:
+            print(f"  [FAIL] Server returned error: {res['error']}")
+            proc.terminate()
+            return False
+            
+        tools = res.get("result", {}).get("tools", [])
+        tool_names = [t.get("name") for t in tools]
+        print(f"  [OK] Successfully retrieved {len(tool_names)} tools: {tool_names[:5]}...")
+        
+        proc.terminate()
+        return True
+    except Exception as e:
+        print(f"  [FAIL] Error during handshake: {e}")
+        proc.terminate()
+        return False
+
+def main():
+    config_path = "/Users/jasonwalker/.gemini/config/mcp_config.json"
+    try:
+        with open(config_path) as f:
+            mcp_config = json.load(f)
+    except Exception as e:
+        print(f"Failed to load mcp_config.json: {e}")
+        sys.exit(1)
+        
+    servers = mcp_config.get("mcpServers", {})
+    failed = False
+    for name in ["gitea-author", "gitea-reviewer"]:
+        if name in servers:
+            if not test_connection(name, servers[name]):
+                failed = True
+        else:
+            print(f"Server '{name}' not found in mcp_config.json")
+            
+    if failed:
+        sys.exit(1)
+    else:
+        print("All Gitea MCP connection tests passed!")
+
+if __name__ == "__main__":
+    main()
@@ -18,14 +18,25 @@ def _reset_mutation_authority(monkeypatch):
    explicitly.
    """
    monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
+    # Isolate durable session-state files so tests never share host cache (#559).
+    import tempfile
+
+    _state_tmp = tempfile.TemporaryDirectory(prefix="gitea-session-state-")
+    monkeypatch.setenv("GITEA_MCP_SESSION_STATE_DIR", _state_tmp.name)
    try:
        import mcp_server
    except Exception:
+        _state_tmp.cleanup()
        yield
        return
    monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
    monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
    monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
+    try:
+        import review_workflow_load
+        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
+    except Exception:
+        pass
    try:
        import capability_stop_terminal
        capability_stop_terminal.clear()
@@ -37,3 +48,13 @@ def _reset_mutation_authority(monkeypatch):
        capability_stop_terminal.clear()
    except Exception:
        pass
+    try:
+        import review_workflow_load
+        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
+    except Exception:
+        pass
+    try:
+        mcp_server._REVIEW_DECISION_LOCK = None
+    except Exception:
+        pass
+    _state_tmp.cleanup()
@@ -498,6 +498,81 @@ class TestCanonicalReconcileSchema(unittest.TestCase):
        )


+class TestReconcilerCloseProof(unittest.TestCase):
+    """Issue #306: a reconciler PR close must carry its proof fields."""
+
+    def _closed_report(self, **kwargs):
+        # A report that claims PR #99 was closed via the reconciler path.
+        report = (
+            _reconcile_handoff(**kwargs)
+            .replace(
+                "- Capabilities proven: gitea.read, gitea.pr.comment",
+                "- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close",
+            )
+            .replace("- Missing capabilities: gitea.pr.close", "- Missing capabilities: none")
+            .replace("- PRs closed: none", "- PRs closed: #99")
+            .replace(
+                "- Blocker: missing gitea.pr.close capability", "- Blocker: none"
+            )
+        )
+        return report
+
+    def test_full_proof_close_passes(self):
+        result = assess_final_report_validator(
+            self._closed_report(), "reconcile_already_landed"
+        )
+        self.assertEqual(result["grade"], "A")
+        self.assertFalse(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+    def test_close_without_capability_proof_blocks(self):
+        # Claims PRs closed: #99 but never proves gitea.pr.close capability.
+        report = _reconcile_handoff().replace("- PRs closed: none", "- PRs closed: #99")
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+    def test_close_without_ancestor_proof_blocks(self):
+        report = self._closed_report(drop=("Ancestor proof",))
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+    def test_close_without_linked_issue_result_blocks(self):
+        report = self._closed_report(drop=("Linked issue live status", "Issues closed"))
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+    def test_close_lock_requires_proof_even_if_text_says_none(self):
+        # Session lock proves a PR was closed; the report omits close proof.
+        result = assess_final_report_validator(
+            _reconcile_handoff(),
+            "reconcile_already_landed",
+            reconciler_close_lock={"pr_closed": True},
+        )
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+    def test_comment_only_reconcile_needs_no_close_proof(self):
+        result = assess_final_report_validator(
+            _reconcile_handoff(), "reconcile_already_landed"
+        )
+        self.assertEqual(result["grade"], "A")
+        self.assertFalse(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+
 class TestEntryPoint(unittest.TestCase):
    def test_unknown_task_kind_blocks(self):
        result = assess_final_report_validator("report", "unknown_mode")
@@ -0,0 +1,183 @@
+"""Tests for controller issue-acceptance gate (#500)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_acceptance_gate  # noqa: E402
+from final_report_validator import assess_final_report_validator  # noqa: E402
+
+
+def _accepted_comment(**overrides):
+    lines = [
+        "## Controller Issue Acceptance",
+        "",
+        "STATE:",
+        "accepted",
+        "",
+        "WHO_IS_NEXT:",
+        "controller",
+        "",
+        "NEXT_ACTION:",
+        "Close the tracker follow-up after verification.",
+        "",
+        "NEXT_PROMPT:",
+        "Verify deployment checklist for issue #500 and post acceptance.",
+        "",
+        "ISSUE:",
+        "#500",
+        "",
+        "MERGED_PR:",
+        "#503",
+        "",
+        "MERGE_COMMIT:",
+        "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "",
+        "ACCEPTANCE_CRITERIA_CHECKED:",
+        "- [x] documentation added",
+        "- [x] validator tests added",
+        "",
+        "VALIDATION_REVIEWED:",
+        "pytest tests/test_issue_acceptance_gate.py -q",
+        "",
+        "CONTROLLER_DECISION:",
+        "accepted",
+        "",
+        "WHY:",
+        "All acceptance criteria satisfied with proof.",
+        "",
+        "MISSING_WORK:",
+        "none",
+        "",
+        "FOLLOW_UP_ISSUES:",
+        "none",
+        "",
+        "BLOCKERS:",
+        "none",
+        "",
+        "LAST_UPDATED_BY:",
+        "jcwalker3 / prgs-controller / 2026-07-08",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        text = text.replace(f"{key}:\n", f"{key}:\n{value}\n", 1)
+    return text
+
+
+def _rejection_comment(state="needs-tests"):
+    text = _accepted_comment()
+    text = text.replace("STATE:\naccepted", f"STATE:\n{state}")
+    text = text.replace(
+        "CONTROLLER_DECISION:\naccepted",
+        "CONTROLLER_DECISION:\nrejected",
+    )
+    text = text.replace(
+        "ACCEPTANCE_CRITERIA_CHECKED:\n- [x] documentation added\n- [x] validator tests added",
+        "ACCEPTANCE_CRITERIA_CHECKED:\n- [ ] regression tests for rejection paths",
+    )
+    text = text.replace(
+        "MISSING_WORK:\nnone",
+        "MISSING_WORK:\nAdd regression tests for controller rejection paths.",
+    )
+    text = text.replace(
+        "NEXT_PROMPT:\nVerify deployment checklist for issue #500 and post acceptance.",
+        "NEXT_PROMPT:\nImplement the missing tests for issue #500 and reopen the PR.",
+    )
+    return text
+
+
+class TestControllerAcceptanceComment(unittest.TestCase):
+    def test_accepted_comment_valid(self):
+        result = issue_acceptance_gate.validate_controller_acceptance_comment(
+            _accepted_comment()
+        )
+        self.assertTrue(result["valid"], result["reasons"])
+
+    def test_missing_who_is_next_rejected(self):
+        text = _accepted_comment().replace("WHO_IS_NEXT:\ncontroller\n", "WHO_IS_NEXT:\n\n")
+        result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
+        self.assertFalse(result["valid"])
+        self.assertTrue(any("WHO_IS_NEXT" in r for r in result["reasons"]))
+
+    def test_missing_next_prompt_on_rejection(self):
+        text = _rejection_comment()
+        text = text.replace(
+            "NEXT_PROMPT:\nImplement the missing tests for issue #500 and reopen the PR.\n",
+            "NEXT_PROMPT:\n\n",
+        )
+        result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
+        self.assertFalse(result["valid"])
+        self.assertTrue(any("NEXT_PROMPT" in r for r in result["reasons"]))
+
+    def test_vague_merge_only_completion_detected(self):
+        text = "PR merged. Issue complete."
+        self.assertTrue(issue_acceptance_gate.claims_merge_only_issue_complete(text))
+
+    def test_rejection_paths_require_missing_work(self):
+        for state in (
+            "more-work-required",
+            "needs-tests",
+            "needs-docs",
+            "needs-feature-enhancement",
+            "needs-follow-up-issue",
+        ):
+            text = _rejection_comment(state=state).replace(
+                "MISSING_WORK:\nAdd regression tests for controller rejection paths.\n",
+                "MISSING_WORK:\n\n",
+            )
+            result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
+            self.assertFalse(result["valid"], state)
+            self.assertTrue(any("MISSING_WORK" in r for r in result["reasons"]))
+
+    def test_accepted_requires_checked_criteria(self):
+        text = _accepted_comment().replace(
+            "ACCEPTANCE_CRITERIA_CHECKED:\n- [x] documentation added\n- [x] validator tests added\n",
+            "ACCEPTANCE_CRITERIA_CHECKED:\n- [ ] documentation added\n",
+        )
+        result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
+        self.assertFalse(result["valid"])
+        self.assertTrue(any("checked acceptance criterion" in r for r in result["reasons"]))
+
+
+class TestFinalReportIntegration(unittest.TestCase):
+    def test_merge_only_complete_blocks_work_issue_report(self):
+        report = (
+            "## Controller Handoff\n\n"
+            "- Task: work issue #500\n"
+            "- Merge result: merged\n"
+            "- Current status: PR merged; issue complete\n"
+        )
+        result = assess_final_report_validator(report, "work_issue")
+        self.assertTrue(
+            any(
+                f["rule_id"] == "shared.issue_acceptance_gate"
+                for f in result["findings"]
+            ),
+            result["findings"],
+        )
+
+    def test_pending_acceptance_allowed(self):
+        report = (
+            "## Controller Handoff\n\n"
+            "- Task: work issue #500\n"
+            "- Merge result: merged\n"
+            "- Current status: controller acceptance pending\n"
+        )
+        result = assess_final_report_validator(report, "work_issue")
+        self.assertFalse(
+            any(
+                f["rule_id"] == "shared.issue_acceptance_gate"
+                for f in result["findings"]
+            ),
+            result["findings"],
+        )
+
+    def test_accepted_block_allows_complete_claim(self):
+        report = _accepted_comment() + "\n\nIssue is complete."
+        gate = issue_acceptance_gate.validate_final_report_issue_acceptance(report)
+        self.assertTrue(gate["valid"], gate["reasons"])
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,207 @@
+"""Tests for durable MCP session state shared across daemon processes (#559)."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys_path_root = str(Path(__file__).resolve().parent.parent)
+import sys
+
+if sys_path_root not in sys.path:
+    sys.path.insert(0, sys_path_root)
+
+import mcp_session_state
+import review_workflow_load
+import mcp_server
+
+
+class TestMcpSessionStateStore(unittest.TestCase):
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self.state_dir = self._tmpdir.name
+        self._env = patch.dict(
+            os.environ,
+            {
+                mcp_session_state.STATE_DIR_ENV: self.state_dir,
+                mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+                "GITEA_MCP_PROFILE": "prgs-reviewer",
+            },
+            clear=False,
+        )
+        self._env.start()
+
+    def tearDown(self):
+        self._env.stop()
+        self._tmpdir.cleanup()
+
+    def test_round_trip_same_identity(self):
+        saved = mcp_session_state.save_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            payload={"loaded": True, "workflow_hash": "abc123def456"},
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertIsNotNone(saved)
+        self.assertEqual(saved["workflow_hash"], "abc123def456")
+        self.assertIn("recorded_at", saved)
+
+        loaded = mcp_session_state.load_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertIsNotNone(loaded)
+        self.assertEqual(loaded["workflow_hash"], "abc123def456")
+        self.assertEqual(loaded["profile_identity"], "prgs-reviewer")
+
+    def test_profile_mismatch_returns_none(self):
+        mcp_session_state.save_state(
+            kind=mcp_session_state.KIND_DECISION_LOCK,
+            payload={"final_review_decision_ready": True, "remote": "prgs"},
+            remote="prgs",
+        )
+        with patch.dict(
+            os.environ,
+            {mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"},
+            clear=False,
+        ):
+            loaded = mcp_session_state.load_state(
+                kind=mcp_session_state.KIND_DECISION_LOCK,
+                remote="prgs",
+            )
+        self.assertIsNone(loaded)
+
+    def test_clear_removes_file(self):
+        mcp_session_state.save_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            payload={"loaded": True},
+            remote="prgs",
+        )
+        path = mcp_session_state.state_file_path(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            remote="prgs",
+            profile_identity="prgs-reviewer",
+            state_dir=self.state_dir,
+        )
+        self.assertTrue(os.path.exists(path))
+        mcp_session_state.clear_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            remote="prgs",
+            profile_identity="prgs-reviewer",
+        )
+        self.assertFalse(os.path.exists(path))
+
+    def test_files_are_private_mode(self):
+        mcp_session_state.save_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            payload={"loaded": True},
+            remote="prgs",
+        )
+        path = mcp_session_state.state_file_path(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            remote="prgs",
+            profile_identity="prgs-reviewer",
+            state_dir=self.state_dir,
+        )
+        mode = os.stat(path).st_mode & 0o777
+        self.assertEqual(mode, 0o600)
+
+
+class TestWorkflowLoadCrossProcess(unittest.TestCase):
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self._env = patch.dict(
+            os.environ,
+            {
+                mcp_session_state.STATE_DIR_ENV: self._tmpdir.name,
+                mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+                "GITEA_MCP_PROFILE": "prgs-reviewer",
+                "GITEA_MCP_REMOTE": "prgs",
+            },
+            clear=False,
+        )
+        self._env.start()
+        review_workflow_load.clear_review_workflow_load()
+
+    def tearDown(self):
+        review_workflow_load.clear_review_workflow_load()
+        self._env.stop()
+        self._tmpdir.cleanup()
+
+    def test_different_pid_same_profile_accepted(self):
+        root = str(Path(__file__).resolve().parent.parent)
+        recorded = review_workflow_load.record_review_workflow_load(root)
+        self.assertTrue(recorded["loaded"])
+
+        # Simulate another daemon process: clear memory, keep durable file,
+        # and change apparent PID identity only (profile remains the same).
+        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
+        status = review_workflow_load.workflow_load_status(root)
+        self.assertTrue(status["workflow_load_proof_present"])
+        self.assertTrue(
+            status["workflow_load_valid"],
+            msg=status.get("reasons"),
+        )
+
+    def test_profile_mismatch_blocks(self):
+        root = str(Path(__file__).resolve().parent.parent)
+        review_workflow_load.record_review_workflow_load(root)
+        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
+        with patch.dict(
+            os.environ,
+            {mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"},
+            clear=False,
+        ):
+            # Durable load uses active identity; mismatched profile key misses.
+            status = review_workflow_load.workflow_load_status(root)
+        self.assertFalse(status["workflow_load_proof_present"])
+
+
+class TestDecisionLockCrossProcess(unittest.TestCase):
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self._env = patch.dict(
+            os.environ,
+            {
+                mcp_session_state.STATE_DIR_ENV: self._tmpdir.name,
+                mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+            },
+            clear=False,
+        )
+        self._env.start()
+        mcp_server._save_review_decision_lock(None)
+        review_workflow_load.clear_review_workflow_load()
+
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+        review_workflow_load.clear_review_workflow_load()
+        self._env.stop()
+        self._tmpdir.cleanup()
+
+    def test_decision_lock_survives_memory_clear(self):
+        with patch.object(mcp_server, "get_profile", return_value={
+            "profile_name": "prgs-reviewer",
+        }):
+            mcp_server.init_review_decision_lock("prgs", "review_pr", force=True)
+            lock = mcp_server._load_review_decision_lock()
+            self.assertIsNotNone(lock)
+            self.assertEqual(lock["remote"], "prgs")
+            self.assertFalse(lock["final_review_decision_ready"])
+
+            # New process: memory empty, durable state remains.
+            mcp_server._REVIEW_DECISION_LOCK = None
+            restored = mcp_server._load_review_decision_lock()
+            self.assertIsNotNone(restored)
+            self.assertEqual(restored["remote"], "prgs")
+            reasons = mcp_server._review_decision_session_reasons(restored)
+            self.assertEqual(reasons, [])
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,71 @@
+import os
+import unittest
+from unittest.mock import patch, MagicMock
+from datetime import datetime
+
+import gitea_mcp_server
+
+class TestMcpStaleRuntime(unittest.TestCase):
+    @patch("subprocess.run")
+    @patch("os.path.getmtime")
+    @patch("os.path.exists")
+    @patch("os.getpid")
+    @patch.dict("os.environ", {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"})
+    def test_stale_and_missing_runtimes(self, mock_getpid, mock_exists, mock_getmtime, mock_run):
+        # Setup mocks
+        mock_getpid.return_value = 12345
+        mock_exists.return_value = True
+        
+        # Code modification time: Jul 8 2026, 14:00:00
+        code_time = datetime(2026, 7, 8, 14, 0, 0)
+        mock_getmtime.return_value = code_time.timestamp()
+        
+        # Mock ps -ax output
+        # PID 12345 is self (started at 13:00:00 - stale)
+        # PID 54321 is prgs-author (started at 15:00:00 - fresh)
+        # prgs-reviewer is missing
+        ps_output = (
+            "  PID LSTART COMMAND\n"
+            "12345 Wed Jul  8 13:00:00 2026 /path/to/python mcp_server.py\n"
+            "54321 Wed Jul  8 15:00:00 2026 /path/to/python mcp_server.py\n"
+        )
+        
+        mock_run_ps = MagicMock()
+        mock_run_ps.stdout = ps_output
+        
+        # Mock env output for ps eww
+        mock_run_env12345 = MagicMock()
+        mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler"
+        
+        mock_run_env54321 = MagicMock()
+        mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author"
+        
+        def side_effect(args, **kwargs):
+            if args[0] == "ps" and "eww" in args:
+                pid = args[2]
+                if pid == "12345":
+                    return mock_run_env12345
+                elif pid == "54321":
+                    return mock_run_env54321
+            elif args[0] == "ps":
+                return mock_run_ps
+            raise ValueError(f"Unexpected subprocess run args: {args}")
+            
+        mock_run.side_effect = side_effect
+        
+        # Test 1: required reviewer role matching prgs-reviewer (which is missing)
+        reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("review_pr", ["prgs-reviewer"])
+        
+        self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons))
+        self.assertTrue(any("stale-runtime: None of the matching profiles for task" in r for r in reasons))
+        
+        # Test 2: required author role matching prgs-author (which is fresh)
+        reasons_author = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
+        
+        # Still contains self stale error
+        self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons_author))
+        # But does NOT contain missing author profile error
+        self.assertFalse(any("None of the matching profiles for task" in r for r in reasons_author))
+
+if __name__ == "__main__":
+    unittest.main()
@@ -33,12 +33,38 @@ class TestReviewWorkflowLoadModule(unittest.TestCase):
        self.assertTrue(status["workflow_load_proof_present"])
        self.assertTrue(status["workflow_load_valid"])

-    def test_stale_session_pid_blocks(self):
+    def test_profile_identity_mismatch_blocks(self):
+        """#559: different daemon PIDs are OK; profile identity mismatch is not."""
+        import tempfile
+        from unittest.mock import patch
+
+        import mcp_session_state
+
        root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
-        review_workflow_load.record_review_workflow_load(root)
-        review_workflow_load._REVIEW_WORKFLOW_LOAD["session_pid"] = 0
-        blockers = review_workflow_load.review_workflow_load_blockers(root)
-        self.assertTrue(any("different process" in b for b in blockers))
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch.dict(
+                os.environ,
+                {
+                    mcp_session_state.STATE_DIR_ENV: tmp,
+                    mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+                },
+                clear=False,
+            ):
+                review_workflow_load.clear_review_workflow_load()
+                review_workflow_load.record_review_workflow_load(root)
+                # Corrupt the in-memory profile identity while keeping PID.
+                review_workflow_load._REVIEW_WORKFLOW_LOAD[
+                    "session_profile_lock"
+                ] = "other-profile"
+                review_workflow_load._REVIEW_WORKFLOW_LOAD[
+                    "profile_identity"
+                ] = "other-profile"
+                blockers = review_workflow_load.review_workflow_load_blockers(root)
+                self.assertTrue(
+                    any("profile identity mismatch" in b for b in blockers),
+                    msg=blockers,
+                )
+                review_workflow_load.clear_review_workflow_load()

    def test_prompt_conflict_detected(self):
        conflict, reasons = review_workflow_load.assess_prompt_conflict(
@@ -0,0 +1,94 @@
+"""Tests for web UI lease visibility (#433)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.testclient import TestClient
+
+from webui.app import create_app
+from webui.lease_loader import (
+    _duplicate_branch_warnings,
+    _duplicate_pr_warnings,
+    load_lease_snapshot,
+    parse_reviewer_lease_comment,
+    snapshot_to_dict,
+)
+
+
+class TestLeaseLoader(unittest.TestCase):
+    def test_parse_reviewer_lease_comment(self):
+        body = (
+            "<!-- mcp-review-lease:v1 -->\n"
+            "pr: #42\n"
+            "reviewer_identity: sysadmin\n"
+            "profile: prgs-reviewer\n"
+            "phase: validating\n"
+            "expires_at: 2026-07-07T20:00:00Z\n"
+        )
+        parsed = parse_reviewer_lease_comment(body)
+        self.assertIsNotNone(parsed)
+        assert parsed is not None
+        self.assertEqual(parsed["pr_number"], 42)
+        self.assertEqual(parsed["phase"], "validating")
+
+    def test_duplicate_pr_warnings(self):
+        raw_prs = [
+            {"number": 1, "title": "Closes #99", "body": ""},
+            {"number": 2, "title": "fixes #99", "body": ""},
+        ]
+        warnings = _duplicate_pr_warnings(raw_prs)
+        self.assertEqual(len(warnings), 1)
+        self.assertEqual(warnings[0].issue_number, 99)
+        self.assertEqual(warnings[0].pr_numbers, (1, 2))
+
+    def test_duplicate_branch_warnings(self):
+        warnings = _duplicate_branch_warnings([
+            "feat/issue-12-a",
+            "feat/issue-12-b",
+            "master",
+        ])
+        self.assertEqual(len(warnings), 1)
+        self.assertEqual(warnings[0].issue_number, 12)
+
+    def test_load_snapshot_with_injected_fetch(self):
+        def fetch_prs(_h, _o, _r, _a):
+            return ([], None)
+
+        def fetch_issues(_h, _o, _r, _a):
+            return ([], None)
+
+        snapshot = load_lease_snapshot(
+            fetch_prs=fetch_prs,
+            fetch_issues=fetch_issues,
+            fetch_comments=lambda *_a, **_k: [],
+            issue_lock_path=str(Path("/nonexistent/lock.json")),
+        )
+        data = snapshot_to_dict(snapshot)
+        self.assertIn("claim_inventory", data)
+        self.assertIn("collision_history", data)
+
+
+class TestLeaseRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_leases_page_renders(self):
+        response = self.client.get("/leases")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Leases", response.text)
+        self.assertIn("Collision warnings", response.text)
+        self.assertNotIn("child issue", response.text.lower())
+
+    def test_api_leases_json(self):
+        response = self.client.get("/api/leases")
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertIn("claim_inventory", data)
+        self.assertIn("duplicate_prs", data)
+        self.assertIn("reviewer_leases", data)
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -46,10 +46,14 @@ class TestWebuiSkeleton(unittest.TestCase):
        self.assertEqual(response.status_code, 200)
        self.assertIn("Gitea-Tools", response.text)

+    def test_leases_is_implemented(self):
+        response = self.client.get("/leases")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Leases", response.text)
+        self.assertIn("Collision warnings", response.text)
+
    def test_extra_stub_routes(self):
-        for path in ("/worktrees", "/leases"):
-            with self.subTest(path=path):
-                self.assertEqual(self.client.get(path).status_code, 200)
+        self.assertEqual(self.client.get("/worktrees").status_code, 200)

    def test_post_is_rejected(self):
        response = self.client.post("/health")
@@ -62,10 +66,10 @@ class TestWebuiSkeleton(unittest.TestCase):
        self.assertIn("Live queue", response.text)

    def test_nav_links_on_all_pages(self):
-        for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit"):
+        for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
            with self.subTest(path=path):
                text = self.client.get(path).text
-                for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"):
+                for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
                    self.assertIn(f'href="{href}"', text)


@@ -14,6 +14,8 @@ from webui.project_registry import find_project, load_registry, registry_to_dict
 from webui.project_views import render_project_detail, render_projects_list
 from webui.prompt_library import find_prompt, library_to_dict
 from webui.prompt_views import render_prompt_detail, render_prompts_page
+from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
+from webui.lease_views import render_leases_page
 from webui.queue_loader import load_queue_snapshot, snapshot_to_dict
 from webui.queue_views import render_queue_page

@@ -141,10 +143,12 @@ async def worktrees(_request: Request) -> HTMLResponse:


 async def leases(_request: Request) -> HTMLResponse:
-    return _stub_page(
-        "Leases",
-        "Lease visibility will show active issue and reviewer PR leases.",
-    )
+    snapshot = load_lease_snapshot()
+    return HTMLResponse(render_leases_page(snapshot))
+
+
+async def api_leases(_request: Request) -> JSONResponse:
+    return JSONResponse(lease_snapshot_to_dict(load_lease_snapshot()))


 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
@@ -175,6 +179,7 @@ def create_app() -> Starlette:
            Route("/audit", audit, methods=["GET"]),
            Route("/worktrees", worktrees, methods=["GET"]),
            Route("/leases", leases, methods=["GET"]),
+            Route("/api/leases", api_leases, methods=["GET"]),
        ],
        exception_handlers={405: method_not_allowed},
    )
@@ -0,0 +1,330 @@
+"""Lease and collision visibility for the web UI (#433)."""
+
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+from dataclasses import dataclass
+from typing import Any, Callable
+from urllib.parse import urlparse
+
+from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
+from issue_claim_heartbeat import build_claim_inventory
+from merged_cleanup_reconcile import ISSUE_LOCK_FILE, read_issue_lock
+
+from webui.project_registry import ProjectRecord, load_registry
+from webui.queue_loader import _extract_linked_issue, _fetch_issues, _fetch_prs
+
+_REVIEWER_LEASE_MARKER = "<!-- mcp-review-lease:v1 -->"
+_REVIEWER_FIELD_RE = re.compile(
+    r"^\s*([a-z_]+)\s*:\s*(.+?)\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_ISSUE_BRANCH_RE = re.compile(r"issue-(\d+)", re.IGNORECASE)
+
+_COLLISION_HISTORY = (
+    {"number": 267, "title": "Author work leases"},
+    {"number": 268, "title": "Issue claim heartbeat leases"},
+    {"number": 400, "title": "Early duplicate-work detection"},
+    {"number": 407, "title": "Per-PR reviewer leases"},
+)
+
+
+@dataclass(frozen=True)
+class CollisionWarning:
+    kind: str
+    message: str
+    issue_number: int | None = None
+    pr_numbers: tuple[int, ...] = ()
+
+
+@dataclass(frozen=True)
+class LeaseSnapshot:
+    project_id: str
+    repo_label: str
+    issue_lock: dict[str, Any] | None
+    claim_inventory: dict[str, Any]
+    reviewer_leases: tuple[dict[str, Any], ...]
+    duplicate_prs: tuple[CollisionWarning, ...]
+    duplicate_branches: tuple[CollisionWarning, ...]
+    collision_history: tuple[dict[str, Any], ...]
+    fetch_error: str | None = None
+
+
+def _repo_root() -> str:
+    override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
+    if override:
+        return os.path.realpath(override)
+    return os.path.realpath(os.path.join(os.path.dirname(__file__), ".."))
+
+
+def _host_from_url(remote_host: str) -> str:
+    parsed = urlparse(remote_host.strip())
+    return parsed.netloc or remote_host.strip().rstrip("/")
+
+
+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_reviewer_lease_comment(body: str) -> dict[str, Any] | None:
+    text = body or ""
+    if _REVIEWER_LEASE_MARKER not in text:
+        return None
+    fields: dict[str, str] = {}
+    for match in _REVIEWER_FIELD_RE.finditer(text):
+        fields[match.group(1).strip().lower()] = match.group(2).strip()
+    if not fields:
+        return None
+    return {
+        "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"),
+        "phase": (fields.get("phase") or "").strip().lower() or None,
+        "expires_at": fields.get("expires_at"),
+        "blocker": fields.get("blocker"),
+    }
+
+
+def _fetch_comments(
+    host: str,
+    org: str,
+    repo: str,
+    auth: str,
+    *,
+    issue_number: int,
+) -> list[dict]:
+    url = f"{repo_api_url(host, org, repo)}/issues/{issue_number}/comments"
+    comments: list[dict] = []
+    page = 1
+    while page <= 10:
+        raw_page, meta = api_fetch_page(url, auth, page=page, limit=50)
+        comments.extend(raw_page)
+        if meta.get("is_final_page"):
+            break
+        page += 1
+    return comments
+
+
+def _list_local_branch_names(project_root: str) -> list[str]:
+    result = subprocess.run(
+        ["git", "-C", project_root, "branch", "--list", "--format=%(refname:short)"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        return []
+    return [line.strip() for line in (result.stdout or "").splitlines() if line.strip()]
+
+
+def _duplicate_pr_warnings(raw_prs: list[dict]) -> list[CollisionWarning]:
+    issue_to_prs: dict[int, list[int]] = {}
+    for pr in raw_prs:
+        linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
+        if linked is None:
+            continue
+        issue_to_prs.setdefault(linked, []).append(int(pr["number"]))
+    warnings: list[CollisionWarning] = []
+    for issue_number, pr_numbers in sorted(issue_to_prs.items()):
+        if len(pr_numbers) > 1:
+            warnings.append(
+                CollisionWarning(
+                    kind="duplicate-pr",
+                    issue_number=issue_number,
+                    pr_numbers=tuple(sorted(pr_numbers)),
+                    message=(
+                        f"Issue #{issue_number} has {len(pr_numbers)} open PRs: "
+                        f"{', '.join(f'#{n}' for n in sorted(pr_numbers))} (#400)"
+                    ),
+                )
+            )
+    return warnings
+
+
+def _duplicate_branch_warnings(branch_names: list[str]) -> list[CollisionWarning]:
+    issue_to_branches: dict[int, list[str]] = {}
+    for name in branch_names:
+        match = _ISSUE_BRANCH_RE.search(name)
+        if not match:
+            continue
+        issue_num = int(match.group(1))
+        issue_to_branches.setdefault(issue_num, []).append(name)
+    warnings: list[CollisionWarning] = []
+    for issue_number, branches in sorted(issue_to_branches.items()):
+        if len(branches) > 1:
+            warnings.append(
+                CollisionWarning(
+                    kind="duplicate-branch",
+                    issue_number=issue_number,
+                    message=(
+                        f"Issue #{issue_number} has {len(branches)} local branches: "
+                        f"{', '.join(branches)}"
+                    ),
+                )
+            )
+    return warnings
+
+
+def _extract_reviewer_leases(
+    raw_prs: list[dict],
+    *,
+    fetch_comments: Callable[[int], list[dict]],
+) -> list[dict[str, Any]]:
+    leases: list[dict[str, Any]] = []
+    for pr in raw_prs:
+        pr_number = int(pr["number"])
+        for comment in fetch_comments(pr_number):
+            parsed = parse_reviewer_lease_comment(comment.get("body") or "")
+            if not parsed:
+                continue
+            leases.append(
+                {
+                    **parsed,
+                    "pr_number": parsed.get("pr_number") or pr_number,
+                    "comment_id": comment.get("id"),
+                    "author": (comment.get("user") or {}).get("login"),
+                    "created_at": comment.get("created_at"),
+                }
+            )
+    active_phases = {"claimed", "validating", "approved", "request-changes", "merging"}
+    return [lease for lease in leases if (lease.get("phase") or "") in active_phases]
+
+
+def load_lease_snapshot(
+    *,
+    project_id: str | None = None,
+    project_root: str | None = None,
+    issue_lock_path: str | None = None,
+    fetch_prs: Callable | None = None,
+    fetch_issues: Callable | None = None,
+    fetch_comments: Callable[[str, str, str, str, int], list[dict]] | None = None,
+) -> LeaseSnapshot:
+    registry = load_registry()
+    project: ProjectRecord | None = None
+    if project_id:
+        project = next((p for p in registry.projects if p.id == project_id), None)
+    else:
+        project = registry.projects[0] if registry.projects else None
+
+    root = project_root or _repo_root()
+    lock = read_issue_lock(issue_lock_path if issue_lock_path is not None else ISSUE_LOCK_FILE)
+
+    if project is None:
+        return LeaseSnapshot(
+            project_id=project_id or "",
+            repo_label="",
+            issue_lock=lock,
+            claim_inventory={"entries": [], "counts": {}},
+            reviewer_leases=(),
+            duplicate_prs=(),
+            duplicate_branches=(),
+            collision_history=_COLLISION_HISTORY,
+            fetch_error="project not found in registry",
+        )
+
+    host = _host_from_url(project.remote_host)
+    pr_fetch = fetch_prs or _fetch_prs
+    issue_fetch = fetch_issues or _fetch_issues
+    comment_fetch = fetch_comments or _fetch_comments
+    using_live = fetch_prs is None or fetch_issues is None
+    auth = get_auth_header(host) if using_live else "test-auth"
+
+    if using_live and not auth:
+        return LeaseSnapshot(
+            project_id=project.id,
+            repo_label=f"{project.gitea_owner}/{project.repo_name}",
+            issue_lock=lock,
+            claim_inventory={"entries": [], "counts": {}},
+            reviewer_leases=(),
+            duplicate_prs=(),
+            duplicate_branches=_duplicate_branch_warnings(_list_local_branch_names(root)),
+            collision_history=_COLLISION_HISTORY,
+            fetch_error=(
+                f"Gitea credentials unavailable for {host}; "
+                "remote lease artifacts cannot be loaded"
+            ),
+        )
+
+    try:
+        raw_prs, _ = pr_fetch(host, project.gitea_owner, project.repo_name, auth)
+        raw_issues, _ = issue_fetch(host, project.gitea_owner, project.repo_name, auth)
+    except Exception as exc:  # noqa: BLE001
+        return LeaseSnapshot(
+            project_id=project.id,
+            repo_label=f"{project.gitea_owner}/{project.repo_name}",
+            issue_lock=lock,
+            claim_inventory={"entries": [], "counts": {}},
+            reviewer_leases=(),
+            duplicate_prs=(),
+            duplicate_branches=_duplicate_branch_warnings(_list_local_branch_names(root)),
+            collision_history=_COLLISION_HISTORY,
+            fetch_error=f"Gitea fetch failed: {exc}",
+        )
+
+    comments_by_issue: dict[int, list[dict]] = {}
+    for issue in raw_issues:
+        if not any(lb.get("name") == "status:in-progress" for lb in issue.get("labels", [])):
+            continue
+        number = int(issue["number"])
+        comments_by_issue[number] = comment_fetch(
+            host, project.gitea_owner, project.repo_name, auth, issue_number=number
+        )
+
+    branch_names = _list_local_branch_names(root)
+    inventory = build_claim_inventory(
+        issues=raw_issues,
+        comments_by_issue=comments_by_issue,
+        open_prs=raw_prs,
+        branch_names=branch_names,
+    )
+
+    def _pr_comments(pr_number: int) -> list[dict]:
+        return comment_fetch(
+            host, project.gitea_owner, project.repo_name, auth, issue_number=pr_number
+        )
+
+    reviewer_leases = tuple(_extract_reviewer_leases(raw_prs, fetch_comments=_pr_comments))
+
+    return LeaseSnapshot(
+        project_id=project.id,
+        repo_label=f"{project.gitea_owner}/{project.repo_name}",
+        issue_lock=lock,
+        claim_inventory=inventory,
+        reviewer_leases=reviewer_leases,
+        duplicate_prs=tuple(_duplicate_pr_warnings(raw_prs)),
+        duplicate_branches=tuple(_duplicate_branch_warnings(branch_names)),
+        collision_history=_COLLISION_HISTORY,
+    )
+
+
+def snapshot_to_dict(snapshot: LeaseSnapshot) -> dict[str, Any]:
+    return {
+        "project_id": snapshot.project_id,
+        "repo_label": snapshot.repo_label,
+        "issue_lock": snapshot.issue_lock,
+        "claim_inventory": snapshot.claim_inventory,
+        "reviewer_leases": list(snapshot.reviewer_leases),
+        "duplicate_prs": [
+            {
+                "kind": w.kind,
+                "message": w.message,
+                "issue_number": w.issue_number,
+                "pr_numbers": list(w.pr_numbers),
+            }
+            for w in snapshot.duplicate_prs
+        ],
+        "duplicate_branches": [
+            {
+                "kind": w.kind,
+                "message": w.message,
+                "issue_number": w.issue_number,
+            }
+            for w in snapshot.duplicate_branches
+        ],
+        "collision_history": list(snapshot.collision_history),
+        "fetch_error": snapshot.fetch_error,
+    }
@@ -0,0 +1,130 @@
+"""HTML views for lease and collision visibility (#433)."""
+
+from __future__ import annotations
+
+import html
+import json
+
+from webui.layout import render_page
+from webui.lease_loader import LeaseSnapshot
+
+
+def _escape(text: str) -> str:
+    return html.escape(text, quote=True)
+
+
+def _warnings_block(snapshot: LeaseSnapshot) -> str:
+    warnings = list(snapshot.duplicate_prs) + list(snapshot.duplicate_branches)
+    if not warnings:
+        return "<p class='muted'>No duplicate PR/branch collisions detected in current inventory.</p>"
+    items = "".join(f"<li>{_escape(w.message)}</li>" for w in warnings)
+    return f"<ul class='collision-warnings'>{items}</ul>"
+
+
+def _lock_block(snapshot: LeaseSnapshot) -> str:
+    lock = snapshot.issue_lock
+    if not lock:
+        return "<p class='muted'>No active local issue lock file.</p>"
+    return (
+        "<pre class='prompt-text'>"
+        f"{_escape(json.dumps(lock, indent=2, sort_keys=True))}"
+        "</pre>"
+    )
+
+
+def _claims_table(snapshot: LeaseSnapshot) -> str:
+    entries = snapshot.claim_inventory.get("entries") or []
+    if not entries:
+        return "<p class='muted'>No in-progress issue claims in fetched inventory.</p>"
+    rows = []
+    for entry in entries:
+        rows.append(
+            "<tr>"
+            f"<td>#{entry.get('issue_number')}</td>"
+            f"<td><span class='badge badge-{_escape(str(entry.get('status')))}'>"
+            f"{_escape(str(entry.get('status')))}</span></td>"
+            f"<td>{_escape(str(entry.get('linked_open_pr') or '—'))}</td>"
+            f"<td>{entry.get('heartbeat_count', 0)}</td>"
+            f"<td>{_escape(', '.join(entry.get('matching_branches') or []) or '—')}</td>"
+            f"<td class='muted'>{_escape('; '.join(entry.get('reasons') or []))}</td>"
+            "</tr>"
+        )
+    return (
+        "<table class='registry leases'>"
+        "<thead><tr><th>Issue</th><th>Claim status</th><th>Open PR</th>"
+        "<th>Heartbeats</th><th>Branches</th><th>Notes</th></tr></thead>"
+        f"<tbody>{''.join(rows)}</tbody></table>"
+    )
+
+
+def _reviewer_leases_table(snapshot: LeaseSnapshot) -> str:
+    if not snapshot.reviewer_leases:
+        return (
+            "<p class='muted'>No active reviewer PR lease comments found "
+            "(marker <code>&lt;!-- mcp-review-lease:v1 --&gt;</code>; see #407).</p>"
+        )
+    rows = []
+    for lease in snapshot.reviewer_leases:
+        rows.append(
+            "<tr>"
+            f"<td>#{lease.get('pr_number')}</td>"
+            f"<td>{_escape(str(lease.get('reviewer_identity') or '—'))}</td>"
+            f"<td><code>{_escape(str(lease.get('profile') or '—'))}</code></td>"
+            f"<td>{_escape(str(lease.get('phase') or '—'))}</td>"
+            f"<td><code>{_escape(str(lease.get('expires_at') or '—'))}</code></td>"
+            "</tr>"
+        )
+    return (
+        "<table class='registry leases'>"
+        "<thead><tr><th>PR</th><th>Reviewer</th><th>Profile</th>"
+        "<th>Phase</th><th>Expires</th></tr></thead>"
+        f"<tbody>{''.join(rows)}</tbody></table>"
+    )
+
+
+def _history_links(snapshot: LeaseSnapshot) -> str:
+    items = "".join(
+        f"<li><a href=\"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/"
+        f"{item['number']}\">#{item['number']}</a> — {_escape(item['title'])}</li>"
+        for item in snapshot.collision_history
+    )
+    return f"<ul>{items}</ul>"
+
+
+LEASE_PAGE_STYLES = """
+<style>
+  .collision-warnings li { color: #f0c4c4; margin: 0.35rem 0; }
+  table.leases td:nth-child(6) { font-size: 0.85rem; }
+  .badge-active, .badge-awaiting_review { color: #8fd19e; border-color: #3d6b4a; }
+  .badge-stale, .badge-phantom, .badge-reclaimable { color: #f0a8a8; border-color: #7a3b3b; }
+</style>
+"""
+
+
+def render_leases_page(snapshot: LeaseSnapshot) -> str:
+    error_block = ""
+    if snapshot.fetch_error:
+        error_block = (
+            '<div class="stub"><p><strong>Partial load:</strong> '
+            f"{_escape(snapshot.fetch_error)}</p></div>"
+        )
+    body = (
+        "<h2>Leases &amp; collisions</h2>"
+        "<p>Read-only visibility for issue claims, reviewer PR leases, and "
+        "duplicate-work risks. Does not acquire or release leases.</p>"
+        f"{error_block}"
+        f"<p class='meta'>Project: <code>{_escape(snapshot.repo_label)}</code></p>"
+        "<h3>Collision warnings</h3>"
+        f"{_warnings_block(snapshot)}"
+        "<h3>Local issue lock</h3>"
+        f"{_lock_block(snapshot)}"
+        "<h3>In-progress issue claims (#268)</h3>"
+        f"{_claims_table(snapshot)}"
+        "<h3>Reviewer PR leases (#407)</h3>"
+        f"{_reviewer_leases_table(snapshot)}"
+        "<h3>Collision / lease history issues</h3>"
+        f"{_history_links(snapshot)}"
+        "<p><a href=\"/api/leases\">JSON API</a></p>"
+        f"{LEASE_PAGE_STYLES}"
+    )
+    return render_page(title="Leases", body_html=body)