feat: make master-parity live-remote aware so a stale daemon fails closed (Closes #610 )

Master-parity previously compared only the daemon's startup commit against the local on-disk HEAD. When the checkout was not pulled, parity reported green even though the live remote master had advanced, so a stale daemon could claim a mutation-safe result while running outdated capability gates (observed during PR #592 recovery, where the resolver correctly required restart but parity said in_parity=true). Changes: - master_parity_gate.assess_master_parity() gains an optional live_remote_head and reports the three commits distinctly (daemon_start_head, local_head, live_remote_head) plus live_known / live_stale / mutation_safe. A result is mutation_safe only when daemon, local checkout, and live remote all agree. - parity_block_reasons() now blocks mutations on live-staleness too; read-only operations remain unblocked (non-goal: never block diagnostics offline). - New parity_resolver_disagreement(): typed fail-closed blocker naming the capability resolver as authoritative when it requires restart but parity looks locally green. - New read_remote_master_head(): best-effort `git ls-remote` for the live target, cached with a 60s TTL (bounded offline latency, no network probe per gate call); env override GITEA_TEST_LIVE_REMOTE_HEAD keeps tests hermetic. - Server: _current_master_parity() reads the live remote head; gitea_assess_master_parity and runtime_context surface the distinguished SHAs, mutation_safe, and resolver-authoritative guidance. Tests: 13 new cases (live-remote parity, live-stale blocking + typed blocker, remote-head reader + TTL cache, server wiring). Full suite: 2423 passed; the 8 remaining failures (test_config TestAuthIntegration, test_credentials TestGetCredentials) are baseline-proven keychain/env failures identical on the unmodified base. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-09 18:52:59 -04:00
13 changed files with 479 additions and 1832 deletions
@@ -1,156 +0,0 @@
 # ADR: Stable control runtime vs dev runtime (Gitea MCP)
 - **Status:** Accepted (policy effective immediately for LLM sessions; tooling may lag)
 - **Date:** 2026-07-09
 - **Tracking issue:** [#615](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/615)
 - **Related:**
  - [#543](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/543) / `docs/mcp-namespace-health.md` — client-namespace health
  - `docs/mcp-namespace-eof-recovery.md` — reconnect-only EOF recovery (no PID kill)
  - [#558](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/558) / `docs/mcp-daemon-import-guard.md` — sanctioned daemon
  - [#557](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/557) / `docs/bootstrap-review-path.md` — controller bootstrap for self-hosted fixes
  - Allocator / control-plane ADR: `docs/architecture/mcp-allocator-control-plane-observability-adr.md` (#613 / PR #614)
 ## 1. Context
 The Gitea MCP server is the **control plane** for real issue/PR mutations (create, comment, lock, review, merge, etc.). When author/reviewer/merger/reconciler sessions kill or restart that process, relaunch it from a feature worktree, or edit the checkout that process loads, operators observe:
 - Mid-session identity/preflight resets
 - Stale-runtime vs master parity failures
 - IDE transport EOF / “tool not found” while code on disk has changed
 - Accidental production mutations from experimental code
 This ADR separates **stable control runtime** from **dev/test runtime** and defines promotion proof.
 ## 2. Decision
 ### 2.1 Stable control runtime
 The Gitea MCP server used for **real workflow mutations** is the **stable control runtime**.
 Characteristics:
 - Loads a known, promoted revision of Gitea-Tools (or the packaged release layout operators designate)
 - Registered in the IDE/client as the production namespaces (`gitea-tools`, `gitea-reviewer`, `gitea-merger`, `gitea-reconciler`, etc.)
 - Holds production profile credentials via sanctioned keychain/env paths only
 ### 2.2 Dev / test runtime
 MCP **server code** development and testing:
 - Happens in isolated **`branches/`** worktrees (or other non-stable checkouts)
 - May use a **separate** dev/test MCP runtime/process when process-level testing is required
 - **Must not** be used for real Gitea mutations on production issues/PRs
 ### 2.3 Forbidden actions (normal sessions)
 Normal **author, reviewer, merger, and reconciler** sessions **must not**:
 | Forbidden | Why |
 |-----------|-----|
 | Kill the running MCP server process | Drops all concurrent sessions; loses preflight state |
 | Restart the MCP server process | Same as kill; causes stale/identity churn mid-workflow |
 | Relaunch MCP from a development worktree | Runs unpromoted code against production mutations |
 | Edit files in the stable runtime checkout | Hot-mutates control plane under concurrent users |
 | Use experimental/dev MCP for real Gitea mutations | Bypasses promotion proof and audit expectations |
 EOF / transport recovery: **client reconnect only** (see `docs/mcp-namespace-eof-recovery.md`). Do not “fix” health by killing PIDs or bumping MCP config mtimes as a normal session procedure.
 ### 2.4 Promotion (operator / release-manager only)
 Promotion of a new revision into the stable control runtime is an **explicit operator/release-manager action**, not an LLM self-service step.
 A promotion **must record** (issue comment, release note, or promotion ledger):
 | Field | Description |
 |-------|-------------|
 | **previous runtime SHA** | Commit previously loaded by stable runtime |
 | **promoted runtime SHA** | Commit after promotion |
 | **source branch/PR** | Where the change was reviewed |
 | **restart/reload method** | How the process was cycled (e.g. supervised restart, client reload) |
 | **health check proof** | Client-namespace probe success (`gitea_whoami` / namespace health) |
 | **identity/profile proof** | Expected profile(s) and username(s) after reload |
 | **workspace/root proof** | Stable checkout path / root matches intended layout |
 | **mutation capability proof** | Required permissions for the target role present; forbidden ops still forbidden |
 | **rollback instructions** | How to restore previous SHA and re-verify health |
 Suggested durable marker:
 ```text
 ## MCP STABLE RUNTIME PROMOTION (#615)
 Status: COMPLETED | ROLLED_BACK | ABORTED
 Previous-SHA: <full sha>
 Promoted-SHA: <full sha>
 Source-PR: <number>
 Source-Branch: <name>
 Reload-Method: <text>
 Health-Proof: client_namespace whoami OK / assess_mcp_namespace_health OK
 Identity-Proof: profile=<name> user=<name>
 Workspace-Proof: root=<path>
 Mutation-Proof: allowed_ops include <…>; forbidden include <…>
 Rollback: checkout <previous sha>; reload method <…>; re-run health/identity proofs
 Operator: <username>
 Timestamp: <ISO-8601>
 ```
 ### 2.5 Unhealthy stable runtime → stop work
 If the stable MCP runtime is **unhealthy** (client-namespace probes fail, wrong identity, wrong root, missing mutation capability, persistent EOF after reconnect):
 1. **Normal PR / review / merge / issue-mutation work must stop.**
 2. Do **not** improvise by switching to a dev worktree MCP for production mutations.
 3. Resume only after:
   - runtime is restored, **or**
   - a **controlled** promotion/rollback completes with the promotion record above, **or**
   - a controller invokes the narrow **bootstrap review path** (#557) when the defect is self-hosted and documented.
 ## 3. Relationship to other controls
 | Doc / mechanism | Interaction |
 |-----------------|-------------|
 | Namespace health (#543) | Proves IDE client can call tools; does not authorize restart |
 | EOF recovery | Reconnect only; no process kill |
 | Daemon import guard (#558) | Mutations require sanctioned daemon; not a bare shell import |
 | Bootstrap path (#557) | Only controller-authorized exception when live runtime cannot review its own fix |
 | Allocator / control-plane ADR | Coordination DB is separate; still depends on a healthy MCP surface for Gitea writes |
 ## 4. Consequences
 ### Positive
 - Predictable control plane for concurrent LLMs
 - Clear operator-only promotion gate with rollback
 - Aligns session behavior with health/EOF docs already landed
 ### Costs
 - LLM sessions must wait when runtime is sick (no DIY restart)
 - Operators must maintain promotion discipline and dual-runtime config if they use a dev MCP
 ### Non-goals
 - Does not ban operator-supervised restarts during incidents
 - Does not replace CI or code review for MCP changes
 - Does not authorize editing stable checkout “because tests need a quick fix”
 ## 5. Implementation follow-ups (optional tooling)
 These may land in later issues; the **policy binds sessions now**:
 1. Session preflight that refuses mutations if workspace root equals a `branches/` feature worktree configured as “dev only.”
 2. Explicit `runtime_kind=stable|dev` in MCP config and `gitea_whoami` profile metadata.
 3. Promotion checklist script that emits the durable promotion marker fields.
 4. Operator Guide wiki cross-link to this ADR.
 ## 6. Acceptance for this ADR
 1. Document merged under `docs/architecture/`.
 2. Issue #615 references this path.
 3. LLM/operator runbooks treat kill/restart/relaunch-from-worktree as **violations**.
 4. Unhealthy runtime stops normal mutation work until restore/promotion/rollback/bootstrap.
 ## 7. Document history
 | Date | Change |
 |------|--------|
 | 2026-07-09 | Initial ADR: stable vs dev runtime, forbidden session actions, promotion proof fields, stop-work rule |
@@ -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.
@@ -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.
@@ -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,335 +4162,6 @@ 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,
@@ -4979,12 +4615,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(
@@ -6979,10 +6609,39 @@ def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
    return False
 def _git_default_remote_name(root: str) -> str:
    """First configured git remote name for *root*, defaulting to 'origin'.
    Used to resolve the live remote master target for parity (#610). Best
    effort: any failure falls back to 'origin' so callers never raise.
    """
    try:
        res = subprocess.run(
            ["git", "-C", root, "remote"],
            capture_output=True, text=True, check=False,
        )
    except Exception:
        return "origin"
    if res.returncode != 0:
        return "origin"
    names = [n.strip() for n in (res.stdout or "").splitlines() if n.strip()]
    return names[0] if names else "origin"
 def _current_master_parity() -> dict:
-    """Assess this process's code against the on-disk master HEAD (#420)."""
+    """Assess this process's code against local and live remote master (#420/#610).
    Compares the daemon's startup commit, the on-disk checkout HEAD, and the
    live remote master target. A stale daemon relative to live master fails
    closed for mutations even when the local checkout HEAD still matches the
    startup commit. The live-remote read is best effort: an unresolved live
    head leaves read-only diagnostics unblocked but is never mutation-safe.
    """
    current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
-    return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head)
+    live_head = master_parity_gate.read_remote_master_head(
        PROJECT_ROOT, remote=_git_default_remote_name(PROJECT_ROOT))
    return master_parity_gate.assess_master_parity(
        _STARTUP_PARITY, current_head, live_remote_head=live_head)
 def _master_parity_block(op: str) -> list[str]:
@@ -8994,45 +8653,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).
+    """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``.
    """
    completion = state_completion or {}
-    blocker_kwargs = {
+    blockers = review_merge_state_machine.assess_workflow_blockers(
-        "infra_stop": infra_stop,
+        infra_stop=infra_stop,
-        "capability_blocked": capability_blocked,
+        capability_blocked=capability_blocked,
-        "live_namespace_broken": live_namespace_broken,
+    )
    }
    blockers = review_merge_state_machine.assess_workflow_blockers(**blocker_kwargs)
    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"] = (
@@ -9317,10 +8971,29 @@ def gitea_get_runtime_context(
        "restart_required": parity["restart_required"],
        "startup_head": parity["startup_head"],
        "current_head": parity["current_head"],
        # #610 distinguished mutation-safety signals:
        "daemon_start_head": parity["daemon_start_head"],
        "local_head": parity["local_head"],
        "live_remote_head": parity["live_remote_head"],
        "live_known": parity["live_known"],
        "live_stale": parity["live_stale"],
        "mutation_safe": parity["mutation_safe"],
        "summary": master_parity_gate.format_parity(parity),
        "mutation_gate_enforced": not master_parity_gate.gate_disabled(),
        # #610: the capability resolver is authoritative for mutation safety;
        # local parity alone must never authorize a mutation.
        "resolver_authoritative_for_mutation_safety": True,
    }
-    if parity["stale"] and not master_parity_gate.gate_disabled():
+    if parity["restart_required"] and not master_parity_gate.gate_disabled():
        if parity["live_stale"]:
            safe_next_action = (
                "Daemon is stale relative to LIVE remote master "
                f"(started {parity['startup_head'][:12] if parity['startup_head'] else 'unknown'}, "
                f"live master {parity['live_remote_head'][:12] if parity['live_remote_head'] else 'unknown'}); "
                "restart/reconnect the Gitea MCP server before mutating. The "
                "capability resolver is authoritative for mutation safety."
            )
        else:
            safe_next_action = (
                "Server code is stale relative to master; restart the Gitea MCP "
                "server to load current capability gates before mutating. "
@@ -9364,12 +9037,19 @@ def gitea_assess_master_parity(
        "determinable": parity["determinable"],
        "startup_head": parity["startup_head"],
        "current_head": parity["current_head"],
        # #610 distinguished mutation-safety signals:
        "daemon_start_head": parity["daemon_start_head"],
        "local_head": parity["local_head"],
        "live_remote_head": parity["live_remote_head"],
        "live_known": parity["live_known"],
        "live_stale": parity["live_stale"],
        "mutation_safe": parity["mutation_safe"],
        "mutation_gate_enforced": enforced,
        "summary": master_parity_gate.format_parity(parity),
        "reasons": parity["reasons"],
        "process_root": PROJECT_ROOT,
    }
-    if parity["stale"] and enforced:
+    if parity["restart_required"] and enforced:
        out["report"] = master_parity_gate.parity_report(parity)
    return out
 def gitea_record_pre_review_command(
@@ -9562,46 +9242,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,
@@ -24,12 +24,26 @@ from __future__ import annotations
 import os
 import subprocess
 import time
 # Live-remote head cache: the parity gate runs on every mutation and every
 # runtime-context read, so the ``git ls-remote`` result is cached briefly to
 # avoid a network round-trip per call (#610). Keyed by (root, remote, branch).
 _REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {}
 _REMOTE_HEAD_TTL = 60.0
 def _clear_remote_head_cache() -> None:
    """Reset the live-remote head cache (test isolation / forced refresh)."""
    _REMOTE_HEAD_CACHE.clear()
 # Environment escape hatches (ops + tests):
 #   GITEA_MCP_DISABLE_PARITY_GATE  -> disable enforcement entirely (fail open).
 #   GITEA_TEST_CURRENT_HEAD        -> force the "current" HEAD read, for tests.
 ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
 ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
 # GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests.
 ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD"
 def read_git_head(root: str) -> str | None:
@@ -58,6 +72,57 @@ def read_git_head(root: str) -> str | None:
    return (res.stdout or "").strip() or None
 def read_remote_master_head(
    root: str,
    remote: str = "origin",
    branch: str = "master",
    ttl: float = _REMOTE_HEAD_TTL,
 ) -> str | None:
    """Return the live remote ``branch`` commit SHA, or ``None`` (#610).
    Resolves the *live* target commit via ``git ls-remote`` so parity can tell
    a daemon that is behind the live remote master apart from one whose local
    checkout simply hasn't been pulled. ``None`` means the live head could not
    be resolved (offline, no such remote, git unavailable, error) -- callers
    must treat unknown live state as *not mutation-safe* while never blocking
    read-only diagnostics. A ``GITEA_TEST_LIVE_REMOTE_HEAD`` override takes
    precedence so the wiring can be exercised deterministically and offline.
    The result is cached for *ttl* seconds per (root, remote, branch) so the
    gate does not run a network probe on every mutation/read (``ttl=0`` forces
    a live probe). Both hits and ``None`` misses are cached to bound offline
    latency; the env override bypasses the cache and the subprocess entirely.
    """
    forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
    if forced is not None:
        return forced.strip() or None
    if not root:
        return None
    key = (root, remote, branch)
    now = time.monotonic()
    if ttl > 0:
        cached = _REMOTE_HEAD_CACHE.get(key)
        if cached is not None and (now - cached[0]) < ttl:
            return cached[1]
    sha: str | None = None
    try:
        res = subprocess.run(
            ["git", "-C", root, "ls-remote", remote, f"refs/heads/{branch}"],
            capture_output=True,
            text=True,
            check=False,
            timeout=5,
        )
        if res.returncode == 0:
            lines = (res.stdout or "").strip().splitlines()
            if lines:
                sha = lines[0].split("\t", 1)[0].split()[0].strip() or None
    except Exception:
        sha = None
    _REMOTE_HEAD_CACHE[key] = (now, sha)
    return sha
 def capture_startup_parity(root: str, head: str | None = None) -> dict:
    """Capture the process source-tree baseline once at server startup.
@@ -72,18 +137,38 @@ def _short(sha: str | None) -> str:
    return sha[:12] if sha else "unknown"
-def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
+def assess_master_parity(
    startup: dict | None,
    current_head: str | None,
    live_remote_head: str | None = None,
 ) -> dict:
    """Compare the startup baseline against the current on-disk ``HEAD``.
-    Pure: both HEADs are supplied by the caller. Returns a structured result:
+    Pure: all HEADs are supplied by the caller. Returns a structured result:
    - ``in_parity``    -- server code matches the on-disk master (or parity
      could not be determined, which is not treated as stale).
    - ``stale``        -- the on-disk master has definitively advanced past the
      running process.
-    - ``restart_required`` -- alias of ``stale``; the recovery action.
+    - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
-    - ``determinable`` -- whether both HEADs were known well enough to compare.
+    - ``determinable`` -- whether both local HEADs were known well enough to
      compare.
    - ``startup_head`` / ``current_head`` / ``reasons``.
    #610 adds live-remote awareness so a daemon that is stale relative to the
    *live* remote master cannot report a mutation-safe result even when the
    local checkout HEAD still matches the daemon's startup commit:
    - ``daemon_start_head`` -- the commit the running process started at
      (alias of ``startup_head``, named for clarity in reports).
    - ``local_head`` -- the on-disk checkout HEAD (alias of ``current_head``).
    - ``live_remote_head`` -- the live remote target commit, or ``None`` when it
      could not be fetched.
    - ``live_known`` -- whether the live remote target was resolved.
    - ``live_stale`` -- the live remote master has advanced past the running
      process (daemon is behind live master) even if local parity is green.
    - ``mutation_safe`` -- the daemon code, local checkout, and live remote
      target all agree; the only state in which a mutation may rely on parity.
    """
    startup_head = (startup or {}).get("startup_head")
    reasons: list[str] = []
@@ -91,32 +176,56 @@ def assess_master_parity(startup: dict | None, current_head: str | None) -> dict
    if startup_head is None:
        reasons.append(
            "startup commit was not captured; code parity cannot be enforced")
-        return _result(True, False, False, startup_head, current_head, reasons)
+        return _result(True, False, False, startup_head, current_head,
                       live_remote_head, False, reasons)
    if current_head is None:
        reasons.append(
            "current workspace HEAD could not be read; code parity cannot be "
            "enforced")
-        return _result(True, False, False, startup_head, current_head, reasons)
+        return _result(True, False, False, startup_head, current_head,
-
+                       live_remote_head, False, reasons)
    if startup_head == current_head:
        return _result(True, False, True, startup_head, current_head, reasons)
    local_in_parity = startup_head == current_head
    local_stale = not local_in_parity
    if local_stale:
        reasons.append(
-        f"MCP server started at commit {_short(startup_head)} but the workspace "
+            f"MCP server started at commit {_short(startup_head)} but the "
-        f"master is now {_short(current_head)}; restart the server to load the "
+            f"workspace master is now {_short(current_head)}; restart the "
-        f"current capability gates")
+            f"server to load the current capability gates")
-    return _result(False, True, True, startup_head, current_head, reasons)
+
    live_known = live_remote_head is not None
    live_stale = live_known and live_remote_head != startup_head
    if live_stale:
        reasons.append(
            f"live remote master is {_short(live_remote_head)} but the MCP "
            f"server started at {_short(startup_head)}; the daemon is stale "
            f"relative to live master -- restart/reconnect before mutating")
    return _result(
        local_in_parity, local_stale, True, startup_head, current_head,
        live_remote_head, live_stale, reasons)
-def _result(in_parity, stale, determinable, startup_head, current_head, reasons):
+def _result(in_parity, stale, determinable, startup_head, current_head,
            live_remote_head, live_stale, reasons):
    live_known = live_remote_head is not None
    mutation_safe = (
        determinable and in_parity and live_known and not live_stale)
    return {
        "in_parity": in_parity,
        "stale": stale,
-        "restart_required": stale,
+        "restart_required": stale or live_stale,
        "determinable": determinable,
        "startup_head": startup_head,
        "current_head": current_head,
        # #610 distinguished signals:
        "daemon_start_head": startup_head,
        "local_head": current_head,
        "live_remote_head": live_remote_head,
        "live_known": live_known,
        "live_stale": live_stale,
        "mutation_safe": mutation_safe,
        "reasons": list(reasons),
    }
@@ -130,11 +239,13 @@ def parity_block_reasons(assessment: dict) -> list[str]:
    """Block reasons for a mutation gate (empty when the mutation may proceed).
    A disabled gate or an in-parity / non-determinable assessment yields no
-    reasons; only a definitively stale server blocks.
+    reasons. A definitively stale server blocks, and (#610) a daemon that is
    stale relative to the *live* remote master blocks even when the local
    checkout HEAD still matches the daemon's startup commit.
    """
    if gate_disabled():
        return []
-    if assessment.get("stale"):
+    if assessment.get("stale") or assessment.get("live_stale"):
        return list(assessment.get("reasons") or
                    ["server code is stale relative to master (fail closed)"])
    return []
@@ -147,6 +258,10 @@ def parity_report(assessment: dict) -> dict:
        "restart_required": True,
        "startup_head": assessment.get("startup_head"),
        "current_head": assessment.get("current_head"),
        # #610: name the live remote target so the report distinguishes a
        # local-code stale from a daemon-behind-live-master stale.
        "live_remote_head": assessment.get("live_remote_head"),
        "live_stale": bool(assessment.get("live_stale")),
        "reasons": list(assessment.get("reasons") or []),
        "recovery": [
            "The running MCP server is executing code older than the current "
@@ -157,6 +272,45 @@ def parity_report(assessment: dict) -> dict:
    }
 def parity_resolver_disagreement(
    assessment: dict,
    resolver_restart_required: bool,
 ) -> dict | None:
    """Typed blocker when the resolver requires restart but parity looks green.
    The capability resolver (``gitea_resolve_task_capability``) detects stale
    runtime authoritatively for mutation safety (#610). When it requires a
    restart, local-only parity must never override it: this returns a typed,
    fail-closed blocker that names the resolver as authoritative. Returns
    ``None`` when the resolver does not require a restart.
    """
    if not resolver_restart_required:
        return None
    parity_optimistic = bool(assessment.get("in_parity")) and not (
        assessment.get("stale") or assessment.get("live_stale"))
    return {
        "kind": "parity_resolver_disagreement",
        "restart_required": True,
        "resolver_authoritative": True,
        "parity_optimistic": parity_optimistic,
        "daemon_start_head": assessment.get("daemon_start_head"),
        "local_head": assessment.get("local_head"),
        "live_remote_head": assessment.get("live_remote_head"),
        "reasons": [
            "The capability resolver requires a restart/reconnect (stale "
            "runtime) but master-parity reported local code as in-parity. "
            "The resolver is authoritative for mutation safety; do not mutate "
            "on local parity alone. Restart/reconnect the Gitea MCP server "
            "and re-verify before mutating.",
        ],
        "recovery": [
            "Trust the resolver: treat this session as stale.",
            "Restart or /mcp reconnect the Gitea MCP namespace so it reloads "
            "current master and live target state, then re-run preflight.",
        ],
    }
 def format_parity(assessment: dict) -> str:
    """One-line human summary for logs / runtime context."""
    if assessment.get("stale"):
@@ -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 []
@@ -30,9 +30,6 @@ 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"
@@ -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).
+    """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``.
    """
    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.
+    """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).
    """
    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:
@@ -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,
+Spawns the MCP server processes as defined in the IDE's global config,
-performs the JSON-RPC handshake, verifies the required tool is registered,
+performs the JSON-RPC handshake, and queries the tools list to verify
-and invokes that tool. Results are classified with
+that the connection is fully operational and doesn't return EOF.
 ``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``.
 """
 import argparse
 import json
 import os
 import subprocess
 import sys
-from mcp_namespace_health import REQUIRED_NAMESPACE_TOOLS, classify_namespace_probe
+def run_connection_test(name, config):
 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):
    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:
+    except Exception as e:
-        print(f"  [FAIL] Failed to spawn process: {exc}")
+        print(f"  [FAIL] Failed to spawn process: {e}")
        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)}")
        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)
+        line = proc.stdout.readline()
-        if res is None:
+        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)
+        line = proc.stdout.readline()
-        if res is None:
+        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:
+    except Exception as e:
-        print(f"  [FAIL] Error during handshake: {exc}")
+        print(f"  [FAIL] Error during handshake: {e}")
        proc.terminate()
        return False
 def main():
-    parser = argparse.ArgumentParser()
+    config_path = "/Users/jasonwalker/.gemini/config/mcp_config.json"
    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
    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(
+            if not run_connection_test(name, servers[name]):
                name,
                servers[name],
                required_tool=REQUIRED_NAMESPACE_TOOLS.get(name),
                config_path=config_path,
                probe_source="offline_spawn",
            ):
                failed = True
        else:
            print(f"Server '{name}' not found in mcp_config.json")
@@ -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)
@@ -78,6 +78,95 @@ class TestBlockReasonsAndReport(unittest.TestCase):
        self.assertTrue(report["recovery"])
 class TestLiveRemoteParity(unittest.TestCase):
    """#610: parity must account for the live remote master, not just local.
    The daemon can be stale relative to the live remote target while the local
    checkout HEAD still matches the daemon's startup commit, so local parity
    reports green even though a mutation would run against outdated code.
    """
    SHA_C = "c" * 40
    def test_distinguishes_three_shas(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        self.assertEqual(res["daemon_start_head"], SHA_A)
        self.assertEqual(res["local_head"], SHA_A)
        self.assertEqual(res["live_remote_head"], SHA_B)
    def test_mutation_safe_only_when_all_three_match(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_A)
        self.assertTrue(res["mutation_safe"])
        self.assertTrue(res["live_known"])
        self.assertFalse(res["live_stale"])
    def test_live_stale_when_remote_advanced_past_daemon(self):
        # Local checkout still matches the daemon start (local parity green),
        # but the live remote master has advanced -> daemon is live-stale.
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        self.assertTrue(res["in_parity"])  # local parity still green
        self.assertTrue(res["live_stale"])
        self.assertFalse(res["mutation_safe"])
        self.assertTrue(any("live" in r.lower() for r in res["reasons"]))
    def test_live_unknown_is_not_mutation_safe_but_not_stale(self):
        # Non-goal: unfetchable live remote must not be treated as stale for
        # read-only, but a mutation-safe claim fails closed.
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=None)
        self.assertFalse(res["live_known"])
        self.assertFalse(res["mutation_safe"])
        self.assertFalse(res["live_stale"])
        self.assertTrue(res["in_parity"])
    def test_default_live_remote_preserves_legacy_shape(self):
        # Callers that do not supply a live head keep the pre-#610 behavior:
        # in-parity, not live-stale, no live-derived block.
        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
        self.assertFalse(res["live_stale"])
        self.assertEqual(mp.parity_block_reasons(res), [])
 class TestLiveStaleBlockAndReport(unittest.TestCase):
    """#610: live-staleness must block mutations and surface a typed blocker."""
    def test_live_stale_produces_block_reasons(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        self.assertTrue(mp.parity_block_reasons(res))
    def test_disable_env_suppresses_live_stale_block(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
            self.assertEqual(mp.parity_block_reasons(res), [])
    def test_resolver_disagreement_returns_typed_blocker(self):
        # Parity says local-green, resolver says restart required -> disagreement
        # is a typed, fail-closed blocker naming the resolver as authoritative.
        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
        blocker = mp.parity_resolver_disagreement(res, resolver_restart_required=True)
        self.assertIsNotNone(blocker)
        self.assertEqual(blocker["kind"], "parity_resolver_disagreement")
        self.assertTrue(blocker["restart_required"])
        self.assertTrue(blocker["resolver_authoritative"])
    def test_no_disagreement_when_resolver_agrees(self):
        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
        self.assertIsNone(
            mp.parity_resolver_disagreement(res, resolver_restart_required=False))
    def test_live_stale_report_names_live_remote(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        report = mp.parity_report(res)
        self.assertEqual(report["live_remote_head"], SHA_B)
        self.assertTrue(report["restart_required"])
 class TestReadGitHead(unittest.TestCase):
    def test_test_override_takes_precedence(self):
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
@@ -95,6 +184,78 @@ class TestReadGitHead(unittest.TestCase):
            self.assertIsNone(mp.read_git_head(""))
 class TestReadRemoteMasterHead(unittest.TestCase):
    """#610: live remote master head reader (env-overridable, fails to None)."""
    def test_test_override_takes_precedence(self):
        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
            self.assertEqual(mp.read_remote_master_head("/nonexistent"), SHA_B)
    def test_blank_override_is_none(self):
        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: "   "}):
            self.assertIsNone(mp.read_remote_master_head("/nonexistent"))
    def test_unfetchable_remote_is_none(self):
        # No override; a bogus root/remote must fail closed to None, never raise.
        env = {k: v for k, v in os.environ.items()
               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
        with patch.dict(os.environ, env, clear=True):
            self.assertIsNone(
                mp.read_remote_master_head("/nonexistent", remote="nope"))
 class TestRemoteHeadCache(unittest.TestCase):
    """#610: live remote reads are cached with a TTL to stay off the network.
    The parity gate runs on every mutation and every runtime-context read, so an
    unbounded ``git ls-remote`` per call would be a latency/flakiness regression.
    """
    def setUp(self):
        mp._clear_remote_head_cache()
        env = {k: v for k, v in os.environ.items()
               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
        self._env = patch.dict(os.environ, env, clear=True)
        self._env.start()
        self.addCleanup(self._env.stop)
        self.addCleanup(mp._clear_remote_head_cache)
    def _fake_run(self, sha):
        class _R:
            returncode = 0
            stdout = f"{sha}\trefs/heads/master\n"
        calls = {"n": 0}
        def run(*args, **kwargs):
            calls["n"] += 1
            return _R()
        return run, calls
    def test_second_call_within_ttl_uses_cache(self):
        run, calls = self._fake_run(SHA_B)
        with patch.object(mp.subprocess, "run", run):
            a = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
            b = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
        self.assertEqual(a, SHA_B)
        self.assertEqual(b, SHA_B)
        self.assertEqual(calls["n"], 1)
    def test_zero_ttl_bypasses_cache(self):
        run, calls = self._fake_run(SHA_B)
        with patch.object(mp.subprocess, "run", run):
            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
        self.assertEqual(calls["n"], 2)
    def test_env_override_never_touches_subprocess(self):
        run, calls = self._fake_run(SHA_B)
        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
            with patch.object(mp.subprocess, "run", run):
                self.assertEqual(
                    mp.read_remote_master_head("/repo", remote="prgs"), SHA_A)
        self.assertEqual(calls["n"], 0)
 class TestServerWiring(unittest.TestCase):
    """Integration with the gate choke point in the server namespace."""
@@ -105,6 +266,13 @@ class TestServerWiring(unittest.TestCase):
        self._saved = self.srv._STARTUP_PARITY
        self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
                                    "startup_head": SHA_A}
        # Keep the live-remote read hermetic (no real ls-remote network call):
        # default the live master to the daemon start so parity is fully green
        # unless a test overrides the live head explicitly (#610).
        self._live_patch = patch.dict(
            os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A})
        self._live_patch.start()
        self.addCleanup(self._live_patch.stop)
    def tearDown(self):
        self.srv._STARTUP_PARITY = self._saved
@@ -147,6 +315,36 @@ class TestServerWiring(unittest.TestCase):
            self.assertTrue(out["in_parity"])
            self.assertNotIn("report", out)
    # --- #610: live-remote wiring -------------------------------------------
    def test_live_stale_blocks_mutation_though_local_green(self):
        # Local checkout matches the daemon start (local parity green) but the
        # live remote master has advanced -> mutations must fail closed.
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
            self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
            self.assertTrue(
                self.srv._master_parity_block("gitea.pr.create"))
    def test_assess_tool_exposes_three_distinct_shas(self):
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
            out = self.srv.gitea_assess_master_parity(remote="prgs")
            self.assertEqual(out["daemon_start_head"], SHA_A)
            self.assertEqual(out["local_head"], SHA_A)
            self.assertEqual(out["live_remote_head"], SHA_B)
            self.assertTrue(out["live_stale"])
            self.assertFalse(out["mutation_safe"])
            self.assertIn("report", out)
    def test_assess_tool_mutation_safe_when_all_three_match(self):
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
            out = self.srv.gitea_assess_master_parity(remote="prgs")
            self.assertTrue(out["mutation_safe"])
            self.assertFalse(out["live_stale"])
            self.assertNotIn("report", out)
 if __name__ == "__main__":
    unittest.main()
@@ -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()
@@ -1,287 +0,0 @@
 """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):
        print("GITEA_MCP_SERVER FILE:", gitea_mcp_server.__file__)
        print("DIR OF GITEA_MCP_SERVER:", dir(gitea_mcp_server))
        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])