fix: isolate review-mutation ledger from host session-state (#590 )

Tests that call patch.dict(os.environ, clear=True) dropped GITEA_MCP_SESSION_STATE_DIR and re-adopted the operator host cache under ~/.cache/gitea-tools/session-state. A residual terminal request_changes mutation then tripped the #332 hard-stop before test_unknown_profile_blocks could assert the empty-profile gate. - Pin mcp_session_state.default_state_dir / DEFAULT_STATE_DIR to a per-test temp dir in conftest (still honors an explicit env override) - Clear the decision lock at the start of each TestMergePR test - Add regression coverage for env-clear isolation Closes #590
2026-07-09 14:16:39 -04:00
36 changed files with 75 additions and 4815 deletions
@@ -1,63 +0,0 @@
 """Controller-closure baseline-proof gate (#529, criterion 6).
 A controller (or any session) may close a tracking issue with a closure
 report that summarizes validation. When that report calls a non-zero
 test-suite exit an "expected pre-existing failure" (or baseline / known
 failure) it must carry pre-merge proof; otherwise the closure buries an
 unproven regression as an accepted baseline and weakens controller
 confidence in the durable state.
 This module fails closed on exactly that case by reusing the pre-merge
 baseline proof verifier (#533). A closure report is allowed when it is
 empty (no validation claim), a clean pass, or a baseline failure proven on
 the PR pre-merge base commit (or a documented known-failure record that
 predates the PR).
 """
 from __future__ import annotations
 from typing import Any
 from premerge_baseline_proof import CLEAN_PASS, assess_premerge_baseline_proof
 def assess_controller_closure_baseline_proof(
    closure_report: str | None,
 ) -> dict[str, Any]:
    """Assess whether an issue-closure report may proceed.
    Returns a dict with ``block``, ``proven``, ``label``, ``reasons``,
    ``skipped`` and ``safe_next_action``. Only blocks when the closure
    report claims a non-zero validation exit is an expected
    pre-existing/baseline failure without valid pre-merge proof.
    """
    text = (closure_report or "").strip()
    if not text:
        return {
            "block": False,
            "proven": True,
            "label": CLEAN_PASS,
            "reasons": [],
            "skipped": True,
            "safe_next_action": "",
        }
    result = assess_premerge_baseline_proof(text)
    block = bool(result.get("block"))
    return {
        "block": block,
        "proven": not block,
        "label": result.get("label"),
        "reasons": list(result.get("reasons") or []),
        "skipped": bool(result.get("skipped", False)),
        "safe_next_action": (
            result.get("safe_next_action")
            or (
                "provide pre-merge baseline proof (base commit, tested commit, "
                "command, exit status, failure signature) before closing on an "
                "expected pre-existing failure"
            )
        )
        if block
        else "",
    }
@@ -1,198 +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** LLM sessions **must not**:
 | Forbidden | Why |
 |-----------|-----|
 | Kill the running MCP server process | Drops all concurrent sessions; loses preflight state |
 | Restart / relaunch 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 |
 | Bypass or self-reset a stale master-parity gate | The gate is fail-closed; only an operator reload restores parity |
 **LLM-allowed vs operator-owned (authoritative split):**
 | Actor | May do | Must not do |
 |-------|--------|-------------|
 | **LLM session** | Call tools on the already-running stable namespaces; **client reconnect** after transport EOF (no process kill); pass `worktree_path` / role worktree args; report blockers and stop mutations when unhealthy/stale | Kill, restart, or relaunch any MCP process; bump config mtimes to force reload; edit the stable checkout; switch to a dev MCP for production mutations |
 | **Operator / release-manager** | Supervised restart/reload of the **stable** control runtime; dual-namespace client configuration; §2.4 promotions; incident recovery | — |
 EOF / transport recovery for LLM sessions: **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.
 This ADR **supersedes** any older runbook wording that told the LLM to relaunch or restart the client/MCP as a self-service step. Where `docs/llm-workflow-runbooks.md` (or wiki runbooks) discuss dual-namespace setup or workspace rebind, **process restart/relaunch is operator-owned**; the LLM stops, reports, and waits.
 ### 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**, including any of:
 - client-namespace probes fail
 - wrong identity / wrong profile
 - wrong workspace root
 - missing mutation capability for the intended role
 - persistent EOF after **client reconnect**
 - **master parity is stale** (`startup_head` behind on-disk `master` / `restart_required` from the parity gate)
 then:
 1. **Normal PR / review / merge / issue-mutation work must stop immediately.**
 2. The session **reports** the unhealthy/stale state (tool error, CTH, or operator handoff) with startup vs current head when known.
 3. Do **not** improvise: no LLM process kill/restart, no dev-worktree MCP for production mutations, no env escape hatches, no manual gate bypass.
 4. Resume only after:
   - an **operator** restores the runtime (see §2.6 for routine post-merge parity reload), **or**
   - a **controlled** promotion/rollback completes with the §2.4 promotion record, **or**
   - a controller invokes the narrow **bootstrap review path** (#557) when the defect is self-hosted and documented,
   - **and** the session re-verifies health (and master parity when applicable) before the next mutation.
 ### 2.6 Routine post-merge master-parity staleness (operator reload, not promotion)
 **Symptom:** After merges land on `master`, a long-lived stable MCP process still runs the pre-merge `startup_head`. The master-parity gate marks the server **stale** / `restart_required` and **blocks mutations**.
 **Sanctioned response (authoritative):**
 | Step | Actor | Action |
 |------|-------|--------|
 | 1 | LLM session | Mutations stop immediately when the gate reports stale. |
 | 2 | LLM session | Report the stale state (startup head, current master head, that operator reload is required). Do not retry mutations. |
 | 3 | **Operator** | Reload/restart the **stable** control MCP so it loads current `master` (supervised client/daemon reload). |
 | 4 | LLM session | Resume only after startup/current-head parity is verified (e.g. `gitea_get_runtime_context` / parity assessment shows in parity). |
 **Not a §2.4 promotion:** Catching the already-designated stable control checkout up to a newly advanced `master` is a **routine operator reload** of the stable runtime. It does **not** require the nine-field promotion ledger. Use §2.4 only when **changing which unpromoted/dev revision** becomes the stable control runtime (new source branch/PR into the stable designation).
 **Code note:** `master_parity_gate.py` may still say “restart the server” in machine-facing reason strings. That string names the **operator recovery action**, not an LLM self-service instruction. This ADR and the runbooks define the actor split.
 ## 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.
 **Not optional (issue #615 acceptance criterion 2):** operator guide and runbooks **must** cross-link this ADR (see §6). Cross-links are documentation acceptance, not deferred tooling.
 ## 6. Acceptance for this ADR
 1. Document merged under `docs/architecture/mcp-stable-control-runtime-policy-adr.md`.
 2. **Operator guide / runbooks cross-link this ADR** (`docs/wiki/Operator-Guide.md`, `docs/wiki/Runbooks.md`, `docs/llm-workflow-runbooks.md`).
 3. Issue #615 references this path.
 4. LLM/operator runbooks treat kill/restart/relaunch-from-worktree as **LLM violations**; process restart is **operator-owned**.
 5. Unhealthy runtime (including **stale master parity**) stops normal mutation work until operator restore/reload, promotion/rollback, or #557 bootstrap — then re-verify parity before mutating.
 6. Routine post-merge parity reload is documented as operator reload (§2.6), not an LLM self-restart and not a full §2.4 promotion.
 ## 7. Document history
 | Date | Change |
 |------|--------|
 | 2026-07-09 | Initial ADR: stable vs dev runtime, forbidden session actions, promotion proof fields, stop-work rule |
 | 2026-07-16 | Review 443 remediation (#615 / PR #616): mandatory cross-links; LLM vs operator restart split; routine post-merge parity staleness (§2.6); stale parity in §2.5 unhealthy triggers |
@@ -195,7 +195,7 @@ To avoid the bottleneck of relaunching/restarting the MCP server to switch betwe
  `gitea_reconcile_already_landed_pr` after ancestry proof — not for normal
  review or author workflows.
-* **Fallback (operator-owned):** If the dual-profile MCP launcher pattern is not supported or configured in the client, **do not** have the LLM relaunch or restart the client/MCP. The LLM **stops** role-switching work, reports that the correct static namespace is missing, and waits for an **operator** to configure dual namespaces or reload the client with the correct `GITEA_MCP_PROFILE` for that role. Process restart/relaunch is operator-owned under the [stable control runtime ADR](architecture/mcp-stable-control-runtime-policy-adr.md) (#615).
+* **Fallback:** If the dual-profile MCP launcher pattern is not supported or configured in the client, the LLM must relaunch or restart the client/MCP with the correct profile environment variable before claiming or working on any tasks.
 ## Setup runbook — interactive menu
@@ -1049,72 +1049,6 @@ Special states:
 Final reports must not claim a comment was posted when
 `canonical_comment_validation.allowed` is false (#496 AC14).
 ## Stale #332 review-decision lock cleanup (#594)
 #332 hard-stops a reviewer session after a terminal live review mutation
 (`approve` / `request_changes`). After #559 those locks are **durable** under
 `~/.cache/gitea-tools/session-state/review_decision_lock-<profile>.json`, so they
 can outlive the PR they protected.
 ### When cleanup is **allowed**
 Use `gitea_cleanup_stale_review_decision_lock` from a **reviewer** profile when:
 1. A durable review-decision lock has a terminal live mutation, **and**
 2. Live Gitea state shows that terminal PR is already **merged** or **closed**
   (so no same-PR merge sequence remains), **and**
 3. The active profile identity matches the lock, **and**
 4. Authenticated identity can be verified.
 Workflow:
 ```text
 # 1) Assess only (default)
 gitea_cleanup_stale_review_decision_lock(apply=false, remote=prgs, ...)
 # 2) Apply after is_moot=true and cleanup_allowed=true
 gitea_cleanup_stale_review_decision_lock(
  apply=true,
  expected_terminal_pr=<merged-or-closed-pr>,
  remote=prgs,
  ...
 )
 ```
 Successful apply:
 * clears in-memory + durable decision lock for that profile
 * returns a structured `audit` payload
 * posts an audit comment on the mooted PR when `gitea.pr.comment` is allowed
  (`post_audit_comment=true` default)
 Same-profile auto-expire: if `gitea_merge_pr` succeeds on a PR that this same
 profile just approved, the decision lock is cleared after merge. Cross-profile
 locks (reviewer vs merger) still require the cleanup tool.
 ### When cleanup is **forbidden**
 Do **not** clear when:
 * the last terminal PR is still **open** (active #332 hard-stop — continue merge
  for that approve, or stop after request_changes)
 * live PR state cannot be fetched (fail closed)
 * profile identity does not match the lock
 * identity cannot be verified
 * `expected_terminal_pr` does not match the last terminal PR
 **Never** use manual deletion of session-state files as the normal workflow.
 **Never** use cleanup to skip review of an open PR or to bypass eligibility /
 mark-ready / submit gates on active work.
 ### Related tools
 | Tool | Purpose |
 |------|---------|
 | `gitea_cleanup_stale_review_decision_lock` | Moot decision-lock cleanup (#594) |
 | `gitea_authorize_review_correction` | Operator-approved correction after a **mistaken** live review on still-active work (#211) |
 | `gitea_cleanup_post_merge_moot_lease` | Moot **lease** cleanup after merge (#515) — different object |
 ## Fail-closed behavior
 Before any mutating action the workflow verifies identity, active profile,
@@ -1200,13 +1134,10 @@ When a mutation blocks on workspace binding:
 1. Read the error — it names the **resolved workspace path**, **role
   namespace**, and **binding source** (tool arg, env var, or process root).
-2. **LLM-allowed:** pass `worktree_path` on reviewer/merger mutation tools when
+2. Reconnect or relaunch the correct namespace MCP server from the intended
-   the active `branches/` worktree differs from the MCP process root; **client
+   workspace (or set the role-specific env var before launch).
-   reconnect** after transport EOF only (no process kill).
+3. Pass `worktree_path` on reviewer/merger mutation tools when the active
-3. **Operator-owned:** if the wrong namespace process was launched, or a role-
+   branches/ worktree differs from the MCP process root.
   specific `GITEA_*_WORKTREE` must be set at process start, an **operator**
   reloads/relaunches the correct static namespace MCP. LLM sessions must not
   kill or restart MCP processes (see [stable control runtime ADR](architecture/mcp-stable-control-runtime-policy-adr.md)).
 4. **Do not** clean, reset, or discard foreign role worktrees to unblock your
   own namespace — that destroys another agent's WIP.
@@ -1216,10 +1147,9 @@ When posting a Canonical Thread Handoff after a binding blocker:
 - State which namespace was active (author / reviewer / merger / reconciler).
 - Quote the resolved workspace path and binding source from the error.
- Name the safe next action: pass `worktree_path` if that unblocks the tool, or
+- Name the safe reconnect action (relaunch MCP from `branches/...`, set
-  request an **operator** reload of the correct namespace MCP / env binding.
+  `GITEA_*_WORKTREE`, or pass `worktree_path`).
- Explicitly note that foreign worktrees must not be cleaned to unblock, and
+- Explicitly note that foreign worktrees must not be cleaned to unblock.
  that the LLM must not self-restart the MCP process.
 ## Safety notes
@@ -1230,7 +1160,6 @@ When posting a Canonical Thread Handoff after a binding blocker:
 ## Related documents
 - [`architecture/mcp-stable-control-runtime-policy-adr.md`](architecture/mcp-stable-control-runtime-policy-adr.md) — stable control runtime vs dev runtime; LLM must not kill/restart MCP; operator-owned reload and promotions; routine post-merge parity staleness (#615).
 - [`reviewer-handoff-consistency.md`](reviewer-handoff-consistency.md) — reject contradictory reviewer handoffs (#501).
 - [`issue-acceptance-gate.md`](issue-acceptance-gate.md) — controller issue-acceptance audit after PR merge (#500).
 - [`../skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md) — portable cross-project LLM workflow skill.
@@ -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.
@@ -14,7 +14,6 @@ Handbook for LLM operators and human developers using the Gitea-Tools MCP server
 4. **No self-review / no self-merge** — The authenticated Gitea user must not approve or merge a PR they authored.
 5. **Follow the gates** — Prompts express intent; MCP tools enforce safety. Never bypass gates via prompt instructions.
 6. **Global LLM Worktree Rule** — Main checkout stays on `master`/`main`/`dev`; all mutations happen under `branches/`. Prove project root, `cwd`, branch, stable main-checkout branch, and session worktree path before editing. No exceptions.
 7. **Stable control runtime** — Real Gitea mutations use only the **stable** MCP control runtime. LLM sessions must not kill, restart, or relaunch MCP processes, edit the stable checkout, or use a dev MCP for production mutations. See the policy ADR: [mcp-stable-control-runtime-policy-adr.md](../architecture/mcp-stable-control-runtime-policy-adr.md) (#615).
 ## Supported Gitea instances
@@ -31,4 +30,3 @@ Always pass `remote` explicitly on tool calls. The server default is `dadeschool
 2. `gitea_get_runtime_context` — allowed/forbidden operations for this session.
 3. `gitea_resolve_task_capability` — prove the session may perform the planned task.
 4. For reviewer work: dry-run validation (`gitea_dry_run_pr_review`) before live review mutations.
 5. Confirm master parity / runtime health when tools report stale or unhealthy control runtime — stop mutations and request an **operator** reload of the stable MCP (see [stable control runtime ADR](../architecture/mcp-stable-control-runtime-policy-adr.md)).
@@ -12,17 +12,6 @@
 4. `gitea_mark_final_review_decision` → approve via `gitea_review_pr`.
 5. `gitea_merge_pr` with pinned head SHA and `confirmation="MERGE PR <n>"`.
 ## Stable MCP control runtime (#615)
 Policy ADR: [mcp-stable-control-runtime-policy-adr.md](../architecture/mcp-stable-control-runtime-policy-adr.md).
 | Situation | Who acts | What to do |
 |-----------|----------|------------|
 | Transport EOF / missing tools | LLM | **Client reconnect only** — do not kill PIDs |
 | Wrong profile / dual-namespace missing | Operator | Configure or reload the correct static namespace(s) |
 | Master parity stale after merge | LLM stops + reports; **operator** reloads stable MCP | Resume only after parity re-verified |
 | Promote unpromoted MCP code to stable | Operator / release-manager only | Full §2.4 promotion record |
 ## Gitea Wiki sync
 The Gitea Wiki mirrors `docs/wiki/` (source of truth). After merging wiki changes:
@@ -15,7 +15,5 @@
 5. **Explicit merge confirmation** — `gitea_merge_pr` requires `confirmation="MERGE PR <n>"`.
 6. **No self-review / self-merge** — Authenticated user must differ from PR author for approve/merge.
 7. **Review decision lock** — Live review mutations require validation-phase dry-run and `gitea_mark_final_review_decision`.
-8. **Terminal review hard-stop (#332)** — After a terminal live review mutation, only same-PR merge after `approve` may continue. Durable locks (#559) must not be deleted by hand.
+8. **Redaction** — Tokens, passwords, and keychain material never appear in tool output.
-9. **Stale decision-lock cleanup (#594)** — `gitea_cleanup_stale_review_decision_lock` may clear a durable #332 lock **only** when the last terminal mutation's PR is live-state **merged or closed**, identity/profile gates pass, and apply uses a reviewer-capable profile. Open/ambiguous locks stay fail-closed. Successful cleanup records a durable audit trail.
+9. **Wiki publication (#224)** — `docs/wiki/` and the sync helper are prerequisites only. Closing a wiki issue requires live Gitea Wiki proof on the repo Wiki tab. See [Runbooks](Runbooks.md#wiki-publication-readiness-gate-224).
 10. **Redaction** — Tokens, passwords, and keychain material never appear in tool output.
 11. **Wiki publication (#224)** — `docs/wiki/` and the sync helper are prerequisites only. Closing a wiki issue requires live Gitea Wiki proof on the repo Wiki tab. See [Runbooks](Runbooks.md#wiki-publication-readiness-gate-224).
@@ -40,7 +40,6 @@ FINAL_REPORT_TASK_KINDS = frozenset({
    "issue_filing",
    "issue_selection",
    "inventory",
    "controller_close",
 })
 _TASK_KIND_ALIASES = {
@@ -52,9 +51,6 @@ _TASK_KIND_ALIASES = {
    "reconcile_already_landed": "reconcile_already_landed",
    "work-issue": "work_issue",
    "create_issue": "issue_filing",
    "close_issue": "controller_close",
    "close-issue": "controller_close",
    "controller-close": "controller_close",
 }
 _HANDOFF_ROLE_BY_TASK = {
@@ -66,7 +62,6 @@ _HANDOFF_ROLE_BY_TASK = {
    "issue_filing": "issue_filing",
    "issue_selection": None,
    "inventory": "inventory",
    "controller_close": None,
 }
 _LEGACY_WORKSPACE_MUTATIONS_RE = re.compile(
@@ -855,27 +850,6 @@ def _rule_reviewer_premerge_baseline_proof(report_text: str) -> list[dict[str, s
    )
 def _rule_reviewer_post_merge_validation(report_text: str) -> list[dict[str, str]]:
    """Block active approval on an already-merged/closed PR, and post-merge moot
    validation claimed without merged-state + merge-commit proof (#529)."""
    from post_merge_validation import assess_post_merge_validation
    result = assess_post_merge_validation(report_text)
    if not result.get("block"):
        return []
    return _findings_from_reasons(
        "reviewer.post_merge_validation",
        result.get("reasons") or [],
        field="Validation status",
        severity="block",
        safe_next_action=result.get("safe_next_action")
        or (
            "record post-merge moot validation instead of an active approval; "
            "cite PR state merged/closed and the merge commit SHA"
        ),
    )
 def _rule_reviewer_validation_status_vocabulary(
    report_text: str,
    *,
@@ -1540,7 +1514,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
        _rule_reviewer_linked_issue,
        _rule_reviewer_baseline_on_failure,
        _rule_reviewer_premerge_baseline_proof,
        _rule_reviewer_post_merge_validation,
        _rule_reviewer_validation_status_vocabulary,
        _rule_reviewer_main_checkout_baseline,
        _rule_reviewer_main_checkout_path,
@@ -1633,13 +1606,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
        *_SHARED_CANONICAL_COMMENT_RULES,
        *_SHARED_ISSUE_LOCK_RULES,
    ],
    # Controller issue closure (#529): a closure report must not bury an
    # unproven non-zero suite exit as an "expected pre-existing failure".
    # Kept intentionally narrow so a closure pre-check does not demand the
    # full reviewer/author handoff schema.
    "controller_close": [
        _rule_reviewer_premerge_baseline_proof,
    ],
 }
@@ -1,180 +0,0 @@
 """Pre-create issue content gate (#582).
 Blocks issue creation when the title/body is a *vague reference* to
 out-of-band content the LLM cannot prove it has ("the drafted issue",
 "the prepared issue", "the issue we discussed", "the previous draft")
 with no durable source pointer.
 The rule from #582: an LLM must not fabricate an issue from stale chat
 memory. When the requested issue content is a vague reference and no
 durable source pointer (existing issue/PR/comment, checked-in file, URL,
 or scratchpad path) is present, the gate fails closed and the caller must
 return BLOCKED + DIAGNOSE naming the exact vague/missing fields.
 This gate intentionally does NOT require a non-empty body: title-only
 issue creation remains allowed for callers that explicitly want it. It
 only blocks content that is a placeholder reference to something the
 current context does not contain.
 """
 from __future__ import annotations
 import re
 import unicodedata
 # Vague reference phrases that point at out-of-band draft content. Stored
 # normalized (lowercase, punctuation-stripped, whitespace-collapsed) so they
 # can be matched against normalized field text.
 VAGUE_REFERENCE_PHRASES: tuple[str, ...] = (
    "the drafted issue",
    "drafted issue",
    "the prepared issue",
    "prepared issue",
    "the issue we discussed",
    "the issue we talked about",
    "the one we discussed",
    "the previous draft",
    "previous draft",
    "the draft above",
    "the issue above",
    "as discussed",
    "as we discussed",
    "as previously discussed",
    "per our discussion",
    "per our conversation",
    "per our chat",
    "see above",
    "same as before",
    "the issue from earlier",
    "the issue i drafted",
    "the issue you drafted",
 )
 # A field that only restates a vague phrase (optionally with a leading verb
 # such as "create"/"add"/"open"/"file") is still a vague reference.
 _LEADING_VERBS = ("create", "add", "open", "file", "make", "please", "the")
 # Durable source pointers: if any of these appears, the content points at a
 # retrievable source of truth and is NOT treated as a fabricated reference.
 _DURABLE_SOURCE_REGEXES: tuple[re.Pattern[str], ...] = (
    re.compile(r"#\d+"),                                   # #402
    re.compile(r"\b(?:issue|pull|pr)\s+#?\d+", re.I),      # issue 402 / PR #17
    re.compile(r"\bcomment\s+#?\d+", re.I),                # comment 8155
    re.compile(r"https?://\S+", re.I),                     # URL
    re.compile(                                            # checked-in file
        r"[\w./-]+\.(?:py|md|json|txt|sh|ya?ml|toml|cfg|ini|rst)\b", re.I
    ),
    re.compile(r"\bscratchpad\b", re.I),                   # scratchpad path
    re.compile(r"(?:^|\s)/[\w./-]+"),                      # absolute path
 )
 def normalize_text(text: str) -> str:
    """Lowercase, punctuation-stripped, whitespace-collapsed text."""
    norm = unicodedata.normalize("NFKC", (text or "").strip().lower())
    norm = re.sub(r"[^\w\s]", " ", norm)
    norm = re.sub(r"\s+", " ", norm).strip()
    return norm
 def has_durable_source_pointer(text: str) -> bool:
    """True when the raw text references a durable, retrievable source."""
    raw = text or ""
    return any(rx.search(raw) for rx in _DURABLE_SOURCE_REGEXES)
 def find_vague_reference(text: str) -> str | None:
    """Return the vague phrase a field is dominated by, else ``None``.
    A field is a vague reference only when it contains a known vague phrase,
    carries no durable source pointer, and the phrase dominates the field
    (the field is essentially just the phrase). Long, specific bodies that
    merely contain "as discussed" in passing are not blocked.
    """
    if has_durable_source_pointer(text):
        return None
    norm = normalize_text(text)
    if not norm:
        return None
    stripped = norm
    for verb in _LEADING_VERBS:
        if stripped.startswith(verb + " "):
            stripped = stripped[len(verb) + 1:]
    for phrase in VAGUE_REFERENCE_PHRASES:
        if phrase in norm and len(stripped) <= len(phrase) + 12:
            return phrase
    return None
 def assess_issue_content(
    title: str,
    body: str = "",
    *,
    allow_incomplete: bool = False,
 ) -> dict:
    """Evaluate whether proposed issue content is durable enough to create.
    Returns a dict with ``complete`` (bool), ``missing_fields``,
    ``vague_fields``, ``reasons``, ``diagnose`` (joined reasons), and
    ``has_durable_source_pointer``. ``allow_incomplete`` is an explicit
    operator override that forces ``complete`` True.
    """
    t = (title or "").strip()
    b = (body or "").strip()
    missing_fields: list[str] = []
    vague_fields: list[str] = []
    reasons: list[str] = []
    if not t:
        missing_fields.append("title")
        reasons.append("issue title is required")
    else:
        phrase = find_vague_reference(t)
        if phrase:
            vague_fields.append("title")
            reasons.append(
                f"title is a vague reference ('{phrase}') with no durable "
                "content or source pointer"
            )
    if b:
        phrase = find_vague_reference(b)
        if phrase:
            vague_fields.append("body")
            reasons.append(
                f"body is a vague reference ('{phrase}') with no durable "
                "content or source pointer"
            )
    complete = allow_incomplete or (not missing_fields and not vague_fields)
    return {
        "complete": complete,
        "missing_fields": missing_fields,
        "vague_fields": vague_fields,
        "reasons": reasons,
        "diagnose": "; ".join(reasons),
        "has_durable_source_pointer": has_durable_source_pointer(b)
        or has_durable_source_pointer(t),
        "override_applied": bool(
            allow_incomplete and (missing_fields or vague_fields)
        ),
    }
 def pre_create_issue_content_gate(
    title: str,
    body: str = "",
    *,
    allow_incomplete: bool = False,
 ) -> dict:
    """Content gate wrapper mirroring the duplicate gate shape.
    ``performed`` is True when the content is durable enough to create (or an
    explicit override was supplied). When False, the caller must return
    BLOCKED + DIAGNOSE and must not create the issue.
    """
    assessment = assess_issue_content(
        title, body, allow_incomplete=allow_incomplete
    )
    assessment["performed"] = assessment["complete"]
    return assessment
@@ -42,38 +42,12 @@ STATUS_LABEL_SPECS: tuple[LabelSpec, ...] = (
    LabelSpec("status:wontfix", "000000", "Issue will not be fixed"),
 )
 # Durable validation-outcome labels (#529): the four canonical distinctions a
 # reviewer report can carry. These are orthogonal to the single active status:*
 # label, so they use their own prefix and are not subject to the one-status
 # invariant.
 VALIDATION_LABEL_SPECS: tuple[LabelSpec, ...] = (
    LabelSpec("validation:clean-pass", "0e8a16", "Validation was a clean pass"),
    LabelSpec(
        "validation:baseline-accepted",
        "fbca04",
        "Validation passed with a baseline-proven unrelated failure",
    ),
    LabelSpec(
        "validation:blocked",
        "b60205",
        "Validation blocked by an unresolved failure",
    ),
    LabelSpec(
        "validation:post-merge-moot",
        "5319e7",
        "Validation is post-merge moot (PR already merged/closed before review)",
    ),
 )
 CANONICAL_LABEL_SPECS: tuple[LabelSpec, ...] = (
-    TYPE_LABEL_SPECS + STATUS_LABEL_SPECS + VALIDATION_LABEL_SPECS
+    TYPE_LABEL_SPECS + STATUS_LABEL_SPECS
 )
 TYPE_LABELS: frozenset[str] = frozenset(spec.name for spec in TYPE_LABEL_SPECS)
 STATUS_LABELS: frozenset[str] = frozenset(spec.name for spec in STATUS_LABEL_SPECS)
 VALIDATION_LABELS: frozenset[str] = frozenset(
    spec.name for spec in VALIDATION_LABEL_SPECS
 )
 CANONICAL_LABELS: frozenset[str] = frozenset(
    spec.name for spec in CANONICAL_LABEL_SPECS
 )
@@ -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"
@@ -1,222 +0,0 @@
 """Canonical post-merge (moot) validation outcome and wording gate (#529).
 When a PR is already merged or closed *before* a reviewer submits their
 verdict, an ordinary "approve" is misleading: the review never gated the
 merge, so a normal active approval overstates controller confidence. The
 sanctioned outcome for that case is a **post-merge moot validation** — the
 reviewer records what validation found, but classifies it as moot rather than
 an active approval.
 This module defines the four canonical validation distinctions #529 requires
 and enforces the post-merge case:
 - ``CLEAN_PASS_OUTCOME`` — clean validation pass on an open, reviewable PR.
 - ``BASELINE_ACCEPTED_OUTCOME`` — validation pass with a baseline-proven
  unrelated failure (proof handled by :mod:`premerge_baseline_proof`).
 - ``BLOCKED_OUTCOME`` — validation blocked by an unresolved failure.
 - ``POST_MERGE_MOOT_OUTCOME`` — the PR was already merged/closed before review
  submission; validation is recorded as post-merge moot, not an active
  approval.
 The gate blocks two mistakes:
 1. Claiming an *active approval* on a PR that was already merged/closed before
   review submission (must be recorded as post-merge moot validation instead).
 2. Claiming post-merge moot validation without proof the PR is actually
   merged/closed (a merged-state field plus a merge commit SHA).
 Each outcome maps to a durable ``validation:*`` process-state label so PRs and
 issues carry the distinction (#529 acceptance criterion).
 """
 from __future__ import annotations
 import re
 from typing import Any
 CLEAN_PASS_OUTCOME = "clean validation pass"
 BASELINE_ACCEPTED_OUTCOME = "validation pass with baseline-proven unrelated failure"
 BLOCKED_OUTCOME = "validation blocked by unresolved failure"
 POST_MERGE_MOOT_OUTCOME = "post-merge moot validation"
 # Canonical validation status label a reviewer writes in the report body for the
 # post-merge case; kept in sync with validation_status_vocabulary.
 POST_MERGE_MOOT_STATUS = "post-merge moot validation"
 # Durable process-state labels (registered in issue_workflow_labels).
 LABEL_CLEAN_PASS = "validation:clean-pass"
 LABEL_BASELINE_ACCEPTED = "validation:baseline-accepted"
 LABEL_BLOCKED = "validation:blocked"
 LABEL_POST_MERGE_MOOT = "validation:post-merge-moot"
 _OUTCOME_TO_LABEL: dict[str, str] = {
    CLEAN_PASS_OUTCOME: LABEL_CLEAN_PASS,
    BASELINE_ACCEPTED_OUTCOME: LABEL_BASELINE_ACCEPTED,
    BLOCKED_OUTCOME: LABEL_BLOCKED,
    POST_MERGE_MOOT_OUTCOME: LABEL_POST_MERGE_MOOT,
 }
 # The PR was already merged/closed before the review was submitted.
 _MERGED_BEFORE_REVIEW_RE = re.compile(
    r"(?:already[- ]merged"
    r"|merged\s+before\s+review"
    r"|closed\s+before\s+review"
    r"|pr\s+(?:is|was)\s+(?:already\s+)?(?:merged|closed)"
    r"|pr\s+state\s*[:=]\s*(?:merged|closed)"
    r"|merged\s*/\s*closed)",
    re.IGNORECASE,
 )
 # An active approval verdict is being claimed.
 _ACTIVE_APPROVAL_RE = re.compile(
    r"(?:review\s+decision\s*[:=]\s*approve"
    r"|submitted\s+['\"]?approve['\"]?"
    r"|active\s+approval"
    r"|approving\s+the\s+pr"
    r"|posting\s+an?\s+approval)",
    re.IGNORECASE,
 )
 # The sanctioned post-merge moot validation wording.
 _POST_MERGE_MOOT_RE = re.compile(
    r"post[- ]merge\s+(?:moot\s+)?validation"
    r"|post[- ]merge\s+moot"
    r"|moot\s+validation",
    re.IGNORECASE,
 )
 # Proof the PR is genuinely merged/closed.
 _MERGED_STATE_PROOF_RE = re.compile(
    r"(?:pr\s+state\s*[:=]\s*(?:merged|closed)"
    r"|merged_at\s*[:=]\s*\S"
    r"|closed_at\s*[:=]\s*\S"
    r"|state\s*[:=]\s*(?:merged|closed))",
    re.IGNORECASE,
 )
 _MERGE_COMMIT_SHA_RE = re.compile(
    r"merge\s+commit(?:\s+sha)?\s*[:=]\s*[0-9a-f]{7,40}",
    re.IGNORECASE,
 )
 POST_MERGE_VALIDATION_TEMPLATE = (
    "Validation status: post-merge moot validation\n"
    "PR state: merged\n"
    "Merge commit sha: <40-hex merge commit>\n"
    "Validation finding: <what the validation run showed, for the record>\n"
    "Note: PR merged/closed before review submission; recorded as post-merge "
    "moot validation, not an active approval."
 )
 def process_state_label(outcome: str) -> str | None:
    """Return the durable ``validation:*`` label for a canonical outcome."""
    return _OUTCOME_TO_LABEL.get(outcome)
 def assess_post_merge_validation(
    report_text: str,
    *,
    pr_merged_or_closed: bool | None = None,
 ) -> dict[str, Any]:
    """Assess post-merge moot validation wording and proof (#529).
    Args:
        report_text: The reviewer final report text.
        pr_merged_or_closed: Optional structured signal that the PR is already
            merged/closed. When ``True`` it forces the merged-before-review
            path even if the report text omits the phrasing; proof fields are
            still required in the report body for durability.
    Returns a dict with ``proven``, ``block``, ``outcome``,
    ``process_state_label``, ``reasons``, ``skipped`` and ``safe_next_action``.
    Only blocks when the report either claims an active approval on a
    merged/closed PR, or claims post-merge moot validation without proof.
    """
    text = report_text or ""
    merged_signal = bool(_MERGED_BEFORE_REVIEW_RE.search(text)) or bool(
        pr_merged_or_closed
    )
    active_approval = bool(_ACTIVE_APPROVAL_RE.search(text))
    moot_wording = bool(_POST_MERGE_MOOT_RE.search(text))
    # Nothing about merged-state or moot validation — not applicable here.
    if not merged_signal and not moot_wording:
        return {
            "proven": True,
            "block": False,
            "outcome": None,
            "process_state_label": None,
            "reasons": [],
            "skipped": True,
            "safe_next_action": "",
        }
    # An active approval on a PR already merged/closed before review, without
    # the sanctioned moot wording, overstates confidence.
    if merged_signal and active_approval and not moot_wording:
        return {
            "proven": False,
            "block": True,
            "outcome": POST_MERGE_MOOT_OUTCOME,
            "process_state_label": LABEL_POST_MERGE_MOOT,
            "reasons": [
                "PR was already merged/closed before review submission; an "
                "active approval overstates confidence — record 'post-merge "
                "moot validation' instead of a normal approval"
            ],
            "skipped": False,
            "safe_next_action": (
                "classify the outcome as post-merge moot validation (not an "
                "active approval); cite PR state merged/closed and the merge "
                "commit SHA. Template:\n" + POST_MERGE_VALIDATION_TEMPLATE
            ),
        }
    # Post-merge moot validation claimed — require merged-state + commit proof.
    if moot_wording:
        reasons: list[str] = []
        if not _MERGED_STATE_PROOF_RE.search(text):
            reasons.append(
                "post-merge moot validation claimed without merged/closed state "
                "proof (state: merged/closed, or merged_at/closed_at)"
            )
        if not _MERGE_COMMIT_SHA_RE.search(text):
            reasons.append(
                "post-merge moot validation claimed without a merge commit SHA"
            )
        if reasons:
            return {
                "proven": False,
                "block": True,
                "outcome": POST_MERGE_MOOT_OUTCOME,
                "process_state_label": LABEL_POST_MERGE_MOOT,
                "reasons": reasons,
                "skipped": False,
                "safe_next_action": (
                    "prove the PR is merged/closed: cite PR state and the merge "
                    "commit SHA. Template:\n" + POST_MERGE_VALIDATION_TEMPLATE
                ),
            }
        return {
            "proven": True,
            "block": False,
            "outcome": POST_MERGE_MOOT_OUTCOME,
            "process_state_label": LABEL_POST_MERGE_MOOT,
            "reasons": [],
            "skipped": False,
            "safe_next_action": "",
        }
    # Merged signal present but no approval claim and no moot wording yet —
    # guide toward the moot outcome without blocking.
    return {
        "proven": True,
        "block": False,
        "outcome": POST_MERGE_MOOT_OUTCOME,
        "process_state_label": LABEL_POST_MERGE_MOOT,
        "reasons": [],
        "skipped": False,
        "safe_next_action": (
            "PR appears merged/closed; if submitting a verdict, record "
            "post-merge moot validation rather than an active approval"
        ),
    }
@@ -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:
@@ -523,240 +523,3 @@ def assess_lease_inventory(
        "stale_review_leases": stale,
        "reclaimable_review_leases": reclaimable,
    }
 # Canonical next-action vocabulary for reviewer lease handoff (#599).
 NEXT_ACTION_ACQUIRE = "acquire"
 NEXT_ACTION_WAIT = "wait"
 NEXT_ACTION_RESUME_EXACT_OWNER_SESSION = "resume_exact_owner_session"
 NEXT_ACTION_RELEASE_EXPIRED_LEASE = "release_expired_lease"
 NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP = "operator_authorized_cleanup"
 NEXT_ACTION_REPAIR_WORKTREE_BINDING = "repair_worktree_binding"
 _HANDOFF_CLASSIFICATIONS = frozenset({
    "no_lease",
    "own_active",
    "own_expired",
    "foreign_active",
    "foreign_reclaimable",
    "foreign_expired",
    "instructed_lease_missing_with_replacement",
    "worktree_binding_mismatch",
 })
 def _norm_path(value: str | None) -> str:
    text = (value or "").strip()
    if not text:
        return ""
    return os.path.normpath(text.rstrip("/"))
 def diagnose_reviewer_pr_lease_handoff(
    comments: list[dict],
    *,
    pr_number: int,
    current_session_id: str | None,
    current_reviewer_identity: str | None,
    proposed_worktree: str | None = None,
    env_bound_worktree: str | None = None,
    instructed_session_id: str | None = None,
    instructed_comment_id: int | None = None,
    now: datetime | None = None,
 ) -> dict[str, Any]:
    """Classify open-PR reviewer lease handoff and emit a canonical next action (#599).
    Read-only diagnosis. Never steals, releases, or adopts a foreign lease.
    Fail-closed acquisition rules for active foreign leases remain intact.
    Returns a structured diagnosis with:
    - classification
    - next_action (one of wait / resume_exact_owner_session /
      release_expired_lease / operator_authorized_cleanup /
      repair_worktree_binding / acquire)
    - active_lease identity fields when present
    - worktree_binding match result
    - instructed-lease mismatch flags
    """
    now = now or datetime.now(timezone.utc)
    session_id = (current_session_id or "").strip()
    identity = (current_reviewer_identity or "").strip()
    instructed_sid = (instructed_session_id or "").strip()
    reasons: list[str] = []
    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
    session_lease = get_session_lease()
    # Worktree binding: env-bound vs proposed vs active lease worktree.
    env_wt = _norm_path(env_bound_worktree)
    prop_wt = _norm_path(proposed_worktree)
    lease_wt = _norm_path((active or {}).get("worktree") if active else None)
    binding_mismatch = False
    binding_details: dict[str, Any] = {
        "env_bound_worktree": env_bound_worktree or None,
        "proposed_worktree": proposed_worktree or None,
        "lease_worktree": (active or {}).get("worktree") if active else None,
        "match": True,
    }
    paths = [p for p in (env_wt, prop_wt, lease_wt) if p]
    if len(paths) >= 2 and len(set(paths)) > 1:
        binding_mismatch = True
        binding_details["match"] = False
        reasons.append(
            "worktree binding mismatch: env/proposed/lease worktree paths disagree"
        )
    # Instructed lease gone while a different lease is active (PR #592-style).
    instructed_missing_with_replacement = False
    if instructed_sid or instructed_comment_id is not None:
        if not active:
            reasons.append(
                "instructed lease is gone and no active replacement lease remains"
            )
        else:
            owner = (active.get("session_id") or "").strip()
            cid = active.get("comment_id")
            sid_mismatch = bool(instructed_sid and owner and owner != instructed_sid)
            cid_mismatch = (
                instructed_comment_id is not None
                and cid is not None
                and int(cid) != int(instructed_comment_id)
            )
            if sid_mismatch or cid_mismatch:
                instructed_missing_with_replacement = True
                reasons.append(
                    "instructed lease is gone; a different active lease replaced it "
                    f"(active session_id={owner}, comment_id={cid})"
                )
    # Classification + next_action.
    classification = "no_lease"
    next_action = NEXT_ACTION_ACQUIRE
    if active:
        owner = (active.get("session_id") or "").strip()
        freshness = active.get("freshness") or classify_lease_freshness(
            active, now=now
        )
        owner_identity = (active.get("reviewer_identity") or "").strip()
        is_own = bool(session_id and owner and owner == session_id)
        # Same identity alone is NOT ownership for resume; session_id must match.
        same_identity = bool(
            identity and owner_identity and identity == owner_identity
        )
        if is_own and freshness in {"active", "stale_warning"}:
            classification = "own_active"
            next_action = NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
        elif is_own and freshness in {"reclaimable", "expired"}:
            classification = "own_expired"
            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
            reasons.append(
                f"own lease freshness is '{freshness}'; release via "
                "gitea_release_reviewer_pr_lease then re-acquire"
            )
        elif not is_own and freshness in {"active", "stale_warning"}:
            classification = "foreign_active"
            next_action = NEXT_ACTION_WAIT
            reasons.append(
                f"foreign active reviewer lease (session_id={owner}, "
                f"phase={active.get('phase')}, freshness={freshness}); "
                "do not submit; do not steal"
            )
            if same_identity:
                reasons.append(
                    "lease identity matches current reviewer but session_id differs; "
                    "resume only from the exact owner session_id or wait"
                )
        elif not is_own and freshness == "reclaimable":
            classification = "foreign_reclaimable"
            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
            reasons.append(
                f"foreign reclaimable lease (session_id={owner}); clear only via "
                "sanctioned gitea_release_reviewer_pr_lease when reclaimable"
            )
        elif not is_own and freshness == "expired":
            classification = "foreign_expired"
            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
            reasons.append(
                f"foreign expired lease (session_id={owner}); use sanctioned release"
            )
        else:
            classification = "foreign_active"
            next_action = NEXT_ACTION_WAIT
            reasons.append(
                f"unclassified active lease state (session_id={owner}, "
                f"freshness={freshness}); wait fail-closed"
            )
        if instructed_missing_with_replacement and classification.startswith(
            "foreign"
        ):
            classification = "instructed_lease_missing_with_replacement"
            # Foreign active still means wait; reclaimable still release.
            if next_action == NEXT_ACTION_WAIT:
                reasons.append(
                    "replacement foreign lease is active — wait; "
                    "operator_authorized_cleanup only with explicit operator authority"
                )
    else:
        classification = "no_lease"
        next_action = NEXT_ACTION_ACQUIRE
        reasons.append("no active reviewer lease; acquire via gitea_acquire_reviewer_pr_lease")
    # Binding mismatch is a first-class blocker before submit, but does not
    # erase foreign-lease wait/release guidance. Override next_action only when
    # the session would otherwise be free to acquire or resume (submit path).
    if binding_mismatch:
        if next_action in {
            NEXT_ACTION_ACQUIRE,
            NEXT_ACTION_RESUME_EXACT_OWNER_SESSION,
        }:
            classification = "worktree_binding_mismatch"
            next_action = NEXT_ACTION_REPAIR_WORKTREE_BINDING
        else:
            reasons.append(
                "also repair worktree binding before submit "
                f"(next_action remains {next_action})"
            )
    lease_summary = None
    if active:
        lease_summary = {
            "comment_id": active.get("comment_id"),
            "session_id": active.get("session_id"),
            "phase": active.get("phase"),
            "candidate_head": active.get("candidate_head"),
            "expires_at": active.get("expires_at"),
            "last_activity": active.get("last_activity"),
            "freshness": active.get("freshness")
            or classify_lease_freshness(active, now=now),
            "reviewer_identity": active.get("reviewer_identity"),
            "profile": active.get("profile"),
            "worktree": active.get("worktree"),
            "blocker": active.get("blocker"),
        }
    return {
        "pr_number": pr_number,
        "classification": classification,
        "next_action": next_action,
        "active_lease": lease_summary,
        "session_lease": session_lease,
        "worktree_binding": binding_details,
        "instructed_session_id": instructed_session_id,
        "instructed_comment_id": instructed_comment_id,
        "instructed_lease_missing_with_replacement": instructed_missing_with_replacement,
        "mutation_allowed": (
            next_action == NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
            and not binding_mismatch
            and bool(session_lease)
        ),
        "reasons": reasons,
        "forbidden": [
            "manual lock deletion",
            "raw API bypass",
            "mtime manipulation",
            "direct _SESSION_LEASE seeding",
            "silent foreign lease steal",
        ],
    }
@@ -55,6 +55,7 @@ def skip_python_scan_walk_root(project_root: str, walk_root: str) -> bool:
 REVIEWER_TASKS = frozenset({
    "review_pr",
    "merge_pr",
    "blind_pr_queue_review",
    "pr_queue_cleanup",
    "pr-queue-cleanup",
@@ -62,10 +63,6 @@ REVIEWER_TASKS = frozenset({
    "approve_pr",
 })
 MERGER_TASKS = frozenset({
    "merge_pr",
 })
 AUTHOR_TASKS = frozenset({
    "create_issue",
    "comment_issue",
@@ -101,7 +98,7 @@ TASK_REQUIRED_ROLE = {
    "address_pr_change_requests": "author",
    "delete_branch": "author",
    "review_pr": "reviewer",
-    "merge_pr": "merger",
+    "merge_pr": "reviewer",
    "blind_pr_queue_review": "reviewer",
    "pr_queue_cleanup": "reviewer",
    "pr-queue-cleanup": "reviewer",
@@ -131,10 +128,6 @@ WRONG_ROLE_RECONCILER_MSG = (
    "MCP namespace/profile with exact close capability."
 )
 WRONG_ROLE_MERGER_MSG = (
    "Wrong role/session for merger task. Launch merger MCP namespace."
 )
 _session_last_route: dict | None = None
@@ -250,25 +243,6 @@ def route_task_session(
        _record_route(result)
        return result
    if required_role == "merger":
        result = {
            "task_type": task_type,
            "required_role": required_role,
            "active_role": active_role_kind,
            "active_profile": active_profile,
            "route_result": ROUTE_WRONG_ROLE,
            "downstream_allowed": False,
            "reasons": [
                WRONG_ROLE_MERGER_MSG,
                "Merger tasks cannot run in author or reviewer sessions.",
            ],
            "message": WRONG_ROLE_MERGER_MSG,
            "runtime_switching_supported": runtime_switching_supported,
            "profile_switch_blocked": not runtime_switching_supported,
        }
        _record_route(result)
        return result
    if required_role == "author":
        route = ROUTE_TO_AUTHOR
        message = (
@@ -344,41 +344,6 @@ If edits are needed, make the smallest correction necessary and report the corre
 Do not silently change requested meaning.
 ## 14a. Enforced content preflight (#582)
 `gitea_create_issue` runs a **content preflight gate** before duplicate
 search or creation. It fails closed (returns `BLOCKED + DIAGNOSE`, no issue
 created) when the title/body is a *vague reference* to out-of-band draft
 content the session cannot prove it holds, and no durable source pointer is
 present.
 An LLM must never fabricate issue content from stale chat memory. If asked to
 create "the drafted issue" / "the prepared issue" / "the issue we discussed"
 / "the previous draft" without the full content in the current context or a
 durable source pointer, stop and return `BLOCKED + DIAGNOSE` naming the exact
 vague/missing fields.
 A **durable source pointer** is one of: an existing issue/PR reference
 (`#582`), a comment id (`comment 8155`), a checked-in file path
 (`task_capability_map.py`), a URL, or a scratchpad path. When a pointer is
 present, retrieve the real content from it before creating.
 The gate does **not** require a non-empty body; explicit title-only creation
 stays allowed. It only blocks placeholder references. An operator may bypass
 it with `allow_incomplete_content=True` (report the override).
 **Invalid prompts (must block):**
 * "Create the drafted issue."
 * "File the prepared issue we discussed."
 * "Open the previous draft."
 **Valid prompts (proceed):**
 * Full title + body + acceptance criteria supplied inline.
 * "Create the issue drafted in `#582`." (durable pointer → fetch and use it)
 * "Create the issue from `scratchpad/issue-draft.md`." (durable file pointer)
 ## 15. Labels, assignees, and metadata
 Apply labels, assignees, milestones, or project fields only if:
@@ -826,75 +826,6 @@ The final report must identify:
 * whether same-PR merge continuation was allowed
 * whether the run stopped as required
 ## 26A-1. Stale durable review-decision lock cleanup (#594)
 After #559, the review-decision lock is durable under
 `~/.cache/gitea-tools/session-state/` (for example
 `review_decision_lock-prgs-reviewer.json`). That durability can outlive the work
 it protects: a terminal `approve` / `request_changes` on a PR that is already
 **merged or closed** still hard-stops later unrelated reviews under #332.
 **Do not** delete session-state files by hand. Use the canonical MCP path.
 ### When cleanup is allowed
 Use `gitea_cleanup_stale_review_decision_lock` when:
 1. `gitea_mark_final_review_decision` / live review is blocked by
   `terminal review mutation already consumed ... (#332)`.
 2. Live Gitea state for the **last terminal mutation's PR** is **merged** or
   **closed** (no same-PR merge sequence remains).
 3. Active profile identity matches the durable lock (reviewer profile with
   `gitea.pr.review` for `apply=true`).
 4. Optional pin: pass `expected_terminal_pr` to the prior terminal PR number
   (for example `586` when resuming after a merged #586 lock).
 Recommended sequence:
 ```text
 gitea_whoami
 gitea_resolve_task_capability(task="cleanup_stale_review_decision_lock")
 gitea_cleanup_stale_review_decision_lock(apply=false, remote=..., org=..., repo=...)
 # inspect is_moot / cleanup_allowed / last_terminal_pr
 gitea_cleanup_stale_review_decision_lock(
    apply=true,
    expected_terminal_pr=<last_terminal_pr>,
    remote=..., org=..., repo=...,
 )
 gitea_resolve_task_capability(task="review_pr")
 gitea_load_review_workflow()
 # continue formal mark + submit for the *new* PR
 ```
 Successful `apply=true` clears memory + durable store and returns an audit
 record (and posts a durable PR comment on the terminal PR when
 `post_audit_comment` is true and comment permission allows).
 Same-profile auto-expire: after `gitea_merge_pr` succeeds for the PR that this
 profile just approved, the decision lock for that profile is cleared
 automatically. Cross-profile locks (for example reviewer approve, merger merge)
 still require `gitea_cleanup_stale_review_decision_lock`.
 ### When cleanup is forbidden
 Refuse / fail-closed — keep #332 hard-stop — when:
 * the last terminal PR is still **open** (active review or merge sequence may
  still apply)
 * live PR state cannot be fetched (ambiguous)
 * no lock or no terminal live mutation exists
 * profile identity does not match the lock
 * apply requested without authenticated identity or `gitea.pr.review`
 * `expected_terminal_pr` does not match the last terminal PR
 Do **not** use cleanup to:
 * bypass #332 on an open PR
 * replace `gitea_authorize_review_correction` for a mistaken review on a still
  active PR (#211)
 * auto-approve or auto-merge any PR
 * justify `rm` of session-state files
 ## 26B. Per-PR reviewer lease (#407)
 Parallel reviewer sessions are allowed only when each session holds a distinct,
@@ -1112,8 +1043,6 @@ Blocked handoffs must say to rerun the full workflow after the blocker clears.
 If the blocker is a terminal review mutation already consumed in the current run, the handoff must say that the next run must start from the beginning and must not reuse stale ready/approved state.
 If the referenced terminal PR is already **merged or closed** on live Gitea, the durable #332 decision lock is **moot**. Do **not** delete session-state files by hand. Use `gitea_cleanup_stale_review_decision_lock` (assess with `apply=false`, then `apply=true` with `expected_terminal_pr` set) from the reviewer profile after identity/profile checks (#594). Cleanup is **forbidden** while that PR is still open.
 ## 30. Final report must be precise
 Include:
@@ -1,252 +0,0 @@
 """Stale / moot #332 review-decision lock detection and cleanup policy (#594).
 #332 correctly hard-stops a reviewer session after a terminal live review
 mutation. After #559 those locks are durable on disk, so they can outlive the
 work they protect: when the referenced PR is already merged or closed, no
 same-PR merge sequence remains, yet the durable ledger still blocks unrelated
 new terminal reviews.
 This module is pure policy (no Gitea I/O). The MCP tool
 ``gitea_cleanup_stale_review_decision_lock`` fetches live PR state, calls these
 helpers, and only then clears durable state when cleanup is allowed.
 Manual deletion of ``~/.cache/gitea-tools/session-state/*`` is never the
 canonical path.
 """
 from __future__ import annotations
 from datetime import datetime, timezone
 from typing import Any
 # Keep in sync with gitea_mcp_server._TERMINAL_REVIEW_ACTIONS.
 TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
 def last_terminal_mutation(lock: dict | None) -> dict | None:
    """Return the last terminal live mutation on *lock*, or None."""
    if not lock:
        return None
    terminals = [
        m
        for m in (lock.get("live_mutations") or [])
        if isinstance(m, dict) and m.get("action") in TERMINAL_REVIEW_ACTIONS
    ]
    return terminals[-1] if terminals else None
 def classify_pr_live_state(pr_live: dict | None) -> dict[str, Any]:
    """Normalize a Gitea PR payload into merge/closed flags."""
    pr = pr_live or {}
    merged = bool(pr.get("merged") or pr.get("merged_at"))
    state = (pr.get("state") or "").strip().lower() or None
    closed = state == "closed"
    return {
        "pr_state": state,
        "pr_merged": merged,
        "pr_merged_or_closed": merged or closed,
        "merge_commit_sha": pr.get("merge_commit_sha")
        or pr.get("merged_commit_sha"),
    }
 def lock_summary(lock: dict | None) -> dict[str, Any] | None:
    """LLM-safe summary of a decision lock (no secrets)."""
    if lock is None:
        return None
    last = last_terminal_mutation(lock)
    mutations = list(lock.get("live_mutations") or [])
    return {
        "task": lock.get("task"),
        "remote": lock.get("remote"),
        "org": lock.get("org") or lock.get("ready_org"),
        "repo": lock.get("repo") or lock.get("ready_repo"),
        "profile_identity": lock.get("profile_identity")
        or lock.get("session_profile_lock")
        or lock.get("session_profile"),
        "session_profile": lock.get("session_profile"),
        "ready_pr_number": lock.get("ready_pr_number"),
        "ready_action": lock.get("ready_action"),
        "live_mutations_count": len(mutations),
        "last_terminal": (
            {
                "pr_number": last.get("pr_number"),
                "action": last.get("action"),
                "review_id": last.get("review_id"),
            }
            if last
            else None
        ),
        "correction_authorized": bool(lock.get("correction_authorized")),
        "updated_at": lock.get("updated_at") or lock.get("recorded_at"),
    }
 def assess_stale_review_decision_lock(
    lock: dict | None,
    *,
    pr_live: dict | None = None,
    pr_lookup_error: str | None = None,
    active_profile_identity: str | None = None,
 ) -> dict[str, Any]:
    """Assess whether a durable review-decision lock is stale/moot (#594).
    *pr_live* must be the live Gitea payload for the **last terminal
    mutation's PR**. When no lock / no terminal mutation exists, cleanup is
    not applicable. When the PR is still open (or lookup fails), cleanup is
    forbidden and #332 hard-stop remains in force.
    """
    summary = lock_summary(lock)
    result: dict[str, Any] = {
        "has_lock": lock is not None,
        "lock_summary": summary,
        "last_terminal_pr": None,
        "last_terminal_action": None,
        "is_moot": False,
        "cleanup_allowed": False,
        "profile_match": True,
        "pr_state": None,
        "pr_merged": False,
        "pr_merged_or_closed": False,
        "merge_commit_sha": None,
        "reasons": [],
        "recovery_tool": "gitea_cleanup_stale_review_decision_lock",
    }
    if lock is None:
        result["reasons"].append("no review decision lock present")
        return result
    # Profile identity gate (caller may re-check with env; pure assess still
    # reports mismatch when active identity is supplied).
    stored = (
        (lock.get("profile_identity") or lock.get("session_profile_lock") or "")
        .strip()
    )
    active = (active_profile_identity or "").strip()
    if active and stored and active != stored:
        result["profile_match"] = False
        result["reasons"].append(
            "review decision lock profile identity does not match active "
            f"profile '{active}' (fail closed, #594)"
        )
        return result
    last = last_terminal_mutation(lock)
    if last is None:
        result["reasons"].append(
            "review decision lock has no terminal live mutations; nothing to clear"
        )
        return result
    pr_number = last.get("pr_number")
    action = last.get("action")
    result["last_terminal_pr"] = pr_number
    result["last_terminal_action"] = action
    if pr_number is None:
        result["reasons"].append(
            "last terminal mutation missing pr_number; refuse cleanup (fail closed)"
        )
        return result
    if pr_lookup_error:
        result["reasons"].append(
            f"could not load live state for PR #{pr_number}: {pr_lookup_error} "
            "(fail closed, #594)"
        )
        return result
    if pr_live is None:
        result["reasons"].append(
            f"live PR state required for terminal PR #{pr_number} before cleanup "
            "(fail closed, #594)"
        )
        return result
    live = classify_pr_live_state(pr_live)
    result.update(
        {
            "pr_state": live["pr_state"],
            "pr_merged": live["pr_merged"],
            "pr_merged_or_closed": live["pr_merged_or_closed"],
            "merge_commit_sha": live["merge_commit_sha"],
        }
    )
    if not live["pr_merged_or_closed"]:
        result["reasons"].append(
            f"terminal review mutation on open PR #{pr_number} is still active "
            f"({action}); #332 hard-stop remains — cleanup forbidden (#594)"
        )
        return result
    # Merged or closed: lock is moot. Same-PR merge continuation is impossible.
    result["is_moot"] = True
    result["cleanup_allowed"] = True
    result["reasons"].append(
        f"terminal mutation ({action} on PR #{pr_number}) is moot: PR is "
        f"{'merged' if live['pr_merged'] else 'closed'}; canonical cleanup allowed (#594)"
    )
    return result
 def build_cleanup_audit_record(
    *,
    assessment: dict[str, Any],
    actor_username: str | None,
    profile_name: str | None,
    applied: bool,
 ) -> dict[str, Any]:
    """Structured durable audit payload for a cleanup attempt."""
    last = (assessment.get("lock_summary") or {}).get("last_terminal") or {}
    return {
        "event": "stale_review_decision_lock_cleanup",
        "issue_ref": "#594",
        "applied": bool(applied),
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "actor_username": actor_username,
        "profile_name": profile_name,
        "last_terminal_pr": assessment.get("last_terminal_pr") or last.get("pr_number"),
        "last_terminal_action": assessment.get("last_terminal_action")
        or last.get("action"),
        "pr_state": assessment.get("pr_state"),
        "pr_merged": assessment.get("pr_merged"),
        "pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
        "merge_commit_sha": assessment.get("merge_commit_sha"),
        "prior_live_mutations_count": (
            (assessment.get("lock_summary") or {}).get("live_mutations_count")
        ),
        "prior_profile_identity": (
            (assessment.get("lock_summary") or {}).get("profile_identity")
        ),
        "cleanup_allowed": assessment.get("cleanup_allowed"),
        "is_moot": assessment.get("is_moot"),
        "reasons": list(assessment.get("reasons") or []),
    }
 def format_cleanup_audit_comment(audit: dict[str, Any]) -> str:
    """Markdown body for a durable Gitea audit comment after cleanup."""
    status = "APPLIED" if audit.get("applied") else "ASSESSED (not applied)"
    lines = [
        "## Stale #332 review-decision lock cleanup (#594)",
        "",
        f"Status: **{status}**",
        "",
        f"- actor: `{audit.get('actor_username')}`",
        f"- profile: `{audit.get('profile_name')}`",
        f"- timestamp: `{audit.get('timestamp')}`",
        f"- last terminal: `{audit.get('last_terminal_action')}` on "
        f"PR #{audit.get('last_terminal_pr')}",
        f"- PR state: `{audit.get('pr_state')}` "
        f"(merged={audit.get('pr_merged')})",
        f"- merge_commit_sha: `{audit.get('merge_commit_sha')}`",
        f"- prior live_mutations_count: "
        f"`{audit.get('prior_live_mutations_count')}`",
        f"- prior profile_identity: `{audit.get('prior_profile_identity')}`",
        "",
        "Manual deletion of session-state files is **not** the workflow.",
        "This path only clears a lock when the referenced PR is merged/closed.",
    ]
    return "\n".join(lines)
@@ -96,17 +96,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
        "permission": "gitea.pr.approve",
        "role": "reviewer",
    },
    # #594: clear durable #332 decision lock only when last terminal PR is
    # already merged/closed (moot). Apply path requires reviewer review
    # permission; assessment itself uses gitea.read inside the tool.
    "cleanup_stale_review_decision_lock": {
        "permission": "gitea.pr.review",
        "role": "reviewer",
    },
    "gitea_cleanup_stale_review_decision_lock": {
        "permission": "gitea.pr.review",
        "role": "reviewer",
    },
    "delete_branch": {
        "permission": "gitea.branch.delete",
        "role": "author",
@@ -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")
@@ -18,13 +18,13 @@ def _reset_mutation_authority(monkeypatch):
    no-op: individual tests that need a specific authority state set it up
    explicitly.
-    Session-state isolation (#559 / #590 / #594): many tests use
+    Session-state isolation (#559 / #590): many tests use
    ``patch.dict(os.environ, ..., clear=True)``, which drops
    ``GITEA_MCP_SESSION_STATE_DIR``. Relying only on the env var therefore
    re-exposes the operator host cache
    (``~/.cache/gitea-tools/session-state``) and its review-mutation ledger.
    Pin ``default_state_dir`` / ``DEFAULT_STATE_DIR`` to a per-test temp dir
-    so durable load/save never touches host state even after env clears.
+    so durable load/save never touches host state, even after env clears.
    """
    monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
    # Isolate durable session-state files so tests never share host cache (#559).
@@ -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)
@@ -75,14 +74,6 @@ def _reset_mutation_authority(monkeypatch):
    monkeypatch.setattr(mcp_server, "_preflight_capability_baseline_porcelain", None)
    monkeypatch.setattr(mcp_server, "_preflight_resolved_task", None)
    monkeypatch.setattr(mcp_server, "_preflight_reviewer_violation_files", [])
    # Unit tests must not depend on live host MCP process health. Tests that
    # use patch.dict(..., clear=True) drop PYTEST_CURRENT_TEST, which would
    # otherwise re-enable the stale-runtime probe against operator daemons.
    monkeypatch.setattr(
        mcp_server,
        "_check_mcp_runtimes_diagnostics",
        lambda *args, **kwargs: [],
    )
    try:
        import review_workflow_load
        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
@@ -1,113 +0,0 @@
 """Controller-closure baseline-proof gate tests (#529, criterion 6)."""
 import unittest
 from controller_closure_baseline_proof import (
    assess_controller_closure_baseline_proof,
 )
 from premerge_baseline_proof import (
    CLEAN_PASS,
    PREMERGE_BASELINE_PROVEN_FAILURE,
    UNRESOLVED_REGRESSION_RISK,
 )
 from final_report_validator import assess_final_report_validator
 # Complete pre-merge proof fields for a baseline claim (see #533).
 _PROOF_FIELDS = (
    "Pre-merge base commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
    "Tested commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
    "Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
    "Exit status: 1\n"
    "Failure signature: AssertionError: None is not true\n"
 )
 class TestControllerClosureBaselineProof(unittest.TestCase):
    def test_empty_report_skipped_and_allowed(self):
        result = assess_controller_closure_baseline_proof("")
        self.assertFalse(result["block"])
        self.assertTrue(result["skipped"])
        self.assertEqual(result["label"], CLEAN_PASS)
    def test_none_report_allowed(self):
        result = assess_controller_closure_baseline_proof(None)
        self.assertFalse(result["block"])
        self.assertTrue(result["skipped"])
    def test_clean_pass_closure_allowed(self):
        result = assess_controller_closure_baseline_proof(
            "Closing #529. Validation: full suite passed, exit status 0. Clean pass."
        )
        self.assertFalse(result["block"])
        self.assertEqual(result["label"], CLEAN_PASS)
    def test_expected_preexisting_failure_without_proof_blocked(self):
        report = (
            "Closing #529. Full suite: 1 failed. This is an expected "
            "pre-existing failure, safe to close."
        )
        result = assess_controller_closure_baseline_proof(report)
        self.assertTrue(result["block"])
        self.assertFalse(result["proven"])
        self.assertEqual(result["label"], UNRESOLVED_REGRESSION_RISK)
        self.assertTrue(result["reasons"])
        self.assertTrue(result["safe_next_action"])
    def test_current_master_reproduction_only_blocked(self):
        report = (
            "Closing #529. Full suite: 1 failed. Pre-existing baseline failure — "
            "reproduced on current master after merge.\n"
            "Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
            "Exit status: 1\n"
            "Failure signature: AssertionError: None is not true\n"
        )
        result = assess_controller_closure_baseline_proof(report)
        self.assertTrue(result["block"])
    def test_proven_baseline_closure_allowed(self):
        report = (
            "Closing #529. Full suite: 1 failed, an expected pre-existing "
            "baseline failure proven on the PR pre-merge base commit.\n"
            + _PROOF_FIELDS
        )
        result = assess_controller_closure_baseline_proof(report)
        self.assertFalse(result["block"])
        self.assertEqual(result["label"], PREMERGE_BASELINE_PROVEN_FAILURE)
 class TestControllerCloseTaskKind(unittest.TestCase):
    """The composable validator must cover the controller_close task kind."""
    def test_controller_close_blocks_unproven_baseline(self):
        report = (
            "Full suite: 1 failed. Expected pre-existing failure, closing the "
            "tracking issue."
        )
        result = assess_final_report_validator(report, "controller_close")
        self.assertTrue(result["blocked"])
        self.assertTrue(
            any(
                "premerge_baseline_proof" in f["rule_id"]
                for f in result["findings"]
            )
        )
    def test_controller_close_alias_close_issue(self):
        report = (
            "Full suite: 1 failed. Expected pre-existing failure, closing the "
            "tracking issue."
        )
        result = assess_final_report_validator(report, "close_issue")
        self.assertTrue(result["blocked"])
    def test_controller_close_allows_proven_baseline(self):
        report = (
            "Full suite: 1 failed, expected pre-existing baseline failure proven "
            "on the PR pre-merge base commit.\n" + _PROOF_FIELDS
        )
        result = assess_final_report_validator(report, "controller_close")
        self.assertFalse(result["blocked"])
 if __name__ == "__main__":
    unittest.main()
@@ -1,195 +0,0 @@
 """Tests for the pre-create issue content gate (Issue #582)."""
 import sys
 import unittest
 from unittest.mock import patch
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 from issue_content_gate import (  # noqa: E402
    assess_issue_content,
    find_vague_reference,
    has_durable_source_pointer,
    normalize_text,
    pre_create_issue_content_gate,
 )
 from mcp_server import gitea_create_issue  # noqa: E402
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 ISSUE_WRITE_ENV = {
    "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
 }
 class TestNormalizeAndPointers(unittest.TestCase):
    def test_normalize_strips_punctuation_and_case(self):
        self.assertEqual(normalize_text("  The Drafted-Issue!  "), "the drafted issue")
    def test_issue_ref_is_durable_pointer(self):
        self.assertTrue(has_durable_source_pointer("See #402 for the spec"))
    def test_file_path_is_durable_pointer(self):
        self.assertTrue(has_durable_source_pointer("Draft in task_capability_map.py"))
    def test_scratchpad_is_durable_pointer(self):
        self.assertTrue(has_durable_source_pointer("Full text in scratchpad/draft.md"))
    def test_url_is_durable_pointer(self):
        self.assertTrue(
            has_durable_source_pointer("https://gitea.prgs.cc/issues/1 has it")
        )
    def test_plain_text_has_no_durable_pointer(self):
        self.assertFalse(has_durable_source_pointer("just some words here"))
 class TestFindVagueReference(unittest.TestCase):
    def test_bare_vague_phrases_are_flagged(self):
        for text in (
            "the drafted issue",
            "The prepared issue",
            "the issue we discussed",
            "the previous draft",
            "Create the drafted issue",
            "please file the prepared issue",
        ):
            with self.subTest(text=text):
                self.assertIsNotNone(find_vague_reference(text))
    def test_durable_pointer_defeats_vague_reference(self):
        # Same vague phrase but with a durable pointer -> not vague.
        self.assertIsNone(find_vague_reference("the drafted issue in #402"))
        self.assertIsNone(
            find_vague_reference("the prepared issue: see scratchpad/draft.md")
        )
    def test_long_specific_body_with_incidental_phrase_is_not_vague(self):
        body = (
            "As discussed, the create_issue tool must reject vague references. "
            "Add a preflight check that lists the exact missing fields and "
            "returns BLOCKED + DIAGNOSE with concrete acceptance criteria."
        )
        self.assertIsNone(find_vague_reference(body))
    def test_concrete_short_body_is_not_vague(self):
        self.assertIsNone(find_vague_reference("Fix null deref in auth handler"))
    def test_empty_is_not_vague(self):
        self.assertIsNone(find_vague_reference(""))
 class TestAssessIssueContent(unittest.TestCase):
    def test_missing_title_is_incomplete(self):
        result = assess_issue_content("", "some body")
        self.assertFalse(result["complete"])
        self.assertIn("title", result["missing_fields"])
    def test_vague_body_is_incomplete_and_diagnosed(self):
        result = assess_issue_content("Add the thing", "the drafted issue")
        self.assertFalse(result["complete"])
        self.assertIn("body", result["vague_fields"])
        self.assertIn("vague reference", result["diagnose"])
    def test_vague_title_is_incomplete(self):
        result = assess_issue_content("the prepared issue", "")
        self.assertFalse(result["complete"])
        self.assertIn("title", result["vague_fields"])
    def test_complete_content_passes(self):
        result = assess_issue_content(
            "Block vague issue creation",
            "Add a preflight gate that rejects placeholder references.",
        )
        self.assertTrue(result["complete"])
        self.assertEqual(result["missing_fields"], [])
        self.assertEqual(result["vague_fields"], [])
    def test_title_only_is_allowed(self):
        # Empty body must NOT block: title-only creation stays allowed.
        result = assess_issue_content("A concrete specific title", "")
        self.assertTrue(result["complete"])
    def test_durable_pointer_body_passes(self):
        result = assess_issue_content(
            "Implement the drafted change",
            "the drafted issue is fully specified in #582",
        )
        self.assertTrue(result["complete"])
    def test_allow_incomplete_override(self):
        result = assess_issue_content(
            "the drafted issue", "the drafted issue", allow_incomplete=True
        )
        self.assertTrue(result["complete"])
        self.assertTrue(result["override_applied"])
    def test_gate_wrapper_sets_performed(self):
        blocked = pre_create_issue_content_gate("Add x", "the drafted issue")
        self.assertFalse(blocked["performed"])
        ok = pre_create_issue_content_gate("Add x", "concrete actionable body")
        self.assertTrue(ok["performed"])
 class TestCreateIssueContentMCPGate(unittest.TestCase):
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_blocks_vague_reference_without_creating(
        self, _auth, mock_get_all, mock_api, mock_role_check
    ):
        mock_role_check.return_value = (True, [])
        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
            result = gitea_create_issue(
                title="Create the drafted issue",
                body="the drafted issue",
                remote="prgs",
            )
        self.assertFalse(result["performed"])
        self.assertIn("content_gate", result)
        self.assertIn("body", result["content_gate"]["vague_fields"])
        self.assertTrue(any("BLOCKED + DIAGNOSE" in r for r in result["reasons"]))
        mock_api.assert_not_called()
        mock_get_all.assert_not_called()
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_creates_when_content_is_concrete(
        self, _auth, _get_all, mock_api, mock_role_check
    ):
        mock_role_check.return_value = (True, [])
        mock_api.return_value = {"number": 7, "html_url": "https://x/issues/7"}
        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
            result = gitea_create_issue(
                title="Block vague issue creation",
                body="Add a preflight gate rejecting placeholder references.",
                remote="prgs",
            )
        self.assertEqual(result["number"], 7)
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_override_creates_despite_vague_content(
        self, _auth, _get_all, mock_api, mock_role_check
    ):
        mock_role_check.return_value = (True, [])
        mock_api.return_value = {"number": 8, "html_url": "https://x/issues/8"}
        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
            result = gitea_create_issue(
                title="the drafted issue",
                body="the drafted issue",
                remote="prgs",
                allow_incomplete_content=True,
            )
        self.assertEqual(result["number"], 8)
 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()
@@ -774,8 +774,8 @@ class TestMergePR(unittest.TestCase):
    def setUp(self):
        import reviewer_pr_lease
-        # Clean ledger each merge test so residual/host durable #332 state
+        # Start each merge test with a clean review-mutation ledger (#590).
-        # cannot short-circuit eligibility gates (#590 / #594).
+        # Host durable state can otherwise leak in when tests clear os.environ.
        mcp_server._save_review_decision_lock(None)
        mcp_server.gitea_load_review_workflow()
        self._lease_patch = _install_owned_reviewer_lease(8)
@@ -2733,56 +2733,6 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
        self.assertTrue(res["success"])
        self.assertEqual(res["cleanup_status"].get(1), "not present")
    def test_close_issue_blocks_unproven_expected_preexisting_failure(self):
        # #529: a closure report calling a non-zero suite exit an "expected
        # pre-existing failure" without pre-merge proof must fail closed and
        # never PATCH the issue state.
        patched = []
        def api_side_effect(method, url, auth, payload=None):
            if method == "PATCH":
                patched.append(url)
            return {}
        self.mock_api.side_effect = api_side_effect
        res = gitea_close_issue(
            issue_number=1,
            closure_report=(
                "Full suite: 1 failed. Expected pre-existing failure, safe to close."
            ),
        )
        self.assertFalse(res["success"])
        self.assertTrue(res.get("blocked"))
        self.assertTrue(res.get("reasons"))
        self.assertFalse(
            patched, "must not PATCH issue state when the closure gate blocks"
        )
    def test_close_issue_allows_proven_baseline_closure(self):
        def api_side_effect(method, url, auth, payload=None):
            if method == "PATCH" and "issues/1" in url:
                return {"state": "closed"}
            if method == "GET" and "labels" in url and "issues" not in url:
                return [{"name": "bug", "id": 2}]
            if method == "GET" and "issues/1" in url:
                return {"labels": [{"name": "bug"}]}
            return {}
        self.mock_api.side_effect = api_side_effect
        res = gitea_close_issue(
            issue_number=1,
            closure_report=(
                "Full suite: 1 failed, expected pre-existing baseline failure "
                "proven on the PR pre-merge base commit.\n"
                "Pre-merge base commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
                "Tested commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
                "Command: venv/bin/python -m pytest -q\n"
                "Exit status: 1\n"
                "Failure signature: AssertionError: None is not true\n"
            ),
        )
        self.assertTrue(res["success"])
    def test_merge_pr_with_closes_removes_label(self):
        import reviewer_pr_lease
@@ -67,66 +67,5 @@ class TestMcpStaleRuntime(unittest.TestCase):
        # But does NOT contain missing author profile error
        self.assertFalse(any("None of the matching profiles for task" in r for r in reasons_author))
    @patch("subprocess.run")
    @patch("os.path.getmtime")
    @patch("os.path.exists")
    @patch("os.getpid")
    @patch.dict("os.environ", {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"})
    def test_git_head_change_triggers_staleness(self, mock_getpid, mock_exists, mock_getmtime, mock_run):
        # Set boot SHA to FAKE1 and current to FAKE2
        gitea_mcp_server._process_boot_head_sha = "FAKE1"
        mock_getpid.return_value = 12345
        mock_exists.return_value = True
        # Code time is in the past (fresh process chronologically)
        code_time = datetime(2026, 7, 8, 11, 0, 0)
        mock_getmtime.return_value = code_time.timestamp()
        ps_output = (
            "  PID LSTART COMMAND\n"
            "12345 Wed Jul  8 12:00:00 2026 /path/to/python mcp_server.py\n"
        )
        mock_run_ps = MagicMock()
        mock_run_ps.stdout = ps_output
        mock_run_env = MagicMock()
        mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author"
        mock_run_git = MagicMock()
        mock_run_git.stdout = "FAKE2" # different SHA
        def side_effect(args, **kwargs):
            if args[0] == "ps" and "eww" in args:
                return mock_run_env
            elif args[0] == "ps":
                return mock_run_ps
            elif args[0] == "git" and "rev-parse" in args:
                return mock_run_git
            raise ValueError(f"Unexpected: {args}")
        mock_run.side_effect = side_effect
        # Stale even though process started AFTER code mtime, because git HEAD changed
        reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
        self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons))
    @patch("threading.Thread")
    @patch("os.utime")
    @patch("os.path.exists")
    @patch.dict("os.environ", {"MCP_CONFIG_PATH": "/tmp/mcp_config.json"})
    def test_auto_restart_trigger_touches_and_spawns(self, mock_exists, mock_utime, mock_thread):
        mock_exists.return_value = True
        gitea_mcp_server._restart_triggered = False
        # Ensure we are not skipped in test mode for testing purposes
        with patch("gitea_mcp_server._preflight_in_test_mode", return_value=False):
            gitea_mcp_server._trigger_mcp_auto_restart()
        mock_utime.assert_called_once_with("/tmp/mcp_config.json", None)
        mock_thread.assert_called_once()
        self.assertTrue(gitea_mcp_server._restart_triggered)
 if __name__ == "__main__":
    unittest.main()
@@ -1,119 +0,0 @@
 """Post-merge moot validation wording + label tests (#529, criteria 1/4/5)."""
 import unittest
 from post_merge_validation import (
    BASELINE_ACCEPTED_OUTCOME,
    BLOCKED_OUTCOME,
    CLEAN_PASS_OUTCOME,
    LABEL_BASELINE_ACCEPTED,
    LABEL_BLOCKED,
    LABEL_CLEAN_PASS,
    LABEL_POST_MERGE_MOOT,
    POST_MERGE_MOOT_OUTCOME,
    assess_post_merge_validation,
    process_state_label,
 )
 from final_report_validator import assess_final_report_validator
 from issue_workflow_labels import CANONICAL_LABELS, VALIDATION_LABELS
 _MOOT_PROOF = (
    "Validation status: post-merge moot validation\n"
    "PR state: merged\n"
    "Merge commit sha: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
    "Validation finding: full suite passed, for the record.\n"
 )
 class TestPostMergeValidation(unittest.TestCase):
    def test_not_applicable_when_no_merged_or_moot_signal(self):
        result = assess_post_merge_validation(
            "Review decision: approve. Validation: full suite passed."
        )
        self.assertFalse(result["block"])
        self.assertTrue(result["skipped"])
    def test_active_approval_on_merged_pr_blocked(self):
        report = (
            "PR is already merged. Review decision: approve. "
            "Validation: full suite passed."
        )
        result = assess_post_merge_validation(report)
        self.assertTrue(result["block"])
        self.assertEqual(result["outcome"], POST_MERGE_MOOT_OUTCOME)
        self.assertEqual(result["process_state_label"], LABEL_POST_MERGE_MOOT)
    def test_active_approval_on_merged_pr_via_structured_signal_blocked(self):
        report = "Review decision: approve. Validation: full suite passed."
        result = assess_post_merge_validation(report, pr_merged_or_closed=True)
        self.assertTrue(result["block"])
    def test_moot_wording_without_proof_blocked(self):
        report = "Recording post-merge moot validation for this PR."
        result = assess_post_merge_validation(report)
        self.assertTrue(result["block"])
        self.assertTrue(result["reasons"])
    def test_moot_wording_with_full_proof_allowed(self):
        result = assess_post_merge_validation(_MOOT_PROOF)
        self.assertFalse(result["block"])
        self.assertEqual(result["outcome"], POST_MERGE_MOOT_OUTCOME)
    def test_merged_signal_only_guides_without_blocking(self):
        result = assess_post_merge_validation(
            "PR was already merged before review; recording findings."
        )
        self.assertFalse(result["block"])
        self.assertTrue(result["safe_next_action"])
    def test_process_state_label_mapping(self):
        self.assertEqual(process_state_label(CLEAN_PASS_OUTCOME), LABEL_CLEAN_PASS)
        self.assertEqual(
            process_state_label(BASELINE_ACCEPTED_OUTCOME), LABEL_BASELINE_ACCEPTED
        )
        self.assertEqual(process_state_label(BLOCKED_OUTCOME), LABEL_BLOCKED)
        self.assertEqual(
            process_state_label(POST_MERGE_MOOT_OUTCOME), LABEL_POST_MERGE_MOOT
        )
        self.assertIsNone(process_state_label("nonsense"))
 class TestValidationLabelsRegistered(unittest.TestCase):
    def test_validation_labels_are_canonical(self):
        for label in (
            LABEL_CLEAN_PASS,
            LABEL_BASELINE_ACCEPTED,
            LABEL_BLOCKED,
            LABEL_POST_MERGE_MOOT,
        ):
            self.assertIn(label, VALIDATION_LABELS)
            self.assertIn(label, CANONICAL_LABELS)
 class TestPostMergeValidationWiredIntoReview(unittest.TestCase):
    def test_review_pr_blocks_active_approval_on_merged_pr(self):
        report = (
            "PR is already merged. Review decision: approve. "
            "Validation: full suite passed."
        )
        result = assess_final_report_validator(report, "review_pr")
        self.assertTrue(result["blocked"])
        self.assertTrue(
            any(
                f["rule_id"] == "reviewer.post_merge_validation"
                for f in result["findings"]
            )
        )
    def test_review_pr_allows_proven_post_merge_moot(self):
        result = assess_final_report_validator(_MOOT_PROOF, "review_pr")
        self.assertFalse(
            any(
                f["rule_id"] == "reviewer.post_merge_validation"
                for f in result["findings"]
            )
        )
 if __name__ == "__main__":
    unittest.main()
@@ -287,49 +287,6 @@ class TestResolveTaskCapability(unittest.TestCase):
            self.assertEqual(res["required_operation_permission"], "gitea.pr.merge")
            self.assertEqual(res["required_role_kind"], "merger")
    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_reviewer_role_with_merge_permission_still_cannot_merge(self, _auth, _api):
        # #539: exact permission alone is insufficient. The active profile
        # role must also be merger for merge_pr.
        config = json.loads(json.dumps(CONFIG_RESOLVER))
        config["profiles"]["reviewer-with-merge"] = {
            "enabled": True,
            "context": "ctx",
            "role": "reviewer",
            "username": "reviewer-user",
            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
            "allowed_operations": [
                "gitea.read", "gitea.pr.review", "gitea.pr.approve",
                "gitea.pr.merge",
            ],
            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
            "execution_profile": "reviewer-with-merge",
        }
        self._write_config(config)
        with patch.dict(os.environ, self._env("reviewer-with-merge")):
            res = mcp_server.gitea_resolve_task_capability(
                task="merge_pr", remote="prgs"
            )
        self.assertEqual(res["required_role_kind"], "merger")
        self.assertEqual(res["active_role_kind"], "reviewer")
        self.assertTrue(res["active_profile_permission_allowed"])
        self.assertFalse(res["allowed_in_current_session"])
        self.assertTrue(res["stop_required"])
        self.assertIn("cannot perform merger task", " ".join(res["task_role_guidance"]))
    @patch("mcp_server.api_request", return_value={"login": "merger-user"})
    @patch("mcp_server.get_auth_header", return_value="token merger-pass")
    def test_merger_profile_can_resolve_merge_task(self, _auth, _api):
        with patch.dict(os.environ, self._env("merger-profile")):
            res = mcp_server.gitea_resolve_task_capability(
                task="merge_pr", remote="prgs"
            )
        self.assertEqual(res["required_role_kind"], "merger")
        self.assertEqual(res["active_role_kind"], "merger")
        self.assertTrue(res["allowed_in_current_session"])
        self.assertFalse(res["stop_required"])
    @patch("mcp_server.api_request")
    def test_self_review_blocked_structured(self, mock_api):
        # Test that self-approve (self-review gate) is blocked and returns structured reasons
@@ -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])
@@ -237,113 +237,5 @@ class TestReviewerLeaseMcpGate(unittest.TestCase):
        self.assertTrue(any("lease" in r.lower() for r in reasons))
 class TestReviewerLeaseHandoffDiagnose(unittest.TestCase):
    """#599: canonical next-action diagnosis for open-PR lease handoff."""
    def setUp(self):
        leases.clear_session_lease()
    def test_no_lease_next_action_acquire(self):
        result = leases.diagnose_reviewer_pr_lease_handoff(
            [],
            pr_number=592,
            current_session_id="me-1",
            current_reviewer_identity="sysadmin",
            proposed_worktree="branches/review-pr-592",
        )
        self.assertEqual(result["classification"], "no_lease")
        self.assertEqual(result["next_action"], leases.NEXT_ACTION_ACQUIRE)
        self.assertIsNone(result["active_lease"])
    def test_foreign_active_wait_pr592_style(self):
        # Instructed lease gone; newer foreign claim is active.
        old = _lease_comment(592, "23139-da018dab43ba", phase="released")
        old["id"] = 8640
        active = _lease_comment(592, "51515-2bfb70f0685f", phase="claimed")
        active["id"] = 8647
        result = leases.diagnose_reviewer_pr_lease_handoff(
            [old, active],
            pr_number=592,
            current_session_id="fresh-session",
            current_reviewer_identity="sysadmin",
            proposed_worktree="branches/review-pr-592-issue-590",
            instructed_session_id="23139-da018dab43ba",
            instructed_comment_id=8640,
        )
        self.assertEqual(
            result["classification"],
            "instructed_lease_missing_with_replacement",
        )
        self.assertEqual(result["next_action"], leases.NEXT_ACTION_WAIT)
        self.assertTrue(result["instructed_lease_missing_with_replacement"])
        self.assertEqual(result["active_lease"]["session_id"], "51515-2bfb70f0685f")
        self.assertEqual(result["active_lease"]["comment_id"], 8647)
        self.assertFalse(result["mutation_allowed"])
    def test_foreign_reclaimable_release_expired(self):
        reclaim = _lease_comment(
            592, "foreign-old", phase="claimed", minutes_ago=65
        )
        reclaim["id"] = 99
        result = leases.diagnose_reviewer_pr_lease_handoff(
            [reclaim],
            pr_number=592,
            current_session_id="me-1",
            current_reviewer_identity="sysadmin",
            proposed_worktree="branches/review-pr-592",
        )
        self.assertEqual(result["classification"], "foreign_reclaimable")
        self.assertEqual(
            result["next_action"], leases.NEXT_ACTION_RELEASE_EXPIRED_LEASE
        )
    def test_own_active_resume(self):
        head = "a" * 40
        claim = _lease_comment(592, "me-1", phase="claimed", candidate_head=head)
        claim["id"] = 10
        leases.record_session_lease({
            "pr_number": 592,
            "session_id": "me-1",
            "candidate_head": head,
            "comment_id": 10,
        }, lease_provenance=mla.build_lease_provenance(
            source=mla.SOURCE_ACQUIRE,
            comment_id=10,
        ))
        result = leases.diagnose_reviewer_pr_lease_handoff(
            [claim],
            pr_number=592,
            current_session_id="me-1",
            current_reviewer_identity="rev1",
            proposed_worktree="branches/review-pr382",
        )
        self.assertEqual(result["classification"], "own_active")
        self.assertEqual(
            result["next_action"], leases.NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
        )
        self.assertTrue(result["mutation_allowed"])
    def test_worktree_binding_mismatch(self):
        claim = _lease_comment(592, "me-1", phase="claimed")
        claim["id"] = 11
        # _lease_comment uses branches/review-pr382
        result = leases.diagnose_reviewer_pr_lease_handoff(
            [claim],
            pr_number=592,
            current_session_id="me-1",
            current_reviewer_identity="rev1",
            proposed_worktree="branches/review-pr-592-issue-590",
            env_bound_worktree="branches/review-pr-538",
        )
        self.assertEqual(result["classification"], "worktree_binding_mismatch")
        self.assertEqual(
            result["next_action"], leases.NEXT_ACTION_REPAIR_WORKTREE_BINDING
        )
        self.assertFalse(result["worktree_binding"]["match"])
        self.assertFalse(result["mutation_allowed"])
 if __name__ == "__main__":
    unittest.main()
@@ -52,20 +52,6 @@ CONFIG = {
            ],
            "execution_profile": "prgs-reviewer",
        },
        "prgs-merger": {
            "enabled": True,
            "context": "ctx",
            "role": "merger",
            "username": "sysadmin",
            "auth": {"type": "env", "name": "GITEA_TOKEN_MERGER"},
            "allowed_operations": [
                "gitea.read", "gitea.pr.merge", "gitea.issue.comment",
            ],
            "forbidden_operations": [
                "gitea.pr.create", "gitea.branch.push", "gitea.pr.approve",
            ],
            "execution_profile": "prgs-merger",
        },
        "prgs-reconciler": {
            "enabled": True,
            "context": "ctx",
@@ -118,7 +104,6 @@ class TestRoleSessionRouter(unittest.TestCase):
            "GITEA_MCP_PROFILE": profile,
            "GITEA_TOKEN_AUTHOR": "author-pass",
            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
            "GITEA_TOKEN_MERGER": "merger-pass",
            "GITEA_TOKEN_RECONCILER": "reconciler-pass",
        }
@@ -242,31 +227,6 @@ class TestRoleSessionRouter(unittest.TestCase):
        self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
        self.assertTrue(route["downstream_allowed"])
    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_merge_task_under_reviewer_profile_wrong_role_stop(self, _auth, _api):
        # #539: a reviewer session must not pass the pre-task merge route
        # even if its config accidentally includes gitea.pr.merge.
        with patch.dict(os.environ, self._env("prgs-reviewer")):
            route = mcp_server.gitea_route_task_session(
                task_type="merge_pr", remote="prgs"
            )
        self.assertEqual(route["required_role"], "merger")
        self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
        self.assertFalse(route["downstream_allowed"])
        self.assertIn("merger", route["message"].lower())
    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
    @patch("mcp_server.get_auth_header", return_value="token merger-pass")
    def test_merge_task_under_merger_profile_allowed(self, _auth, _api):
        with patch.dict(os.environ, self._env("prgs-merger")):
            route = mcp_server.gitea_route_task_session(
                task_type="merge_pr", remote="prgs"
            )
        self.assertEqual(route["required_role"], "merger")
        self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
        self.assertTrue(route["downstream_allowed"])
    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_author_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
@@ -1,139 +0,0 @@
 """Documentation acceptance for the stable control runtime ADR (#615 / PR #616).
 Enforces review 443 remediation:
 * F1 — operator guide / runbooks cross-link the ADR (issue #615 AC2).
 * F2 — LLM sessions are not instructed to kill/restart/relaunch MCP; process
  restart is operator-owned; ADR is the authoritative split.
 * F3 — routine post-merge master-parity staleness has a sanctioned response
  (stop → report → operator reload → re-verify parity) and is not a full
  promotion ledger requirement.
 """
 from pathlib import Path
 REPO_ROOT = Path(__file__).resolve().parent.parent
 ADR = (
    REPO_ROOT
    / "docs"
    / "architecture"
    / "mcp-stable-control-runtime-policy-adr.md"
 )
 ADR_REL = "architecture/mcp-stable-control-runtime-policy-adr.md"
 ADR_BASENAME = "mcp-stable-control-runtime-policy-adr.md"
 CROSS_LINK_DOCS = (
    REPO_ROOT / "docs" / "wiki" / "Operator-Guide.md",
    REPO_ROOT / "docs" / "wiki" / "Runbooks.md",
    REPO_ROOT / "docs" / "llm-workflow-runbooks.md",
 )
 def _read(path: Path) -> str:
    assert path.is_file(), f"missing {path.relative_to(REPO_ROOT)}"
    return path.read_text(encoding="utf-8")
 def test_adr_exists_with_policy_core():
    text = _read(ADR)
    assert text.lstrip().startswith("#"), "ADR lacks a title"
    assert "#615" in text
    assert "stable control runtime" in text.lower()
    assert "2.3" in text and "2.4" in text and "2.5" in text and "2.6" in text
 def test_f1_operator_docs_cross_link_adr():
    for path in CROSS_LINK_DOCS:
        text = _read(path)
        assert ADR_BASENAME in text, (
            f"{path.relative_to(REPO_ROOT)} must cross-link {ADR_BASENAME} "
            f"(issue #615 acceptance criterion 2)"
        )
 def test_f1_adr_lists_cross_links_as_acceptance_not_optional_tooling():
    text = _read(ADR)
    # Acceptance section must require guide/runbook cross-links.
    assert "Operator guide / runbooks cross-link" in text or (
        "operator guide" in text.lower() and "cross-link" in text.lower()
        and "Acceptance" in text
    )
    # Cross-link must not remain only as optional tooling item #4.
    optional = text.split("## 5. Implementation follow-ups", 1)[-1].split(
        "## 6. Acceptance", 1
    )[0]
    assert "Operator Guide wiki cross-link to this ADR" not in optional, (
        "ADR §5 must not list operator-guide cross-link as optional tooling"
    )
    assert "Not optional" in text or "must** cross-link" in text.lower() or (
        "must cross-link" in text.lower()
    )
 def test_f2_runbook_fallback_is_operator_owned_not_llm_restart():
    runbooks = _read(REPO_ROOT / "docs" / "llm-workflow-runbooks.md")
    # Forbidden historical self-service instruction.
    forbidden = (
        "the LLM must relaunch or restart the client/MCP with the correct "
        "profile environment variable before claiming or working on any tasks"
    )
    assert forbidden not in runbooks, (
        "llm-workflow-runbooks must not instruct the LLM to relaunch/restart MCP"
    )
    assert "operator-owned" in runbooks.lower() or "Operator-owned" in runbooks
    assert ADR_BASENAME in runbooks
 def test_f2_workspace_rebind_does_not_require_llm_process_relaunch():
    runbooks = _read(REPO_ROOT / "docs" / "llm-workflow-runbooks.md")
    section = runbooks.split("### Safe reconnect / rebind procedure", 1)[-1]
    section = section.split("## Safety notes", 1)[0]
    # Must not tell the LLM alone to relaunch the MCP process as step 2.
    assert "Reconnect or relaunch the correct namespace MCP server from the intended" not in section
    assert "Operator-owned" in section or "operator" in section.lower()
    assert "worktree_path" in section
    collapsed = " ".join(section.lower().split())
    assert "client reconnect" in collapsed
 def test_f2_adr_forbids_llm_restart_and_supersedes_self_service_relaunch():
    text = _read(ADR)
    lower = text.lower()
    assert "must not" in lower and "restart" in lower
    assert "operator" in lower and "reload" in lower
    assert "supersedes" in lower
    assert "llm" in lower
 def test_f3_adr_defines_post_merge_parity_staleness_response():
    text = _read(ADR)
    lower = text.lower()
    assert "2.6" in text
    assert "post-merge" in lower or "post merge" in lower
    assert "parity" in lower and "stale" in lower
    # Sanctioned steps: stop, report, operator reload, re-verify.
    assert "stop" in lower
    assert "report" in lower
    assert "operator" in lower
    assert "parity is verified" in lower or "re-verif" in lower or (
        "startup/current-head" in lower
    )
    # Not a full promotion ledger for routine reload.
    assert "not a §2.4 promotion" in lower or "not a section 2.4 promotion" in lower or (
        "not a §2.4" in text or "Not a §2.4 promotion" in text
    )
    # Stale parity listed among unhealthy triggers.
    assert "master parity is stale" in lower or "stale master parity" in lower
 def test_f3_adr_forbids_llm_bypass_of_parity_gate():
    text = _read(ADR)
    lower = text.lower()
    assert "bypass" in lower or "self-reset" in lower or "self-service" in lower
    assert "parity" in lower
 def test_cross_links_do_not_embed_secrets_or_raw_hosts_in_wiki_snippets():
    for path in CROSS_LINK_DOCS:
        text = _read(path)
        for marker in ("ghp_", "BEGIN PRIVATE KEY", "Authorization: Bearer"):
            assert marker not in text, f"{path} contains {marker!r}"
@@ -1,378 +0,0 @@
 """Stale #332 review-decision lock cleanup (#594).
 Proves:
  a. active terminal locks still block unrelated terminal reviews
  b. merged/closed PR locks can be cleared canonically
  c. cleanup cannot bypass review gates on open PRs
  d. after moot cleanup, mark_ready for a different PR can proceed
     (PR #592-style after PR #586-style stale approve)
 """
 from __future__ import annotations
 import os
 import unittest
 from unittest.mock import patch
 import mcp_server
 import stale_review_decision_lock as srdl
 from mcp_server import (
    gitea_cleanup_stale_review_decision_lock,
    gitea_mark_final_review_decision,
 )
 HEAD = "a" * 40
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 REVIEWER_ENV = {
    "GITEA_PROFILE_NAME": "prgs-reviewer",
    "GITEA_ALLOWED_OPERATIONS": (
        "gitea.read,gitea.pr.review,gitea.pr.approve,"
        "gitea.pr.request_changes,gitea.pr.comment"
    ),
    "GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer",
 }
 REVIEWER_PROFILE = {
    "profile_name": "prgs-reviewer",
    "allowed_operations": [
        "gitea.read",
        "gitea.pr.review",
        "gitea.pr.approve",
        "gitea.pr.request_changes",
        "gitea.pr.comment",
    ],
    "forbidden_operations": [
        "gitea.pr.merge",
        "gitea.pr.create",
        "gitea.branch.push",
    ],
 }
 def _reviewer_profile_patch():
    return patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE)
 def _lock(mutations=None, correction=False):
    return {
        "task": "review_pr",
        "remote": "prgs",
        "org": "Scaled-Tech-Consulting",
        "repo": "Gitea-Tools",
        "session_pid": os.getpid(),
        "session_profile": "prgs-reviewer",
        "session_profile_lock": "prgs-reviewer",
        "profile_identity": "prgs-reviewer",
        "final_review_decision_ready": False,
        "ready_pr_number": None,
        "ready_action": None,
        "ready_expected_head_sha": None,
        "ready_remote": None,
        "ready_org": None,
        "ready_repo": None,
        "live_mutations": list(mutations or []),
        "correction_authorized": correction,
        "correction_reason": None,
    }
 APPROVED_586 = {
    "pr_number": 586,
    "action": "approve",
    "review_id": 395,
    "review_state": "approve",
 }
 APPROVED_OPEN = {
    "pr_number": 700,
    "action": "approve",
    "review_id": 1,
    "review_state": "approve",
 }
 RC_OPEN = {
    "pr_number": 701,
    "action": "request_changes",
    "review_id": 2,
    "review_state": "request_changes",
 }
 def _seed(mutations=None, correction=False):
    import review_workflow_load
    review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
    mcp_server._save_review_decision_lock(_lock(mutations, correction))
    mcp_server.gitea_load_review_workflow()
 def _merged_pr(pr_number=586, sha="6913ac9" + "0" * 33):
    return {
        "number": pr_number,
        "state": "closed",
        "merged": True,
        "merged_at": "2026-07-09T18:20:03Z",
        "merge_commit_sha": sha,
    }
 def _open_pr(pr_number=700):
    return {
        "number": pr_number,
        "state": "open",
        "merged": False,
        "merged_at": None,
        "merge_commit_sha": None,
    }
 # --------------------------------------------------------------------------- #
 # Pure assessment
 # --------------------------------------------------------------------------- #
 class TestAssessStaleLock(unittest.TestCase):
    def test_no_lock(self):
        a = srdl.assess_stale_review_decision_lock(None)
        self.assertFalse(a["cleanup_allowed"])
        self.assertFalse(a["is_moot"])
    def test_no_terminal_mutations(self):
        a = srdl.assess_stale_review_decision_lock(_lock([]))
        self.assertFalse(a["cleanup_allowed"])
    def test_open_pr_not_cleanable(self):
        a = srdl.assess_stale_review_decision_lock(
            _lock([APPROVED_OPEN]), pr_live=_open_pr(700)
        )
        self.assertFalse(a["is_moot"])
        self.assertFalse(a["cleanup_allowed"])
        self.assertTrue(any("open PR" in r for r in a["reasons"]))
    def test_merged_pr_is_moot_and_cleanable(self):
        a = srdl.assess_stale_review_decision_lock(
            _lock([APPROVED_586]), pr_live=_merged_pr(586)
        )
        self.assertTrue(a["is_moot"])
        self.assertTrue(a["cleanup_allowed"])
        self.assertEqual(a["last_terminal_pr"], 586)
    def test_closed_unmerged_is_moot(self):
        a = srdl.assess_stale_review_decision_lock(
            _lock([RC_OPEN]),
            pr_live={"number": 701, "state": "closed", "merged": False},
        )
        self.assertTrue(a["is_moot"])
        self.assertTrue(a["cleanup_allowed"])
    def test_profile_mismatch_blocks(self):
        a = srdl.assess_stale_review_decision_lock(
            _lock([APPROVED_586]),
            pr_live=_merged_pr(586),
            active_profile_identity="other-profile",
        )
        self.assertFalse(a["cleanup_allowed"])
        self.assertFalse(a["profile_match"])
    def test_lookup_error_fail_closed(self):
        a = srdl.assess_stale_review_decision_lock(
            _lock([APPROVED_586]),
            pr_lookup_error="timeout",
        )
        self.assertFalse(a["cleanup_allowed"])
        self.assertTrue(any("timeout" in r for r in a["reasons"]))
 # --------------------------------------------------------------------------- #
 # Hard-stop still blocks active locks
 # --------------------------------------------------------------------------- #
 class TestActiveHardStopPreserved(unittest.TestCase):
    def tearDown(self):
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()
    def test_open_approve_still_blocks_other_pr_mark_ready(self):
        _seed([APPROVED_OPEN])
        reasons = mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready")
        self.assertTrue(reasons)
        self.assertIn("#332", reasons[0])
        self.assertIn("gitea_cleanup_stale_review_decision_lock", reasons[0])
    def test_request_changes_still_blocks_everything(self):
        _seed([RC_OPEN])
        for op in ("merge", "mark_ready", "review"):
            self.assertTrue(
                mcp_server.terminal_review_hard_stop_reasons(592, op),
                msg=op,
            )
 # --------------------------------------------------------------------------- #
 # MCP cleanup tool
 # --------------------------------------------------------------------------- #
 class TestCleanupTool(unittest.TestCase):
    def setUp(self):
        self._env = patch.dict(os.environ, REVIEWER_ENV, clear=False)
        self._env.start()
    def tearDown(self):
        self._env.stop()
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()
    def test_read_only_reports_moot_without_clearing(self):
        _seed([APPROVED_586])
        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ), \
                _reviewer_profile_patch(), \
                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
            r = gitea_cleanup_stale_review_decision_lock(
                apply=False, remote="prgs",
                org="Scaled-Tech-Consulting", repo="Gitea-Tools",
            )
        self.assertTrue(r["success"])
        self.assertTrue(r["is_moot"])
        self.assertTrue(r["cleanup_allowed"])
        self.assertFalse(r["cleanup_performed"])
        self.assertIsNotNone(mcp_server._load_review_decision_lock())
    def test_apply_clears_moot_lock(self):
        _seed([APPROVED_586])
        posts = []
        def _api(method, url, auth, payload=None):
            if method == "GET" and "/pulls/586" in url:
                return _merged_pr()
            if method == "POST" and "/comments" in url:
                posts.append(payload)
                return {"id": 9999}
            return {}
        with patch.object(mcp_server, "api_request", side_effect=_api), \
                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ), \
                _reviewer_profile_patch(), \
                patch.object(mcp_server, "_profile_operation_gate", return_value=[]), \
                patch.object(
                    mcp_server, "_audited",
                    side_effect=lambda *a, **k: __import__(
                        "contextlib"
                    ).nullcontext(),
                ):
            r = gitea_cleanup_stale_review_decision_lock(
                apply=True,
                expected_terminal_pr=586,
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo="Gitea-Tools",
            )
        self.assertTrue(r["cleanup_performed"], r)
        self.assertTrue(r["audit"]["applied"])
        self.assertIsNone(mcp_server._load_review_decision_lock())
        self.assertEqual(r.get("audit_comment_id"), 9999)
        self.assertTrue(posts)
    def test_apply_refuses_open_pr(self):
        _seed([APPROVED_OPEN])
        with patch.object(mcp_server, "api_request", return_value=_open_pr(700)), \
                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ), \
                _reviewer_profile_patch(), \
                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
            r = gitea_cleanup_stale_review_decision_lock(
                apply=True, remote="prgs",
                org="Scaled-Tech-Consulting", repo="Gitea-Tools",
            )
        self.assertFalse(r["cleanup_performed"])
        self.assertFalse(r["cleanup_allowed"])
        self.assertIsNotNone(mcp_server._load_review_decision_lock())
    def test_expected_terminal_pr_pin(self):
        _seed([APPROVED_586])
        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ), \
                _reviewer_profile_patch(), \
                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
            r = gitea_cleanup_stale_review_decision_lock(
                apply=True,
                expected_terminal_pr=999,
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo="Gitea-Tools",
            )
        self.assertFalse(r["cleanup_performed"])
        self.assertTrue(any("expected_terminal_pr" in x for x in r["reasons"]))
    def test_after_moot_cleanup_other_pr_mark_ready_unblocked(self):
        """PR #592-style: after clearing merged #586 lock, new mark_ready works."""
        _seed([APPROVED_586])
        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ), \
                _reviewer_profile_patch(), \
                patch.object(mcp_server, "_profile_operation_gate", return_value=[]), \
                patch.object(
                    mcp_server, "_audited",
                    side_effect=lambda *a, **k: __import__(
                        "contextlib"
                    ).nullcontext(),
                ):
            cleaned = gitea_cleanup_stale_review_decision_lock(
                apply=True,
                expected_terminal_pr=586,
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo="Gitea-Tools",
                post_audit_comment=False,
            )
        self.assertTrue(cleaned["cleanup_performed"], cleaned)
        # Hard-stop gone; re-init clean lock like a fresh review_pr resolve.
        mcp_server.init_review_decision_lock(remote="prgs", task="review_pr")
        mcp_server.gitea_load_review_workflow()
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready"),
            [],
        )
        no_lease = {"block": False, "reasons": [], "mutation_allowed": True}
        with patch.object(mcp_server, "_list_pr_lease_comments", return_value=[]), \
                patch.object(
                    mcp_server, "_pr_work_lease_reviewer_block", return_value=no_lease
                ), \
                patch.object(
                    mcp_server,
                    "gitea_check_pr_eligibility",
                    return_value={
                        "eligible": True,
                        "reasons": [],
                        "authenticated_user": "sysadmin",
                        "pr_author": "jcwalker3",
                        "self_author": False,
                    },
                ), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ):
            marked = gitea_mark_final_review_decision(
                pr_number=592,
                action="approve",
                expected_head_sha=HEAD,
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo="Gitea-Tools",
            )
        self.assertTrue(marked.get("marked_ready"), marked)
 if __name__ == "__main__":
    unittest.main()