feat: preserve and resume prepared review verdict drafts (#609)

Add non-terminal review draft save/resume tools so a fully validated
reviewer verdict can survive MCP disconnect or stale-runtime restart
without counting as a formal review.

- gitea_save_review_draft: durable draft with PR, head SHA, target SHA,
  worktree, validation evidence, intended verdict, and blocker reason
- gitea_resume_review_draft: live revalidation of head/target/#332/lease/
  identity/worktree before mark-ready or optional submit
- Drafts never substitute for formal approval without revalidation

Closes #609

docs/mcp-namespace-eof-recovery.md

@@ -1,118 +0,0 @@
-# Recovering from `client is closing: EOF` on a Gitea MCP namespace (#543)
-
-## Symptom
-
-A tool call through a Gitea MCP namespace — `gitea-author`, `gitea-reviewer`,
-`gitea-merger`, or the shared `gitea-tools` namespace — fails immediately with:
-
-```
-client is closing: EOF
-```
-
-Every subsequent call to that same namespace returns the same error, including
-cheap read tools such as `gitea_whoami` and `gitea_list_profiles`. Other MCP
-servers registered with the same client (for example `context7`) keep working,
-so this is **not** a global MCP-client outage.
-
-## Why this is not a code defect
-
-This failure is a **transport-level** condition in the IDE / MCP client manager,
-not a missing or broken tool:
-
-- The tool can be present and registered in the Python `FastMCP` tool manager.
-- Direct Python inspection of the server confirms the tool exists.
-- Running the server manually and sending JSON-RPC over stdio works fine
-  (offline spawn) — that path does **not** prove the IDE namespace is healthy.
-
-The client manager entered a closed state after the backing subprocess for that
-namespace terminated (or was killed) behind its back. Once closed, the client
-does **not** re-spawn the child on the next tool call — it just replays
-`client is closing: EOF`. The OS process may even still be alive if a parent
-language-server process is holding the stdio pipes open.
-
-This is the canonical "registered in FastMCP ≠ callable through the namespace"
-false-ready state. It is distinct from the **stale-runtime** family in #531 /
-#544, where the process is reachable but running behind `master`; that case is
-detected by the `ps`-based `_check_mcp_runtimes_diagnostics` in
-`gitea_mcp_server.py`. The EOF case is a dead/closed transport, not a stale one,
-so the `ps` check alone will not surface it.
-
-## Recovery path (canonical — client reconnect only)
-
-Do the steps in order. Stop as soon as a live **client-namespace** call succeeds.
-
-1. **Confirm the blast radius.** Call a cheap read tool on the failing namespace
-   (`gitea_whoami` or `gitea_list_profiles`). Then call the same tool on a
-   different MCP server (e.g. `context7`).
-   - Only the Gitea namespace fails → single-namespace transport close. Continue.
-   - Every server fails → restart the whole MCP client, not just one namespace.
-
-2. **Reconnect the namespace through the client, not the shell.** Use the IDE /
-   client MCP-reconnect action for that server entry (in Claude Code:
-   `/mcp` → reconnect the affected `gitea-*` server). Reconnecting forces the
-   client to spawn a fresh subprocess and re-open the pipe. This clears the
-   closed-client state that a bare `kill`/respawn from a terminal does **not**.
-
-3. **Do not "fix" it by importing the server or poking the process.** Reaching
-   for `python -c 'import gitea_mcp_server ...'`, raw JSON-RPC from a shell,
-   killing PIDs to force a respawn, or touching MCP config mtimes does **not**
-   restore the *client's* view of the namespace and violates the daemon-import
-   guard (#558, `docs/mcp-daemon-import-guard.md`). The only sanctioned repair
-   is a **client reconnect / relaunch**.
-
-4. **Verify through the same path the workflow will use.** After reconnect, call
-   the specific tool the blocked workflow needs — not just any tool — through
-   the target namespace. For a merge that means calling the merger-authorized
-   adoption/merge tool through `gitea-merger`. A green `gitea_whoami` on one
-   namespace does **not** prove another namespace or another tool is callable.
-   Record success with:
-
-   ```text
-   gitea_assess_mcp_namespace_health(..., probe_source="client_namespace")
-   ```
-
-5. **If reconnect does not clear it,** relaunch the client entirely, then repeat
-   step 4. If EOF persists after a full relaunch, the backing subprocess is
-   failing to start — inspect its stderr / launch config (command path, venv,
-   `*_MCP_CONFIG`, `*_MCP_PROFILE` env) rather than retrying the call. Still
-   do not use PID kill or config-touch as the primary recovery.
-
-## Diagnostics to capture when reporting EOF
-
-Include all of these so the failure is actionable and reproducible:
-
-- **Namespace name** that returned EOF (`gitea-author` / `gitea-reviewer` /
-  `gitea-merger` / `gitea-tools`).
-- **Tool** that was called and the **exact** error string.
-- **PID** of the backing process (if any) and whether it was still alive
-  (informational only — not a recovery action).
-- **Profile / env** for that namespace (execution profile, `*_MCP_PROFILE`,
-  worktree binding such as `GITEA_AUTHOR_WORKTREE`).
-- **Config path** the client launched the server from.
-- Result of the **cross-server control** call (did `context7` succeed?).
-
-## Offline spawn probe (non-authoritative)
-
-`test_mcp_conn.py` performs a full JSON-RPC handshake against a **fresh
-subprocess** (`initialize` → `initialized` → `tools/list` → `tools/call`) and
-classifies with `probe_source=offline_spawn`. That is useful for offline
-launch/registration debugging. It is **not** proof the IDE-managed namespace is
-healthy. See `docs/mcp-namespace-health.md`.
-
-## Do-not list during EOF recovery
-
-- Do **not** retry a blocked merge/adoption until the required tool is confirmed
-  callable through the merger-authorized **client** namespace (see #543).
-- Do **not** clean, reset, or rebind a **foreign** worktree to work around the
-  error.
-- Do **not** bypass the namespace with direct imports, raw API/curl, or
-  in-memory state restoration.
-- Do **not** kill MCP PIDs or touch config mtimes as a substitute for client
-  reconnect.
-
-## Related
-
-- #531 / #544 — stale-runtime detection (`ps`-based); sibling failure mode.
-- #558 / `docs/mcp-daemon-import-guard.md` — why shell imports are not a repair.
-- `docs/mcp-client-registration.md` — per-server registration contract.
-- `docs/mcp-namespace-health.md` — probe sources and mutation enforcement.

docs/mcp-namespace-health.md

@@ -1,88 +0,0 @@
-# MCP namespace health diagnostics (#543)
-
-Gitea MCP tools can be registered in the Python FastMCP server while the IDE's
-live MCP namespace is still unusable. The failure usually appears as
-`client is closing: EOF`, `transport closed`, or an empty response when calling
-a tool such as `gitea_whoami`.
-
-Do not treat static tool registration as proof that review or merge workflows
-can proceed. A reviewer or merger flow must have **client-namespace** evidence
-that the required tool is callable through the configured IDE MCP namespace.
-
-## Probe sources (do not confuse them)
-
-| Source | How obtained | Proves IDE namespace? |
-| --- | --- | --- |
-| `client_namespace` | Tool call through the IDE-managed MCP client | **Yes** |
-| `offline_spawn` | `test_mcp_conn.py` subprocess JSON-RPC handshake | **No** (offline only) |
-
-`gitea_assess_mcp_namespace_health` accepts `probe_source` and only sets
-`ide_namespace_proven=true` for `client_namespace` success. Offline spawn
-success never clears review/merge mutation gates.
-
-## Client-namespace health check (canonical)
-
-1. Through the IDE client, call a cheap tool on the target namespace
-   (`gitea_whoami` or `gitea_list_profiles`).
-2. Feed the live result into:
-
-```text
-gitea_assess_mcp_namespace_health(
-    namespace="gitea-merger",          # or reviewer / author / tools
-    registered_tools=[...],            # optional static list
-    probe_result={"success": true, "result": {...}},
-    probe_source="client_namespace",
-)
-```
-
-3. A healthy client-namespace assessment is recorded in the MCP session and
-   consulted by **live** `gitea_submit_pr_review` / `gitea_merge_pr` gates.
-4. If the probe fails with EOF, recover via **client reconnect only** — see
-   `docs/mcp-namespace-eof-recovery.md`. Do **not** kill PIDs or touch MCP
-   config mtimes as a recovery procedure.
-
-## Offline spawn probe (non-authoritative)
-
-```bash
-python3 test_mcp_conn.py --config ~/.gemini/config/mcp_config.json
-```
-
-This script spawns a **separate** server process from config, performs
-JSON-RPC `initialize` → `tools/list` → `tools/call`, and classifies the
-result with `probe_source=offline_spawn`. Use it for offline debugging of
-launch command / registration. It does **not** prove the IDE-managed
-namespace is healthy.
-
-By default the script checks:
-
-| Namespace | Required tool |
-| --- | --- |
-| `gitea-author` | `gitea_whoami` |
-| `gitea-reviewer` | `gitea_whoami` |
-| `gitea-merger` | `gitea_whoami` |
-| `gitea-tools` | `gitea_list_profiles` |
-
-## Recovery (canonical)
-
-When a namespace returns EOF, follow
-`docs/mcp-namespace-eof-recovery.md` in order:
-
-1. Confirm blast radius (Gitea namespace vs all MCP servers).
-2. **Reconnect the namespace through the client** (IDE reconnect / relaunch).
-3. Do not repair via shell imports, raw JSON-RPC, PID kills, or config mtime
-   touches — those do not restore the client's closed transport.
-4. Re-verify the **specific** required tool through the target namespace.
-5. Resume review/merge only after a successful `client_namespace` assessment.
-
-## Enforcement
-
-1. **State machine (read-only):** feed `blocks_merge_workflow` from a
-   `client_namespace` assessment into
-   `gitea_assess_review_merge_state_machine(live_namespace_broken=...)`.
-2. **Live mutations:** `gitea_submit_pr_review` and `gitea_merge_pr` call
-   `_live_namespace_health_gate` and fail closed when the session has a
-   recorded unhealthy or non-client probe for the required namespace
-   (`gitea-reviewer` for review, `gitea-merger` for merge).
-
-When blocked, repair the IDE namespace and re-record a healthy
-`client_namespace` assessment before retrying the mutation.

gitea_mcp_server.py

@@ -191,7 +191,6 @@ MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
 RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
 
 import namespace_workspace_binding as nwb  # noqa: E402
-import mcp_namespace_health  # noqa: E402
 
 
 def _preflight_in_test_mode() -> bool:
@@ -2825,35 +2824,6 @@ _TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
 # remote + profile identity with TTL.
 _REVIEW_DECISION_LOCK: dict | None = None
 
-# Session-scoped live MCP namespace health assessments (#543).
-# Keyed by namespace name. Only client_namespace entries authorize mutations.
-_LIVE_NAMESPACE_HEALTH: dict[str, dict] = {}
-
-
-def _record_live_namespace_health(assessment: dict | None) -> None:
-    """Store a namespace health assessment for mutation gates."""
-    if not isinstance(assessment, dict):
-        return
-    ns = str(assessment.get("namespace") or "").strip()
-    if not ns:
-        return
-    _LIVE_NAMESPACE_HEALTH[ns] = {
-        "namespace": ns,
-        "healthy": bool(assessment.get("healthy")),
-        "ide_namespace_proven": bool(assessment.get("ide_namespace_proven")),
-        "probe_source": assessment.get("probe_source"),
-        "blocks_merge_workflow": bool(assessment.get("blocks_merge_workflow")),
-        "error_type": assessment.get("error_type"),
-        "required_tool": assessment.get("required_tool"),
-    }
-
-
-def _live_namespace_health_gate(task: str) -> list[str]:
-    """Fail closed on recorded broken/non-client IDE namespace health (#543)."""
-    return mcp_namespace_health.mutation_gate_from_session(
-        task, _LIVE_NAMESPACE_HEALTH
-    )
-
 
 def _decision_lock_binding(lock: dict | None = None) -> dict:
     """Resolve key fields for durable decision-lock storage."""
@@ -3544,11 +3514,6 @@ def _evaluate_pr_review_submission(
         reasons.extend(workflow_blockers)
         reasons.extend(review_workflow_load.recovery_handoff_without_replay())
         return result
-    if live:
-        ns_gate = _live_namespace_health_gate("review_pr")
-        if ns_gate:
-            reasons.extend(ns_gate)
-            return result
 
     if action not in _REVIEW_ACTIONS:
         reasons.append(
@@ -4197,6 +4162,335 @@ def gitea_submit_pr_review(
     )
 
 
+@mcp.tool()
+def gitea_save_review_draft(
+    pr_number: int,
+    action: str,
+    body: str = "",
+    expected_head_sha: str | None = None,
+    validation_commands: str | None = None,
+    validation_results: str | None = None,
+    blocker_reason: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    worktree_path: str | None = None,
+) -> dict:
+    """Save a prepared review verdict as non-terminal draft evidence without submitting a formal review.
+
+    This permits resuming the review later (e.g. after a stale runtime restart or connection drop).
+    """
+    action = (action or "").strip().lower()
+    if action not in _REVIEW_ACTIONS:
+        return {
+            "success": False,
+            "reasons": [
+                f"unknown review action '{action}'; expected one of "
+                f"{sorted(_REVIEW_ACTIONS)}"
+            ],
+        }
+
+    # 1. Resolve remote context
+    try:
+        h, resolved_org, resolved_repo = _resolve(remote, host, org, repo)
+    except ValueError as exc:
+        return {"success": False, "reasons": [str(exc)]}
+
+    auth = _auth(h)
+
+    # 2. Fetch current PR details to get target base branch and base branch SHA
+    try:
+        pr_url = f"{repo_api_url(h, resolved_org, resolved_repo)}/pulls/{pr_number}"
+        pr = api_request("GET", pr_url, auth) or {}
+    except Exception as exc:
+        return {
+            "success": False,
+            "reasons": [f"failed to fetch PR #{pr_number} from Gitea: {str(exc)}"],
+        }
+
+    current_head = (pr.get("head") or {}).get("sha")
+    target_branch = (pr.get("base") or {}).get("ref")
+    target_branch_sha = (pr.get("base") or {}).get("sha")
+
+    if expected_head_sha and current_head and expected_head_sha != current_head:
+        return {
+            "success": False,
+            "reasons": [
+                f"expected_head_sha '{expected_head_sha}' does not match current PR head SHA '{current_head}'"
+            ],
+        }
+
+    # 3. Resolve worktree path
+    try:
+        resolved_worktree = _resolve_preflight_workspace_path(worktree_path)
+    except Exception as exc:
+        return {"success": False, "reasons": [f"failed to resolve worktree path: {str(exc)}"]}
+
+    # 4. Save payload to durable session state
+    profile = get_profile()
+    payload = {
+        "pr_number": pr_number,
+        "action": action,
+        "body": body,
+        "head_sha": expected_head_sha or current_head,
+        "target_branch": target_branch,
+        "target_branch_sha": target_branch_sha,
+        "worktree_path": resolved_worktree,
+        "validation_commands": validation_commands,
+        "validation_results": validation_results,
+        "blocker_reason": blocker_reason,
+        "saved_by_identity": _authenticated_username(h) or "",
+        "saved_by_profile": profile.get("profile_name"),
+        "remote": remote,
+        "org": resolved_org,
+        "repo": resolved_repo,
+    }
+
+    try:
+        mcp_session_state.save_state(
+            kind=mcp_session_state.KIND_REVIEW_DRAFT,
+            payload=payload,
+            remote=remote,
+            org=resolved_org,
+            repo=resolved_repo,
+            profile_identity=profile.get("profile_name"),
+        )
+    except Exception as exc:
+        return {"success": False, "reasons": [f"failed to save draft state: {str(exc)}"]}
+
+    return {
+        "success": True,
+        "pr_number": pr_number,
+        "action": action,
+        "head_sha": payload["head_sha"],
+        "target_branch": target_branch,
+        "target_branch_sha": target_branch_sha,
+        "worktree_path": resolved_worktree,
+    }
+
+
+@mcp.tool()
+def gitea_resume_review_draft(
+    pr_number: int,
+    submit: bool = False,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    worktree_path: str | None = None,
+) -> dict:
+    """Resume a previously saved review draft for the given PR number.
+
+    Performs live-state checks (PR head SHA, target branch SHA, terminal lock, active leases,
+    runtime/master parity, reviewer identity, worktree binding, validation freshness).
+    Refuses to submit or mark ready if any state has changed unsafely.
+    """
+    profile = get_profile()
+    active_profile = profile.get("profile_name")
+
+    # 1. Load draft state
+    try:
+        draft = mcp_session_state.load_state(
+            kind=mcp_session_state.KIND_REVIEW_DRAFT,
+            remote=remote,
+            org=org,
+            repo=repo,
+            profile_identity=active_profile,
+        )
+    except Exception as exc:
+        return {"success": False, "reasons": [f"failed to load draft state: {str(exc)}"]}
+
+    if not draft:
+        return {
+            "success": False,
+            "reasons": [f"no review draft found for PR #{pr_number} under profile '{active_profile}'"],
+        }
+
+    if draft.get("pr_number") != pr_number:
+        return {
+            "success": False,
+            "reasons": [
+                f"loaded draft is for PR #{draft.get('pr_number')}, but requested PR #{pr_number}"
+            ],
+        }
+
+    # 2. Resolve remote context
+    try:
+        h, resolved_org, resolved_repo = _resolve(remote, host, org, repo)
+    except ValueError as exc:
+        return {"success": False, "reasons": [str(exc)]}
+
+    auth = _auth(h)
+
+    # 3. Fetch live PR state
+    try:
+        pr_url = f"{repo_api_url(h, resolved_org, resolved_repo)}/pulls/{pr_number}"
+        pr = api_request("GET", pr_url, auth) or {}
+    except Exception as exc:
+        return {
+            "success": False,
+            "reasons": [f"failed to fetch PR #{pr_number} from Gitea: {str(exc)}"],
+        }
+
+    # 4. Perform live-state checks
+    reasons = []
+
+    # PR must be open
+    if pr.get("state") != "open":
+        reasons.append(f"PR #{pr_number} is not open (state: '{pr.get('state')}')")
+
+    # Live head SHA must match saved head_sha
+    live_head = (pr.get("head") or {}).get("sha")
+    saved_head = draft.get("head_sha")
+    if live_head != saved_head:
+        reasons.append(
+            f"PR head SHA changed (live={live_head!r}, saved={saved_head!r})"
+        )
+
+    # Live target branch SHA must match saved target_branch_sha
+    live_target_sha = (pr.get("base") or {}).get("sha")
+    saved_target_sha = draft.get("target_branch_sha")
+    if live_target_sha != saved_target_sha:
+        reasons.append(
+            f"target branch SHA changed (live={live_target_sha!r}, saved={saved_target_sha!r})"
+        )
+
+    # Check #332 terminal lock
+    hard_stop = terminal_review_hard_stop_reasons(pr_number, "resume")
+    if hard_stop:
+        reasons.extend(hard_stop)
+
+    # Check active reviewer lease (must be claimed by the current session)
+    try:
+        comments = _fetch_pr_comments(pr_number, remote=remote, host=host, org=resolved_org, repo=resolved_repo)
+    except Exception as exc:
+        reasons.append(f"failed to fetch PR comments: {str(exc)}")
+        comments = []
+
+    active_lease = reviewer_pr_lease.find_active_reviewer_lease(comments, pr_number=pr_number)
+    if not active_lease:
+        reasons.append(f"no active reviewer lease found on PR #{pr_number}")
+    else:
+        # Check that lease owner is our session_id
+        session = reviewer_pr_lease.get_session_lease()
+        session_id = (session or {}).get("session_id")
+        owner_session_id = active_lease.get("session_id")
+        if owner_session_id != session_id:
+            reasons.append(
+                f"active reviewer lease belongs to session_id={owner_session_id!r}, "
+                f"but current session has session_id={session_id!r}"
+            )
+
+    # Check runtime/master parity
+    if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
+        config = gitea_config.load_config()
+        matching_profiles = []
+        if 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
+                # Check for "review_pr" or similar capability
+                ok, _ = gitea_config.check_operation("gitea.pr.review", p_allowed_n, p_forbidden_n)
+                if ok:
+                    matching_profiles.append(p_name)
+        runtime_reasons = _check_mcp_runtimes_diagnostics("review_pr", matching_profiles)
+        if runtime_reasons:
+            reasons.extend(runtime_reasons)
+
+    # Check reviewer identity
+    identity = _authenticated_username(h)
+    if identity != draft.get("saved_by_identity"):
+        reasons.append(
+            f"reviewer identity mismatch (active={identity!r}, saved={draft.get('saved_by_identity')!r})"
+        )
+
+    # Check worktree binding
+    try:
+        resolved_worktree = _resolve_preflight_workspace_path(worktree_path)
+    except Exception as exc:
+        reasons.append(f"failed to resolve current worktree: {str(exc)}")
+        resolved_worktree = None
+    if resolved_worktree != draft.get("worktree_path"):
+        reasons.append(
+            f"worktree binding mismatch (current={resolved_worktree!r}, saved={draft.get('worktree_path')!r})"
+        )
+
+    if reasons:
+        return {"success": False, "reasons": reasons}
+
+    # 5. All checks pass! Mark decision lock as ready.
+    action = draft.get("action")
+    body = draft.get("body")
+    saved_head = draft.get("head_sha")
+
+    lock = _load_review_decision_lock() or {}
+    lock["final_review_decision_ready"] = True
+    lock["ready_pr_number"] = pr_number
+    lock["ready_action"] = action
+    lock["ready_expected_head_sha"] = saved_head
+    lock["ready_remote"] = remote
+    lock["ready_org"] = resolved_org
+    lock["ready_repo"] = resolved_repo
+    _save_review_decision_lock(lock)
+
+    submitted = False
+    submission_res = None
+    if submit:
+        # Submit the review!
+        submission_res = gitea_submit_pr_review(
+            pr_number=pr_number,
+            action=action,
+            body=body,
+            expected_head_sha=saved_head,
+            remote=remote,
+            host=host,
+            org=resolved_org,
+            repo=resolved_repo,
+            final_review_decision_ready=True,
+            worktree_path=resolved_worktree,
+        )
+        if submission_res.get("performed") or submission_res.get("success"):
+            submitted = True
+            # Delete the draft payload on success
+            mcp_session_state.save_state(
+                kind=mcp_session_state.KIND_REVIEW_DRAFT,
+                payload=None,
+                remote=remote,
+                org=resolved_org,
+                repo=resolved_repo,
+                profile_identity=active_profile,
+            )
+        else:
+            # If submit fails, return the submit failure reasons
+            return {
+                "success": False,
+                "reasons": submission_res.get("reasons") or ["submission failed"],
+                "submission_result": submission_res,
+            }
+
+    return {
+        "success": True,
+        "pr_number": pr_number,
+        "action": action,
+        "marked_ready": True,
+        "submitted": submitted,
+        "submission_result": submission_res,
+    }
+
+
 @mcp.tool()
 def gitea_edit_pr(
     pr_number: int,
@@ -4650,12 +4944,6 @@ def gitea_merge_pr(
         reasons.extend(review_workflow_load.recovery_handoff_without_replay())
         return result
 
-    # Gate 0b — recorded live MCP namespace health must not be broken (#543).
-    ns_gate = _live_namespace_health_gate("merge_pr")
-    if ns_gate:
-        reasons.extend(ns_gate)
-        return result
-
     # Gate 1 — valid merge method (no API call on a bad method).
     if do not in _MERGE_METHODS:
         reasons.append(
@@ -8665,45 +8953,40 @@ def gitea_assess_review_merge_state_machine(
     pre_merge_gates: dict[str, bool] | None = None,
     infra_stop: bool = False,
     capability_blocked: bool = False,
-    live_namespace_broken: bool = False,
     recovery_handoff_text: str | None = None,
     final_report_text: str | None = None,
 ) -> dict:
-    """Read-only: assess enforced PR review/merge workflow state (#290).
-
-    ``live_namespace_broken`` fails review/merge closed when the live MCP
-    namespace call path is unusable even though the tool is registered in
-    FastMCP (#543 AC5). Supply the ``blocks_merge_workflow`` verdict from
-    ``gitea_assess_mcp_namespace_health``.
-    """
+    """Read-only: assess enforced PR review/merge workflow state (#290)."""
     completion = state_completion or {}
-    blocker_kwargs = {
-        "infra_stop": infra_stop,
-        "capability_blocked": capability_blocked,
-        "live_namespace_broken": live_namespace_broken,
-    }
-    blockers = review_merge_state_machine.assess_workflow_blockers(**blocker_kwargs)
+    blockers = review_merge_state_machine.assess_workflow_blockers(
+        infra_stop=infra_stop,
+        capability_blocked=capability_blocked,
+    )
     result = {
         "workflow": review_merge_state_machine.workflow_status(
             completion,
-            **blocker_kwargs,
+            infra_stop=infra_stop,
+            capability_blocked=capability_blocked,
         ),
         "blockers": blockers,
         "approve": review_merge_state_machine.can_approve(
             completion,
-            **blocker_kwargs,
+            infra_stop=infra_stop,
+            capability_blocked=capability_blocked,
         ),
         "merge": review_merge_state_machine.can_merge(
             completion,
             pre_merge_gates=pre_merge_gates,
-            **blocker_kwargs,
+            infra_stop=infra_stop,
+            capability_blocked=capability_blocked,
         ),
     }
     if target_state:
         result["advancement"] = review_merge_state_machine.assess_state_advancement(
             completion,
             target_state=target_state,
-            **blocker_kwargs,
+            infra_stop=infra_stop,
+            capability_blocked=capability_blocked,
         )
     if recovery_handoff_text is not None:
         result["recovery_handoff"] = (
@@ -9233,46 +9516,6 @@ def gitea_list_profiles() -> dict:
     return {"profiles": profiles_out}
 
 
-@mcp.tool()
-def gitea_assess_mcp_namespace_health(
-    namespace: str,
-    required_tool: str | None = None,
-    registered_tools: list[str] | None = None,
-    probe_result: dict | None = None,
-    process: dict | None = None,
-    config_path: str | None = None,
-    profile: str | None = None,
-    configured: bool = True,
-    probe_source: str | None = None,
-) -> dict:
-    """Classify MCP namespace health for required Gitea tools (#543).
-
-    Static FastMCP registration is not enough to prove a namespace works: IDE
-    clients can keep a registered tool list while live calls fail with
-    ``client is closing: EOF``. Pass live IDE invocation evidence with
-    ``probe_source='client_namespace'``. Offline subprocess probes
-    (``test_mcp_conn.py``) must use ``probe_source='offline_spawn'`` and never
-    count as IDE proof.
-
-    Assessments are recorded in the session so live
-    ``gitea_submit_pr_review`` / ``gitea_merge_pr`` can fail closed when a
-    client-namespace probe reported unhealthy.
-    """
-    result = mcp_namespace_health.classify_namespace_probe(
-        namespace,
-        required_tool=required_tool,
-        registered_tools=registered_tools,
-        probe_result=probe_result,
-        process=process,
-        config_path=config_path,
-        profile=profile,
-        configured=configured,
-        probe_source=probe_source,
-    )
-    _record_live_namespace_health(result)
-    return result
-
-
 @mcp.tool()
 def gitea_activate_profile(
     profile_name: str,

mcp_namespace_health.py

@@ -1,306 +0,0 @@
-"""Assess live MCP namespace health without trusting static registration.
-
-The IDE/client namespace can fail with EOF even when this Python process still
-registers the Gitea tools with FastMCP.  These helpers keep that distinction
-explicit so reviewer/merger flows can fail closed on the live path.
-
-Probe sources
--------------
-* ``client_namespace`` — evidence from the IDE-managed MCP client path (the
-  only source that can prove the workflow namespace is healthy).
-* ``offline_spawn`` — a separate ``subprocess.Popen`` JSON-RPC handshake
-  (e.g. ``test_mcp_conn.py``). Useful offline, but **never** proves the
-  IDE-managed namespace is callable.
-* ``unknown`` — legacy/unspecified; treated as not IDE-proven.
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-
-REQUIRED_NAMESPACE_TOOLS = {
-    "gitea-author": "gitea_whoami",
-    "gitea-reviewer": "gitea_whoami",
-    "gitea-merger": "gitea_whoami",
-    "gitea-tools": "gitea_list_profiles",
-}
-
-DEFAULT_NAMESPACES = tuple(REQUIRED_NAMESPACE_TOOLS)
-
-# Namespaces that must be healthy for a given mutation task.
-TASK_REQUIRED_NAMESPACES = {
-    "review_pr": "gitea-reviewer",
-    "submit_review": "gitea-reviewer",
-    "merge_pr": "gitea-merger",
-}
-
-PROBE_SOURCE_CLIENT = "client_namespace"
-PROBE_SOURCE_OFFLINE = "offline_spawn"
-PROBE_SOURCE_UNKNOWN = "unknown"
-VALID_PROBE_SOURCES = frozenset(
-    {PROBE_SOURCE_CLIENT, PROBE_SOURCE_OFFLINE, PROBE_SOURCE_UNKNOWN}
-)
-
-EOF_PATTERNS = (
-    "client is closing: eof",
-    "transport closed",
-    "connection closed",
-    "broken pipe",
-    "end of file",
-    "eof",
-)
-
-SAFE_ENV_KEYS = (
-    "GITEA_MCP_PROFILE",
-    "GITEA_PROFILE_NAME",
-    "GITEA_SERVICE",
-    "GITEA_EXECUTION_ROLE",
-    "GITEA_MCP_CONFIG",
-)
-
-
-def _as_list(value: Any) -> list[str] | None:
-    if value is None:
-        return None
-    if isinstance(value, (list, tuple, set)):
-        return [str(v) for v in value]
-    return [str(value)]
-
-
-def _contains_eof(text: str | None) -> bool:
-    lowered = (text or "").lower()
-    return any(pattern in lowered for pattern in EOF_PATTERNS)
-
-
-def _normalize_probe_source(probe_source: str | None) -> str:
-    raw = (probe_source or PROBE_SOURCE_UNKNOWN).strip().lower()
-    if raw in VALID_PROBE_SOURCES:
-        return raw
-    return PROBE_SOURCE_UNKNOWN
-
-
-def _safe_env_summary(process: dict[str, Any] | None) -> dict[str, str]:
-    if not process:
-        return {}
-    env = process.get("env") or process.get("environment") or {}
-    if not isinstance(env, dict):
-        return {}
-    return {
-        key: str(env[key])
-        for key in SAFE_ENV_KEYS
-        if key in env and env[key] not in (None, "")
-    }
-
-
-def classify_namespace_probe(
-    namespace: str,
-    *,
-    required_tool: str | None = None,
-    registered_tools: list[str] | tuple[str, ...] | set[str] | None = None,
-    probe_result: dict[str, Any] | None = None,
-    process: dict[str, Any] | None = None,
-    config_path: str | None = None,
-    profile: str | None = None,
-    configured: bool = True,
-    probe_source: str | None = None,
-) -> dict[str, Any]:
-    """Classify whether a required tool is callable through a live namespace.
-
-    ``registered_tools`` is static/server-side evidence. ``probe_result`` is
-    live invocation evidence. Only ``probe_source=client_namespace`` proves the
-    IDE-managed path; ``offline_spawn`` is an offline subprocess check only.
-    """
-    ns = (namespace or "").strip()
-    tool = required_tool or REQUIRED_NAMESPACE_TOOLS.get(ns) or "gitea_whoami"
-    source = _normalize_probe_source(probe_source)
-    registered_list = _as_list(registered_tools)
-    registered = None if registered_list is None else tool in registered_list
-
-    probe = probe_result or {}
-    probe_success = bool(probe.get("success"))
-    error_message = str(
-        probe.get("error")
-        or probe.get("message")
-        or probe.get("stderr")
-        or probe.get("exception")
-        or ""
-    )
-    error_type = str(probe.get("error_type") or "").strip()
-    if not error_type and error_message:
-        if _contains_eof(error_message):
-            error_type = "namespace_eof"
-        elif "timeout" in error_message.lower():
-            error_type = "namespace_timeout"
-        else:
-            error_type = "namespace_call_failed"
-
-    if not configured:
-        error_type = "namespace_not_configured"
-    elif registered is False:
-        error_type = "tool_missing"
-    elif not probe_result:
-        error_type = "live_probe_missing"
-    elif not probe_success and not error_type:
-        error_type = "namespace_call_failed"
-
-    callable_live = bool(configured and probe_result and probe_success)
-    # Probe-path health (spawn or client). IDE-proven only for client path.
-    healthy = bool(configured and registered is not False and callable_live)
-    ide_namespace_proven = bool(healthy and source == PROBE_SOURCE_CLIENT)
-    process_pid = process.get("pid") if isinstance(process, dict) else None
-    profile_name = profile or (
-        process.get("profile") if isinstance(process, dict) else None
-    )
-    env_summary = _safe_env_summary(process)
-
-    reasons: list[str] = []
-    if not configured:
-        reasons.append(f"MCP namespace '{ns}' is not configured.")
-    if registered is False:
-        reasons.append(
-            f"Required tool '{tool}' is not registered in namespace '{ns}'."
-        )
-    if error_type == "live_probe_missing":
-        reasons.append(
-            f"No live client invocation proof was supplied for '{ns}.{tool}'."
-        )
-    elif error_type == "namespace_eof":
-        reasons.append(
-            f"Live MCP namespace '{ns}' returned EOF while invoking '{tool}'."
-        )
-    elif error_type == "namespace_timeout":
-        reasons.append(
-            f"Live MCP namespace '{ns}' timed out while invoking '{tool}'."
-        )
-    elif error_type == "namespace_call_failed":
-        reasons.append(
-            f"Live MCP namespace '{ns}' failed while invoking '{tool}'."
-        )
-    if source == PROBE_SOURCE_OFFLINE:
-        reasons.append(
-            "Probe source is offline_spawn (subprocess JSON-RPC); this does "
-            "not prove the IDE-managed MCP namespace is healthy."
-        )
-    elif source == PROBE_SOURCE_UNKNOWN and probe_result:
-        reasons.append(
-            "Probe source unspecified; treat as not IDE-namespace proof unless "
-            "re-supplied with probe_source='client_namespace'."
-        )
-
-    remediation = []
-    if not healthy or not ide_namespace_proven:
-        remediation.append(
-            "Reconnect the IDE MCP client namespace (client reconnect / "
-            f"relaunch), then invoke '{tool}' through namespace '{ns}' and "
-            "record the result with probe_source='client_namespace'."
-        )
-        remediation.append(
-            "Do not treat offline subprocess probes (test_mcp_conn.py) or "
-            "shell kill/PID respawn as proof the IDE namespace is repaired."
-        )
-        if process_pid and source != PROBE_SOURCE_OFFLINE:
-            remediation.append(
-                f"Diagnostics may include PID {process_pid}; process details "
-                "are informational only — recovery is client-layer reconnect."
-            )
-    else:
-        remediation.append(
-            f"IDE-managed namespace '{ns}' can invoke '{tool}' "
-            f"(probe_source={source})."
-        )
-
-    # Client-namespace broken health blocks review/merge. Offline probes never
-    # authorize mutations and only block when they report unhealthy (still
-    # fail-closed for known bad spawn evidence).
-    blocks = False
-    if source == PROBE_SOURCE_CLIENT:
-        blocks = namespace_health_blocks_task("merge_pr", healthy)
-    elif source == PROBE_SOURCE_OFFLINE:
-        # Offline never proves IDE health; never unblock. Unhealthy offline
-        # still surfaces as a soft diagnostic, not a mutation-ledger block.
-        blocks = False
-    else:
-        # Unknown source: only block when evidence is unhealthy (fail closed
-        # on bad data without treating success as IDE proof).
-        blocks = namespace_health_blocks_task("merge_pr", healthy)
-
-    return {
-        "success": healthy,
-        "healthy": healthy,
-        "namespace": ns,
-        "required_tool": tool,
-        "configured": configured,
-        "registered_tools_checked": registered_list is not None,
-        "required_tool_registered": registered,
-        "required_tool_callable": callable_live,
-        "probe_source": source,
-        "ide_namespace_proven": ide_namespace_proven,
-        "error_type": None if healthy else error_type,
-        "error_message": error_message or None,
-        "reasons": reasons,
-        "remediation": remediation,
-        "diagnostics": {
-            "namespace": ns,
-            "required_tool": tool,
-            "process_pid": process_pid,
-            "profile": profile_name,
-            "env": env_summary,
-            "config_path": config_path,
-            "probe_source": source,
-        },
-        "blocks_merge_workflow": blocks,
-    }
-
-
-def namespace_health_blocks_task(task: str, healthy: bool) -> bool:
-    """Return whether a broken namespace must block a workflow task."""
-    if healthy:
-        return False
-    return (task or "").strip() in {"merge_pr", "review_pr", "submit_review"}
-
-
-def required_namespace_for_task(task: str) -> str | None:
-    """Map a mutation task to the MCP namespace that must be healthy."""
-    return TASK_REQUIRED_NAMESPACES.get((task or "").strip())
-
-
-def mutation_gate_from_session(
-    task: str,
-    session_health: dict[str, dict[str, Any]] | None,
-) -> list[str]:
-    """Fail-closed gate using recorded client-namespace health assessments.
-
-    * Missing session entry → no gate (caller has not assessed yet).
-    * Client-namespace unhealthy / not IDE-proven → block mutation.
-    * Offline-only session entries never authorize mutations.
-    """
-    ns = required_namespace_for_task(task)
-    if not ns:
-        return []
-    store = session_health or {}
-    entry = store.get(ns)
-    if not entry:
-        return []
-    source = _normalize_probe_source(entry.get("probe_source"))
-    if source != PROBE_SOURCE_CLIENT:
-        return [
-            f"recorded namespace health for '{ns}' is probe_source={source}, "
-            "not client_namespace; re-probe through the IDE-managed path "
-            f"before {(task or 'mutation')}"
-        ]
-    if entry.get("ide_namespace_proven") and entry.get("healthy"):
-        return []
-    if entry.get("blocks_merge_workflow") or not entry.get("healthy"):
-        detail = entry.get("error_type") or "unhealthy"
-        return [
-            f"live MCP namespace '{ns}' is recorded {detail} "
-            f"(probe_source={source}); repair the IDE namespace before "
-            f"{(task or 'mutation')} (fail closed, #543)"
-        ]
-    if not entry.get("ide_namespace_proven"):
-        return [
-            f"live MCP namespace '{ns}' is not IDE-proven; supply a "
-            "client_namespace probe before mutation (fail closed, #543)"
-        ]
-    return []

mcp_session_state.py

@@ -30,6 +30,8 @@ DEFAULT_TTL_HOURS = 4.0
 
 KIND_WORKFLOW_LOAD = "review_workflow_load"
 KIND_DECISION_LOCK = "review_decision_lock"
+KIND_REVIEW_DRAFT = "review_draft"
+
 
 _SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
 SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"

review_merge_state_machine.py

@@ -93,16 +93,8 @@ def assess_workflow_blockers(
     capability_blocked: bool = False,
     mcp_reconnect_failed: bool = False,
     stale_capability_state: bool = False,
-    live_namespace_broken: bool = False,
 ) -> dict[str, Any]:
-    """Return hard blockers that forbid all PR queue work (#290 AC3).
-
-    ``live_namespace_broken`` fails the merge/review path closed when the live
-    MCP namespace call path is unusable (e.g. ``client is closing: EOF``) even
-    though the tool is registered in FastMCP (#543 AC5). Feed it the
-    ``blocks_merge_workflow`` verdict from
-    ``mcp_namespace_health.classify_namespace_probe``.
-    """
+    """Return hard blockers that forbid all PR queue work (#290 AC3)."""
     reasons: list[str] = []
     if infra_stop:
         reasons.append("infra_stop is active; PR selection/review/merge is forbidden")
@@ -112,12 +104,6 @@ def assess_workflow_blockers(
         reasons.append("MCP reconnect failed; stale session state cannot be reused")
     if stale_capability_state:
         reasons.append("stale MCP capability state detected after reconnect failure")
-    if live_namespace_broken:
-        reasons.append(
-            "live MCP namespace call path is broken (registered in FastMCP but "
-            "not callable through the namespace); repair the namespace before "
-            "review/merge"
-        )
     return {
         "block": bool(reasons),
         "reasons": reasons,
@@ -132,19 +118,16 @@ def assess_state_advancement(
     state_completion: dict[str, bool] | None,
     *,
     target_state: str,
-    **blocker_kwargs,
+    infra_stop: bool = False,
+    capability_blocked: bool = False,
 ) -> dict[str, Any]:
-    """Fail closed when *target_state* is requested before upstream gates pass.
-
-    Forwards every blocker flag (``infra_stop``, ``capability_blocked``,
-    ``mcp_reconnect_failed``, ``stale_capability_state``,
-    ``live_namespace_broken``) to :func:`assess_workflow_blockers` so
-    ``can_approve``/``can_merge`` honor the full blocker set, not just
-    infra_stop/capability_blocked (#543 AC5).
-    """
+    """Fail closed when *target_state* is requested before upstream gates pass."""
     completion = dict(state_completion or {})
     target = _clean(target_state).upper()
-    blockers = assess_workflow_blockers(**blocker_kwargs)
+    blockers = assess_workflow_blockers(
+        infra_stop=infra_stop,
+        capability_blocked=capability_blocked,
+    )
     reasons = list(blockers["reasons"])
 
     try:

test_mcp_conn.py

@@ -1,45 +1,20 @@
 #!/usr/bin/env python3
-"""Offline-only MCP namespace spawn probe (NOT IDE-namespace proof).
+"""Live health-check script to verify Gitea MCP namespace connections.
 
-Spawns a *separate* MCP server process from config via subprocess.Popen,
-performs the JSON-RPC handshake, verifies the required tool is registered,
-and invokes that tool. Results are classified with
-``probe_source=offline_spawn``.
-
-This path is useful for offline launch/registration debugging. It does
-**not** prove the IDE-managed MCP client namespace is healthy (#543). For
-workflow gates, pass live IDE call evidence with
-``probe_source=client_namespace`` to ``gitea_assess_mcp_namespace_health``.
+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 argparse
 import json
 import os
 import subprocess
 import sys
 
-from mcp_namespace_health import REQUIRED_NAMESPACE_TOOLS, classify_namespace_probe
-
-
-def _read_json_line(proc):
-    line = proc.stdout.readline()
-    if not line:
-        stderr_content = proc.stderr.read()
-        return None, stderr_content
-    return json.loads(line), None
-
-
-def _write_message(proc, payload):
-    proc.stdin.write(json.dumps(payload) + "\n")
-    proc.stdin.flush()
-
-
-def run_connection_test(name, config, *, required_tool=None, config_path=None):
+def run_connection_test(name, config):
     print(f"Testing MCP connection for '{name}'...")
     command = config.get("command")
     args = config.get("args", [])
     env = config.get("env", {})
-    tool_name = required_tool or REQUIRED_NAMESPACE_TOOLS.get(name) or "gitea_whoami"
     
     # Merge current environment
     run_env = os.environ.copy()
@@ -56,17 +31,8 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
             bufsize=1,
             env=run_env
         )
-    except Exception as exc:
-        print(f"  [FAIL] Failed to spawn process: {exc}")
-        assessment = classify_namespace_probe(
-            name,
-            required_tool=tool_name,
-            probe_result={"success": False, "error": str(exc)},
-            process={"profile": env.get("GITEA_MCP_PROFILE"), "env": env},
-            config_path=config_path,
-                probe_source="offline_spawn",
-        )
-        print(f"  diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
+    except Exception as e:
+        print(f"  [FAIL] Failed to spawn process: {e}")
         return False
 
     # Send initialize request
@@ -82,36 +48,26 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
     }
     
     try:
-        _write_message(proc, init_req)
+        proc.stdin.write(json.dumps(init_req) + "\n")
+        proc.stdin.flush()
         
         # Read response
-        res, stderr_content = _read_json_line(proc)
-        if res is None:
+        line = proc.stdout.readline()
+        if not line:
+            stderr_content = proc.stderr.read()
             print(f"  [FAIL] Received EOF from process. Stderr:\n{stderr_content}")
-            assessment = classify_namespace_probe(
-                name,
-                required_tool=tool_name,
-                probe_result={"success": False, "error": stderr_content or "EOF"},
-                process={
-                    "pid": proc.pid,
-                    "profile": env.get("GITEA_MCP_PROFILE"),
-                    "env": env,
-                },
-                config_path=config_path,
-                probe_source="offline_spawn",
-            )
-            print(f"  remediation: {' '.join(assessment['remediation'])}")
             proc.terminate()
             return False
             
-        print(f"  [OK] Received initialize response: {str(res)[:150]}...")
+        print(f"  [OK] Received initialize response: {line.strip()[:150]}...")
         
         # Send initialized notification
         init_notif = {
             "jsonrpc": "2.0",
             "method": "notifications/initialized"
         }
-        _write_message(proc, init_notif)
+        proc.stdin.write(json.dumps(init_notif) + "\n")
+        proc.stdin.flush()
         
         # Send tools/list request
         list_req = {
@@ -120,27 +76,16 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
             "params": {},
             "id": 2
         }
-        _write_message(proc, list_req)
+        proc.stdin.write(json.dumps(list_req) + "\n")
+        proc.stdin.flush()
         
-        res, stderr_content = _read_json_line(proc)
-        if res is None:
+        line = proc.stdout.readline()
+        if not line:
             print("  [FAIL] Received EOF on tools/list request.")
-            assessment = classify_namespace_probe(
-                name,
-                required_tool=tool_name,
-                probe_result={"success": False, "error": stderr_content or "EOF"},
-                process={
-                    "pid": proc.pid,
-                    "profile": env.get("GITEA_MCP_PROFILE"),
-                    "env": env,
-                },
-                config_path=config_path,
-                probe_source="offline_spawn",
-            )
-            print(f"  remediation: {' '.join(assessment['remediation'])}")
             proc.terminate()
             return False
             
+        res = json.loads(line)
         if "error" in res:
             print(f"  [FAIL] Server returned error: {res['error']}")
             proc.terminate()
@@ -149,115 +94,16 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
         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]}...")
-        if tool_name not in tool_names:
-            assessment = classify_namespace_probe(
-                name,
-                required_tool=tool_name,
-                registered_tools=tool_names,
-                probe_result={"success": False, "error": "required tool missing"},
-                process={
-                    "pid": proc.pid,
-                    "profile": env.get("GITEA_MCP_PROFILE"),
-                    "env": env,
-                },
-                config_path=config_path,
-                probe_source="offline_spawn",
-            )
-            print(f"  [FAIL] Required tool '{tool_name}' is not registered.")
-            print(f"  remediation: {' '.join(assessment['remediation'])}")
-            proc.terminate()
-            return False
-
-        call_req = {
-            "jsonrpc": "2.0",
-            "method": "tools/call",
-            "params": {"name": tool_name, "arguments": {}},
-            "id": 3,
-        }
-        _write_message(proc, call_req)
-
-        call_res, stderr_content = _read_json_line(proc)
-        if call_res is None:
-            assessment = classify_namespace_probe(
-                name,
-                required_tool=tool_name,
-                registered_tools=tool_names,
-                probe_result={"success": False, "error": stderr_content or "EOF"},
-                process={
-                    "pid": proc.pid,
-                    "profile": env.get("GITEA_MCP_PROFILE"),
-                    "env": env,
-                },
-                config_path=config_path,
-                probe_source="offline_spawn",
-            )
-            print(f"  [FAIL] Received EOF on {tool_name} invocation.")
-            print(f"  diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
-            print(f"  remediation: {' '.join(assessment['remediation'])}")
-            proc.terminate()
-            return False
-        if "error" in call_res:
-            assessment = classify_namespace_probe(
-                name,
-                required_tool=tool_name,
-                registered_tools=tool_names,
-                probe_result={"success": False, "error": call_res["error"]},
-                process={
-                    "pid": proc.pid,
-                    "profile": env.get("GITEA_MCP_PROFILE"),
-                    "env": env,
-                },
-                config_path=config_path,
-                probe_source="offline_spawn",
-            )
-            print(f"  [FAIL] {tool_name} invocation returned error: {call_res['error']}")
-            print(f"  remediation: {' '.join(assessment['remediation'])}")
-            proc.terminate()
-            return False
-
-        assessment = classify_namespace_probe(
-            name,
-            required_tool=tool_name,
-            registered_tools=tool_names,
-            probe_result={"success": True, "result": call_res.get("result")},
-            process={
-                "pid": proc.pid,
-                "profile": env.get("GITEA_MCP_PROFILE"),
-                "env": env,
-            },
-            config_path=config_path,
-                probe_source="offline_spawn",
-        )
-        print(f"  [OK] Successfully invoked required tool '{tool_name}'.")
-        print(f"  diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
         
         proc.terminate()
         return True
-    except Exception as exc:
-        print(f"  [FAIL] Error during handshake: {exc}")
+    except Exception as e:
+        print(f"  [FAIL] Error during handshake: {e}")
         proc.terminate()
         return False
 
-
 def main():
-    parser = argparse.ArgumentParser()
-    parser.add_argument(
-        "--config",
-        default=os.environ.get(
-            "MCP_CONFIG_PATH",
-            os.path.expanduser("~/.gemini/config/mcp_config.json"),
-        ),
-        help="Path to MCP config JSON.",
-    )
-    parser.add_argument(
-        "--namespace",
-        action="append",
-        dest="namespaces",
-        help="Namespace to test. May be repeated.",
-    )
-    args = parser.parse_args()
-
-    config_path = args.config
+    config_path = "/Users/jasonwalker/.gemini/config/mcp_config.json"
     try:
         with open(config_path) as f:
             mcp_config = json.load(f)
@@ -266,17 +112,10 @@ def main():
         sys.exit(1)
         
     servers = mcp_config.get("mcpServers", {})
-    namespaces = args.namespaces or list(REQUIRED_NAMESPACE_TOOLS)
     failed = False
-    for name in namespaces:
+    for name in ["gitea-author", "gitea-reviewer"]:
         if name in servers:
-            if not run_connection_test(
-                name,
-                servers[name],
-                required_tool=REQUIRED_NAMESPACE_TOOLS.get(name),
-                config_path=config_path,
-                probe_source="offline_spawn",
-            ):
+            if not run_connection_test(name, servers[name]):
                 failed = True
         else:
             print(f"Server '{name}' not found in mcp_config.json")

tests/conftest.py

@@ -66,7 +66,6 @@ def _reset_mutation_authority(monkeypatch):
     monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
     monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
     monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
-    monkeypatch.setattr(mcp_server, "_LIVE_NAMESPACE_HEALTH", {})
     monkeypatch.setattr(mcp_server, "_preflight_whoami_called", False)
     monkeypatch.setattr(mcp_server, "_preflight_capability_called", False)
     monkeypatch.setattr(mcp_server, "_preflight_resolved_role", None)

tests/test_mcp_namespace_health.py

@@ -1,208 +0,0 @@
-import unittest
-
-import gitea_mcp_server
-import mcp_namespace_health
-import review_merge_state_machine as rmsm
-
-
-class TestMcpNamespaceHealth(unittest.TestCase):
-    def setUp(self):
-        gitea_mcp_server._LIVE_NAMESPACE_HEALTH.clear()
-
-    def test_registered_tool_but_live_eof_blocks_merge(self):
-        result = mcp_namespace_health.classify_namespace_probe(
-            "gitea-reviewer",
-            required_tool="gitea_whoami",
-            registered_tools=["gitea_whoami", "gitea_list_profiles"],
-            probe_result={
-                "success": False,
-                "error": "client is closing: EOF",
-            },
-            process={
-                "pid": 4321,
-                "profile": "prgs-reviewer",
-                "env": {
-                    "GITEA_MCP_PROFILE": "prgs-reviewer",
-                    "GITEA_TOKEN": "must-not-leak",
-                },
-            },
-            config_path="/Users/jasonwalker/.gemini/config/mcp_config.json",
-            probe_source="client_namespace",
-        )
-
-        self.assertFalse(result["healthy"])
-        self.assertEqual(result["error_type"], "namespace_eof")
-        self.assertTrue(result["required_tool_registered"])
-        self.assertFalse(result["required_tool_callable"])
-        self.assertTrue(result["blocks_merge_workflow"])
-        self.assertFalse(result["ide_namespace_proven"])
-        self.assertEqual(result["probe_source"], "client_namespace")
-        self.assertEqual(result["diagnostics"]["process_pid"], 4321)
-        self.assertEqual(result["diagnostics"]["profile"], "prgs-reviewer")
-        self.assertEqual(
-            result["diagnostics"]["env"]["GITEA_MCP_PROFILE"],
-            "prgs-reviewer",
-        )
-        self.assertNotIn("GITEA_TOKEN", result["diagnostics"]["env"])
-        self.assertIn("gitea-reviewer", result["reasons"][0])
-        self.assertIn("gitea_whoami", result["reasons"][0])
-
-    def test_successful_client_namespace_invocation_is_ide_proven(self):
-        result = mcp_namespace_health.classify_namespace_probe(
-            "gitea-author",
-            registered_tools=["gitea_whoami"],
-            probe_result={"success": True, "result": {"authenticated": True}},
-            process={"pid": 1234, "profile": "prgs-author"},
-            config_path="/tmp/mcp_config.json",
-            probe_source="client_namespace",
-        )
-
-        self.assertTrue(result["healthy"])
-        self.assertTrue(result["required_tool_registered"])
-        self.assertTrue(result["required_tool_callable"])
-        self.assertTrue(result["ide_namespace_proven"])
-        self.assertFalse(result["blocks_merge_workflow"])
-        self.assertIsNone(result["error_type"])
-
-    def test_offline_spawn_success_is_not_ide_proof(self):
-        result = mcp_namespace_health.classify_namespace_probe(
-            "gitea-merger",
-            registered_tools=["gitea_whoami"],
-            probe_result={"success": True, "result": {}},
-            process={"pid": 99, "profile": "prgs-merger"},
-            probe_source="offline_spawn",
-        )
-        self.assertTrue(result["healthy"])
-        self.assertFalse(result["ide_namespace_proven"])
-        self.assertFalse(result["blocks_merge_workflow"])
-        self.assertTrue(
-            any("offline_spawn" in r for r in result["reasons"])
-        )
-
-    def test_registered_missing_required_tool_fails_before_probe_success(self):
-        result = mcp_namespace_health.classify_namespace_probe(
-            "gitea-tools",
-            registered_tools=["gitea_whoami"],
-            probe_result={"success": True, "result": {}},
-            probe_source="client_namespace",
-        )
-
-        self.assertFalse(result["healthy"])
-        self.assertEqual(result["required_tool"], "gitea_list_profiles")
-        self.assertFalse(result["required_tool_registered"])
-        self.assertEqual(result["error_type"], "tool_missing")
-
-    def test_server_tool_exposes_same_assessment_and_records_session(self):
-        result = gitea_mcp_server.gitea_assess_mcp_namespace_health(
-            "gitea-merger",
-            registered_tools=["gitea_whoami"],
-            probe_result={"success": False, "error": "transport closed"},
-            process={"pid": 9876, "profile": "prgs-merger"},
-            probe_source="client_namespace",
-        )
-
-        self.assertFalse(result["success"])
-        self.assertEqual(result["error_type"], "namespace_eof")
-        self.assertEqual(result["namespace"], "gitea-merger")
-        self.assertEqual(result["diagnostics"]["process_pid"], 9876)
-        self.assertIn("gitea-merger", gitea_mcp_server._LIVE_NAMESPACE_HEALTH)
-        gate = gitea_mcp_server._live_namespace_health_gate("merge_pr")
-        self.assertTrue(gate)
-        self.assertTrue(any("gitea-merger" in r for r in gate))
-
-
-class TestLiveNamespaceBlocksMerge(unittest.TestCase):
-    """AC5: a broken live namespace must hard-block the review/merge state machine."""
-
-    def setUp(self):
-        gitea_mcp_server._LIVE_NAMESPACE_HEALTH.clear()
-
-    def _merge_ready_completion(self):
-        # Completion through PRE_MERGE_RECHECK is the state where a clean merge
-        # is otherwise allowed (see test_merge_allowed_with_pre_merge_gates).
-        idx = rmsm.REVIEW_MERGE_STATES.index("PRE_MERGE_RECHECK")
-        return {state: True for state in rmsm.REVIEW_MERGE_STATES[: idx + 1]}
-
-    def _all_gates(self):
-        return {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES}
-
-    def test_assess_workflow_blockers_flags_live_namespace_broken(self):
-        clean = rmsm.assess_workflow_blockers()
-        self.assertFalse(clean["block"])
-
-        broken = rmsm.assess_workflow_blockers(live_namespace_broken=True)
-        self.assertTrue(broken["block"])
-        self.assertTrue(any("namespace" in r.lower() for r in broken["reasons"]))
-
-    def test_can_merge_blocks_even_when_all_gates_pass(self):
-        # Without the namespace blocker a fully-complete workflow can merge...
-        allowed = rmsm.can_merge(
-            self._merge_ready_completion(), pre_merge_gates=self._all_gates()
-        )
-        self.assertTrue(allowed["allowed"])
-
-        # ...but a broken live namespace overrides every satisfied gate.
-        blocked = rmsm.can_merge(
-            self._merge_ready_completion(),
-            pre_merge_gates=self._all_gates(),
-            live_namespace_broken=True,
-        )
-        self.assertTrue(blocked["block"])
-        self.assertFalse(blocked["allowed"])
-
-    def test_workflow_status_forwards_namespace_blocker(self):
-        status = rmsm.workflow_status(
-            self._merge_ready_completion(), live_namespace_broken=True
-        )
-        self.assertFalse(status["merge_allowed"])
-        self.assertFalse(status["approve_allowed"])
-
-    def test_server_tool_blocks_merge_on_live_namespace_broken(self):
-        result = gitea_mcp_server.gitea_assess_review_merge_state_machine(
-            state_completion=self._merge_ready_completion(),
-            pre_merge_gates=self._all_gates(),
-            live_namespace_broken=True,
-        )
-        self.assertTrue(result["blockers"]["block"])
-        self.assertFalse(result["merge"]["allowed"])
-
-    def test_classify_verdict_bridges_into_merge_block(self):
-        # The classify verdict is the intended feed for live_namespace_broken.
-        verdict = mcp_namespace_health.classify_namespace_probe(
-            "gitea-merger",
-            registered_tools=["gitea_whoami", "gitea_adopt_merger_pr_lease"],
-            probe_result={"success": False, "error": "client is closing: EOF"},
-            process={"pid": 555, "profile": "prgs-merger"},
-            probe_source="client_namespace",
-        )
-        self.assertTrue(verdict["blocks_merge_workflow"])
-        blocked = rmsm.can_merge(
-            self._merge_ready_completion(),
-            pre_merge_gates=self._all_gates(),
-            live_namespace_broken=verdict["blocks_merge_workflow"],
-        )
-        self.assertFalse(blocked["allowed"])
-
-    def test_offline_spawn_does_not_authorize_mutation_gate(self):
-        gitea_mcp_server.gitea_assess_mcp_namespace_health(
-            "gitea-merger",
-            registered_tools=["gitea_whoami"],
-            probe_result={"success": True, "result": {}},
-            probe_source="offline_spawn",
-        )
-        gate = gitea_mcp_server._live_namespace_health_gate("merge_pr")
-        self.assertTrue(gate)
-        self.assertTrue(any("offline_spawn" in r or "client_namespace" in r for r in gate))
-
-    def test_client_namespace_healthy_clears_mutation_gate(self):
-        gitea_mcp_server.gitea_assess_mcp_namespace_health(
-            "gitea-merger",
-            registered_tools=["gitea_whoami"],
-            probe_result={"success": True, "result": {}},
-            probe_source="client_namespace",
-        )
-        self.assertEqual(gitea_mcp_server._live_namespace_health_gate("merge_pr"), [])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_review_drafts.py

@@ -0,0 +1,386 @@
+"""Tests for preserving and resuming prepared Gitea review drafts (#609)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+sys_path_root = str(Path(__file__).resolve().parent.parent)
+if sys_path_root not in sys.path:
+    sys.path.insert(0, sys_path_root)
+
+import mcp_session_state
+import gitea_mcp_server
+import reviewer_pr_lease
+
+
+class TestReviewDrafts(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",
+                "GITEA_PROFILE_NAME": "prgs-reviewer",
+                "PYTEST_CURRENT_TEST": "1",  # skip runtime stale checks
+            },
+            clear=False,
+        )
+        self._env.start()
+        
+        # Mock workspace binding to pass in test context
+        self._nwb_patch = patch(
+            "gitea_mcp_server.nwb.assess_namespace_mutation_workspace",
+            return_value={"block": False, "mutation_workspace": "/tmp/test-worktree"},
+        )
+        self._nwb_patch.start()
+
+        # Mock authentication headers
+        self._auth_patch = patch(
+            "gitea_mcp_server.get_auth_header",
+            return_value="Bearer mock-token",
+        )
+        self._auth_patch.start()
+
+        self._username_patch = patch(
+            "gitea_mcp_server._authenticated_username",
+            return_value="sysadmin",
+        )
+        self._username_patch.start()
+        
+        # Clear/Mock local session lease in-memory state
+        reviewer_pr_lease.clear_session_lease()
+
+    def tearDown(self):
+        self._username_patch.stop()
+        self._auth_patch.stop()
+        self._nwb_patch.stop()
+        self._env.stop()
+        self._tmpdir.cleanup()
+        reviewer_pr_lease.clear_session_lease()
+
+    @patch("gitea_mcp_server.api_request")
+    def test_save_draft_success(self, mock_api):
+        mock_api.return_value = {
+            "head": {"sha": "headsha1234567890"},
+            "base": {"ref": "master", "sha": "basesha0987654321"},
+        }
+        res = gitea_mcp_server.gitea_save_review_draft(
+            pr_number=587,
+            action="approve",
+            body="Good changes",
+            expected_head_sha="headsha1234567890",
+            validation_commands="pytest tests/",
+            validation_results="all pass",
+            blocker_reason="stale daemon",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        self.assertTrue(res.get("success"))
+        self.assertEqual(res.get("head_sha"), "headsha1234567890")
+        
+        # Load draft and verify fields
+        draft = mcp_session_state.load_state(
+            kind=mcp_session_state.KIND_REVIEW_DRAFT,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            profile_identity="prgs-reviewer",
+        )
+        self.assertIsNotNone(draft)
+        self.assertEqual(draft.get("pr_number"), 587)
+        self.assertEqual(draft.get("action"), "approve")
+        self.assertEqual(draft.get("body"), "Good changes")
+        self.assertEqual(draft.get("validation_commands"), "pytest tests/")
+        self.assertEqual(draft.get("validation_results"), "all pass")
+        self.assertEqual(draft.get("blocker_reason"), "stale daemon")
+        self.assertEqual(os.path.realpath(draft.get("worktree_path")), os.path.realpath("/tmp/test-worktree"))
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._fetch_pr_comments")
+    @patch("gitea_mcp_server.gitea_submit_pr_review")
+    def test_resume_draft_and_submit_success(self, mock_submit, mock_comments, mock_api):
+        # 1. Save draft
+        mock_api.return_value = {
+            "state": "open",
+            "head": {"sha": "headsha1234567890"},
+            "base": {"ref": "master", "sha": "basesha0987654321"},
+        }
+        
+        gitea_mcp_server.gitea_save_review_draft(
+            pr_number=587,
+            action="approve",
+            body="Good changes",
+            expected_head_sha="headsha1234567890",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        
+        # Seed local session lease
+        reviewer_pr_lease.record_session_lease({
+            "pr_number": 587,
+            "session_id": "session-1234",
+        })
+        
+        # Mock Gitea comments to return active reviewer lease owned by us
+        mock_comments.return_value = [
+            {
+                "id": 1,
+                "user": {"username": "sysadmin"},
+                "body": (
+                    "<!-- mcp-review-lease:v1 -->\n"
+                    "repo: Scaled-Tech-Consulting/Gitea-Tools\n"
+                    "pr: #587\n"
+                    "reviewer_identity: sysadmin\n"
+                    "profile: prgs-reviewer\n"
+                    "session_id: session-1234\n"
+                    "phase: claimed\n"
+                ),
+            }
+        ]
+        
+        mock_submit.return_value = {"performed": True, "success": True}
+        
+        # 2. Resume draft
+        res = gitea_mcp_server.gitea_resume_review_draft(
+            pr_number=587,
+            submit=True,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        
+        self.assertTrue(res.get("success"))
+        self.assertTrue(res.get("marked_ready"))
+        self.assertTrue(res.get("submitted"))
+        mock_submit.assert_called_once()
+        
+        # Verify draft was deleted after successful submit
+        draft = mcp_session_state.load_state(
+            kind=mcp_session_state.KIND_REVIEW_DRAFT,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            profile_identity="prgs-reviewer",
+        )
+        self.assertIsNone(draft)
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._fetch_pr_comments")
+    def test_resume_draft_fails_checks(self, mock_comments, mock_api):
+        # Save a draft
+        mock_api.return_value = {
+            "state": "open",
+            "head": {"sha": "headsha1234567890"},
+            "base": {"ref": "master", "sha": "basesha0987654321"},
+        }
+        gitea_mcp_server.gitea_save_review_draft(
+            pr_number=587,
+            action="approve",
+            body="Good changes",
+            expected_head_sha="headsha1234567890",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        
+        # Seed local session lease
+        reviewer_pr_lease.record_session_lease({
+            "pr_number": 587,
+            "session_id": "session-1234",
+        })
+        
+        # Mock Gitea comments to return active reviewer lease owned by us
+        mock_comments.return_value = [
+            {
+                "id": 1,
+                "user": {"username": "sysadmin"},
+                "body": (
+                    "<!-- mcp-review-lease:v1 -->\n"
+                    "repo: Scaled-Tech-Consulting/Gitea-Tools\n"
+                    "pr: #587\n"
+                    "reviewer_identity: sysadmin\n"
+                    "profile: prgs-reviewer\n"
+                    "session_id: session-1234\n"
+                    "phase: claimed\n"
+                ),
+            }
+        ]
+        
+        # Case A: PR closed
+        mock_api.return_value = {
+            "state": "closed",
+            "head": {"sha": "headsha1234567890"},
+            "base": {"ref": "master", "sha": "basesha0987654321"},
+        }
+        res = gitea_mcp_server.gitea_resume_review_draft(
+            pr_number=587,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        self.assertFalse(res.get("success"))
+        self.assertIn("PR #587 is not open", res.get("reasons")[0])
+        
+        # Case B: Head changed
+        mock_api.return_value = {
+            "state": "open",
+            "head": {"sha": "headsha_NEW"},
+            "base": {"ref": "master", "sha": "basesha0987654321"},
+        }
+        res = gitea_mcp_server.gitea_resume_review_draft(
+            pr_number=587,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        self.assertFalse(res.get("success"))
+        self.assertIn("PR head SHA changed", res.get("reasons")[0])
+        
+        # Case C: Target branch changed
+        mock_api.return_value = {
+            "state": "open",
+            "head": {"sha": "headsha1234567890"},
+            "base": {"ref": "master", "sha": "basesha_NEW"},
+        }
+        res = gitea_mcp_server.gitea_resume_review_draft(
+            pr_number=587,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        self.assertFalse(res.get("success"))
+        self.assertIn("target branch SHA changed", res.get("reasons")[0])
+        
+        # Case D: Worktree mismatch
+        mock_api.return_value = {
+            "state": "open",
+            "head": {"sha": "headsha1234567890"},
+            "base": {"ref": "master", "sha": "basesha0987654321"},
+        }
+        res = gitea_mcp_server.gitea_resume_review_draft(
+            pr_number=587,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/different-worktree",
+        )
+        self.assertFalse(res.get("success"))
+        self.assertIn("worktree binding mismatch", res.get("reasons")[0])
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._fetch_pr_comments")
+    def test_resume_draft_foreign_lease(self, mock_comments, mock_api):
+        mock_api.return_value = {
+            "state": "open",
+            "head": {"sha": "headsha1234567890"},
+            "base": {"ref": "master", "sha": "basesha0987654321"},
+        }
+        gitea_mcp_server.gitea_save_review_draft(
+            pr_number=587,
+            action="approve",
+            body="Good changes",
+            expected_head_sha="headsha1234567890",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        reviewer_pr_lease.record_session_lease({
+            "pr_number": 587,
+            "session_id": "my-session",
+        })
+        mock_comments.return_value = [
+            {
+                "id": 9,
+                "user": {"username": "sysadmin"},
+                "body": (
+                    "<!-- mcp-review-lease:v1 -->\n"
+                    "repo: Scaled-Tech-Consulting/Gitea-Tools\n"
+                    "pr: #587\n"
+                    "reviewer_identity: sysadmin\n"
+                    "profile: prgs-reviewer\n"
+                    "session_id: foreign-session-999\n"
+                    "phase: claimed\n"
+                ),
+            }
+        ]
+        res = gitea_mcp_server.gitea_resume_review_draft(
+            pr_number=587,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        self.assertFalse(res.get("success"))
+        self.assertTrue(
+            any("foreign-session-999" in r or "session_id" in r for r in (res.get("reasons") or []))
+        )
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._fetch_pr_comments")
+    @patch("gitea_mcp_server.terminal_review_hard_stop_reasons")
+    def test_resume_draft_terminal_lock_blocks(self, mock_hard_stop, mock_comments, mock_api):
+        mock_api.return_value = {
+            "state": "open",
+            "head": {"sha": "headsha1234567890"},
+            "base": {"ref": "master", "sha": "basesha0987654321"},
+        }
+        gitea_mcp_server.gitea_save_review_draft(
+            pr_number=587,
+            action="approve",
+            body="Good changes",
+            expected_head_sha="headsha1234567890",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        reviewer_pr_lease.record_session_lease({
+            "pr_number": 587,
+            "session_id": "session-1234",
+        })
+        mock_comments.return_value = [
+            {
+                "id": 1,
+                "user": {"username": "sysadmin"},
+                "body": (
+                    "<!-- mcp-review-lease:v1 -->\n"
+                    "repo: Scaled-Tech-Consulting/Gitea-Tools\n"
+                    "pr: #587\n"
+                    "reviewer_identity: sysadmin\n"
+                    "profile: prgs-reviewer\n"
+                    "session_id: session-1234\n"
+                    "phase: claimed\n"
+                ),
+            }
+        ]
+        mock_hard_stop.return_value = [
+            "terminal review mutation already recorded for PR #587 (approve); #332 hard-stop"
+        ]
+        res = gitea_mcp_server.gitea_resume_review_draft(
+            pr_number=587,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/test-worktree",
+        )
+        self.assertFalse(res.get("success"))
+        self.assertTrue(any("#332" in r or "terminal" in r for r in (res.get("reasons") or [])))
+