fix(mcp): make capability resolver side-effect free (Closes #685 )

Stale-runtime detection remains fail-closed, but gitea_resolve_task_capability must never touch mcp_config.json, spawn recovery threads, or call os._exit. - Remove _trigger_mcp_auto_restart from the read-only diagnostics path - Return blocker_kind=runtime_reconnect_required with mutation_performed=false - Keep exact_safe_next_action pointing at IDE/client reconnect when stale - Add regression suite across author/reviewer/merger/reconciler profiles
2026-07-16 21:21:39 -04:00
6 changed files with 387 additions and 2220 deletions
@@ -1,807 +0,0 @@
 """Common anti-stomp preflight for every MCP mutation tool (#604).
 Even with leases, mutation tools need a **shared** preflight that prevents
 stale sessions, wrong worktrees, wrong repos, old prompts, terminal locks,
 foreign leases, contaminated approvals, and root-checkout mutations from
 slipping through.
 This module is the pure assessment core. Callers (MCP mutation entrypoints)
 gather live facts and pass them in — nothing here performs git, network, or
 durable-state I/O. Existing happy-path guards remain authoritative; this
 module composes them into one typed, fail-closed result.
 Required checks (issue #604):
 * repo/org verification
 * profile/role verification
 * root checkout clean and not used for mutation (when role requires branches/)
 * worktree under branches when required
 * active lease ownership (when lease is required for the mutation)
 * terminal lock status (when applicable)
 * expected head SHA (when pinned)
 * stale runtime status
 * workflow hash (when a workflow load is required)
 * source contamination status
 * no manual state/mtime/source-file bypass
 Failure returns a typed blocker and an exact next action. There is **no**
 agent-facing bypass flag.
 """
 from __future__ import annotations
 from typing import Any
 import author_mutation_worktree
 import master_parity_gate
 import remote_repo_guard
 import root_checkout_guard
 import stable_branch_push_guard
 # ── public constants ──────────────────────────────────────────────────────────
 # Typed blocker kinds returned in the structured result.
 BLOCKER_WRONG_REPO = "wrong_repo"
 BLOCKER_WRONG_ROLE = "wrong_role"
 BLOCKER_ROOT_CHECKOUT = "root_checkout_mutation"
 BLOCKER_WRONG_WORKTREE = "wrong_worktree"
 BLOCKER_FOREIGN_LEASE = "foreign_lease"
 BLOCKER_TERMINAL_LOCK = "terminal_lock"
 BLOCKER_HEAD_SHA = "head_sha_mismatch"
 BLOCKER_STALE_RUNTIME = "stale_runtime"
 BLOCKER_WORKFLOW_HASH = "workflow_hash"
 BLOCKER_SOURCE_CONTAMINATION = "source_contamination"
 BLOCKER_MANUAL_BYPASS = "manual_bypass"
 BLOCKER_KINDS = frozenset({
    BLOCKER_WRONG_REPO,
    BLOCKER_WRONG_ROLE,
    BLOCKER_ROOT_CHECKOUT,
    BLOCKER_WRONG_WORKTREE,
    BLOCKER_FOREIGN_LEASE,
    BLOCKER_TERMINAL_LOCK,
    BLOCKER_HEAD_SHA,
    BLOCKER_STALE_RUNTIME,
    BLOCKER_WORKFLOW_HASH,
    BLOCKER_SOURCE_CONTAMINATION,
    BLOCKER_MANUAL_BYPASS,
 })
 # Mutation tasks that must invoke the shared anti-stomp preflight before
 # acting (issue #604 AC1). Every member must appear as a live task= kwarg
 # to verify_preflight_purity / _run_anti_stomp_preflight (or the review
 # anti_task map) in gitea_mcp_server.py. Inventory↔wiring tests fail if
 # a declared task is not wired.
 MUTATION_TASKS = frozenset({
    "create_issue",
    "comment_issue",
    "close_issue",
    "mark_issue",
    "lock_issue",
    "set_issue_labels",
    "create_label",
    "create_pr",
    "close_pr",
    "edit_pr",
    "commit_files",
    "gitea_commit_files",
    "delete_branch",
    "cleanup_merged_pr_branch",
    "cleanup_stale_claims",
    "reconcile_merged_cleanups",
    "reconcile_already_landed_pr",
    "reconcile_close_superseded_pr",
    "post_heartbeat",
    "acquire_reviewer_pr_lease",
    "gitea_acquire_reviewer_pr_lease",
    "adopt_merger_pr_lease",
    "review_pr",
    "submit_pr_review",
    "approve_pr",
    "request_changes_pr",
    "merge_pr",
 })
 # Intentionally excluded from MUTATION_TASKS. Each has a dedicated
 # fail-closed gate that preserves decision-lock ownership, workflow-hash,
 # head-SHA, and repository consistency. Do not re-add without either
 # wiring shared preflight or updating this rationale (#604 Blocker B).
 DEDICATED_GATE_MUTATIONS: dict[str, str] = {
    "mark_final_review_decision": (
        "Local review-decision lock only. Enforced by session lock ownership, "
        "workflow-hash, expected head SHA, PR work lease, eligibility, and "
        "terminal-lock gates. No Gitea review POST; shared anti-stomp would "
        "duplicate without side-effect-ordering value."
    ),
    "save_review_draft": (
        "Local session-state draft save with live PR head/base consistency "
        "checks. Not a Gitea review/merge mutation; dedicated head-SHA and "
        "worktree resolution gates apply."
    ),
    "resume_review_draft": (
        "Live-state resume with terminal lock, lease ownership, head/base SHA, "
        "and parity checks. When submit=True, delegates to gitea_submit_pr_review "
        "which runs shared anti-stomp before the Gitea review POST."
    ),
    "cleanup_stale_review_decision_lock": (
        "Moot-lock cleanup with identity match, live PR merged/closed proof, "
        "and reviewer capability gate (#594). Distinct from shared anti-stomp."
    ),
    "gitea_cleanup_stale_review_decision_lock": (
        "Alias of cleanup_stale_review_decision_lock; dedicated #594 path."
    ),
    "heartbeat_reviewer_pr_lease": (
        "Lease heartbeat posts via verify_preflight_purity(task='review_pr') "
        "plus in-session lease ownership checks; not a separate mutation class."
    ),
    "release_reviewer_pr_lease": (
        "Lease release uses workspace binding + ownership checks; posts only "
        "when the session owns the active lease."
    ),
 }
 # Roles that must mutate from a branches/ worktree (not the control checkout).
 _BRANCHES_REQUIRED_ROLES = frozenset({"author"})
 # Reviewer/merger mutations that require an owned lease + head pinning when
 # the caller supplies those facts.
 _LEASE_AWARE_TASKS = frozenset({
    "review_pr",
    "submit_pr_review",
    "approve_pr",
    "request_changes_pr",
    "merge_pr",
    "acquire_reviewer_pr_lease",
    "gitea_acquire_reviewer_pr_lease",
    "adopt_merger_pr_lease",
 })
 _NEXT_ACTIONS: dict[str, str] = {
    BLOCKER_WRONG_REPO: (
        "Pass explicit org= and repo= matching the local git remote "
        "(e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools) and retry."
    ),
    BLOCKER_WRONG_ROLE: (
        "Call gitea_resolve_task_capability, switch to the required "
        "namespace/profile, re-verify with gitea_whoami, then retry."
    ),
    BLOCKER_ROOT_CHECKOUT: (
        "Leave the root control checkout on clean master; create or switch "
        "to a session-owned worktree under branches/ and retry the mutation "
        "with worktree_path set."
    ),
    BLOCKER_WRONG_WORKTREE: (
        "Create or bind a session-owned worktree under branches/, set "
        "worktree_path (or GITEA_ACTIVE_WORKTREE), and retry."
    ),
    BLOCKER_FOREIGN_LEASE: (
        "Stop. Do not stomp a foreign lease. Wait for expiry, request "
        "takeover through the sanctioned path, or hand off to the owner."
    ),
    BLOCKER_TERMINAL_LOCK: (
        "Stop. A terminal review-decision lock is active for this head. "
        "Do not re-approve/merge; follow the #332/#620 recovery path."
    ),
    BLOCKER_HEAD_SHA: (
        "Re-fetch the live PR head with gitea_view_pr, re-pin "
        "expected_head_sha to the current head, re-validate, then retry."
    ),
    BLOCKER_STALE_RUNTIME: (
        "Restart the Gitea MCP server so it reloads master's capability "
        "gates, re-run gitea_whoami + gitea_resolve_task_capability, then retry."
    ),
    BLOCKER_WORKFLOW_HASH: (
        "Reload the canonical workflow via gitea_load_review_workflow, then "
        "rerun the full review-merge workflow from inventory (no approve/merge replay)."
    ),
    BLOCKER_SOURCE_CONTAMINATION: (
        "Stop. Session is source-contaminated. Hand off to a reconciler for "
        "audit/clear; do not clear markers by deleting session-state files."
    ),
    BLOCKER_MANUAL_BYPASS: (
        "Stop. Manual state/mtime/source-file bypass is forbidden. Use "
        "sanctioned MCP tools only; never delete or rewrite session-state "
        "or lock files by hand."
    ),
 }
 def is_mutation_task(task: str | None) -> bool:
    """True when *task* is in the #604 mutation set (normalized)."""
    name = (task or "").strip().lower().removeprefix("gitea_")
    if not name:
        return False
    if name in MUTATION_TASKS:
        return True
    # Accept gitea_ prefixed membership for aliases already in the set.
    return f"gitea_{name}" in MUTATION_TASKS
 def roles_compatible(active_role: str | None, required_role: str | None) -> bool:
    """Whether *active_role* may perform a task that maps to *required_role*.
    Exact match always passes. Reconcilers may run author-class mutations
    (comment/close/cleanup) that the capability map stamps as ``author`` —
    matching the long-standing reconciler exemption for control-checkout
    work. No other cross-role substitutions are allowed.
    Capability-authorized multi-role tasks (e.g. reviewer holding
    ``gitea.issue.comment`` for ``comment_issue``) are handled by
    :func:`authorization_compatible`, not by expanding this role matrix.
    """
    active = (active_role or "").strip().lower()
    required = (required_role or "").strip().lower()
    if not active or not required:
        return True
    if active == required:
        return True
    if active == "reconciler" and required in {"author", "reconciler"}:
        return True
    return False
 def authorization_compatible(
    active_role: str | None,
    required_role: str | None,
    *,
    required_permission: str | None = None,
    allowed_operations: Any = None,
 ) -> bool:
    """Whether the active session may run a task given role *and* capability.
    Order:
    1. :func:`roles_compatible` (exact role or narrow reconciler→author).
    2. Capability possession: when *required_permission* is non-empty and
       present in *allowed_operations*, allow even if the nominal task role
       differs (e.g. reviewer/merger with ``gitea.issue.comment`` on
       ``comment_issue`` / ``mark_issue`` / ``set_issue_labels`` /
       ``lock_issue``).
    Fail closed otherwise. This is **not** a broad reviewer→author or
    merger→author role rewrite: without the specific permission the check
    still denies. Missing *allowed_operations* when a permission is required
    also fails closed (callers must supply the live profile op list).
    """
    if roles_compatible(active_role, required_role):
        return True
    perm = (required_permission or "").strip()
    if not perm:
        return False
    if allowed_operations is None:
        return False
    try:
        ops = {str(o).strip() for o in allowed_operations if o is not None}
    except TypeError:
        return False
    return perm in ops
 def _blocker(
    kind: str,
    reasons: list[str],
    *,
    exact_next_action: str | None = None,
    detail: dict | None = None,
 ) -> dict[str, Any]:
    if kind not in BLOCKER_KINDS:
        raise ValueError(f"unknown anti-stomp blocker kind: {kind!r}")
    return {
        "kind": kind,
        "reasons": list(reasons),
        "exact_next_action": exact_next_action or _NEXT_ACTIONS[kind],
        "detail": dict(detail or {}),
    }
 def _allowed_result(checks: dict[str, Any]) -> dict[str, Any]:
    return {
        "allowed": True,
        "block": False,
        "blockers": [],
        "reasons": [],
        "exact_next_action": "proceed",
        "blocker_kind": None,
        "checks": checks,
    }
 def _blocked_result(
    blockers: list[dict[str, Any]],
    checks: dict[str, Any],
 ) -> dict[str, Any]:
    primary = blockers[0]
    all_reasons: list[str] = []
    for b in blockers:
        for r in b.get("reasons") or []:
            if r not in all_reasons:
                all_reasons.append(r)
    return {
        "allowed": False,
        "block": True,
        "blockers": blockers,
        "reasons": all_reasons,
        "exact_next_action": primary["exact_next_action"],
        "blocker_kind": primary["kind"],
        "checks": checks,
    }
 def assess_anti_stomp_preflight(
    *,
    task: str | None = None,
    # repo/org
    remote: str | None = None,
    resolved_org: str | None = None,
    resolved_repo: str | None = None,
    local_remote_url: str | None = None,
    org_explicit: bool = False,
    repo_explicit: bool = False,
    check_repo: bool = True,
    # profile/role + capability
    profile_name: str | None = None,
    profile_role: str | None = None,
    required_role: str | None = None,
    required_permission: str | None = None,
    allowed_operations: Any = None,
    check_role: bool = True,
    # root checkout + worktree
    workspace_path: str | None = None,
    project_root: str | None = None,
    current_branch: str | None = None,
    root_head_sha: str | None = None,
    root_porcelain: str | None = None,
    remote_master_sha: str | None = None,
    check_root_checkout: bool = True,
    check_worktree: bool = True,
    # stale runtime (master parity)
    startup_head: str | None = None,
    current_code_head: str | None = None,
    check_stale_runtime: bool = True,
    # lease ownership
    lease_required: bool = False,
    foreign_lease: bool | None = None,
    lease_owner_session: str | None = None,
    active_session_id: str | None = None,
    lease_owner_identity: str | None = None,
    active_identity: str | None = None,
    lease_reasons: list[str] | None = None,
    # terminal lock
    terminal_lock_blocks: bool | None = None,
    terminal_lock_reasons: list[str] | None = None,
    # expected head SHA
    require_head_sha: bool = False,
    expected_head_sha: str | None = None,
    live_head_sha: str | None = None,
    # workflow hash
    workflow_hash_valid: bool | None = None,
    workflow_hash_reasons: list[str] | None = None,
    # source contamination (stable-branch push / approval contamination)
    source_contaminated: bool | None = None,
    contamination_reasons: list[str] | None = None,
    # manual bypass
    manual_bypass_attempted: bool = False,
    manual_bypass_reasons: list[str] | None = None,
 ) -> dict[str, Any]:
    """Fail-closed common anti-stomp assessment (pure).
    Only checks for which facts are supplied (or which are required by the
    mutation category) are evaluated. Unsupplied optional facts are recorded
    as ``skipped`` so callers can see coverage without inventing defaults.
    Returns a structured dict with ``allowed``/``block``, typed ``blockers``,
    aggregated ``reasons``, ``exact_next_action``, and per-check ``checks``.
    """
    task_name = (task or "").strip()
    role = (profile_role or "").strip().lower() or None
    req_role = (required_role or "").strip().lower() or None
    req_perm = (required_permission or "").strip() or None
    checks: dict[str, Any] = {
        "task": task_name or None,
        "profile_name": profile_name,
        "profile_role": role,
        "required_role": req_role,
        "required_permission": req_perm,
    }
    blockers: list[dict[str, Any]] = []
    # ── manual bypass (always first; never skippable when attempted) ─────────
    if manual_bypass_attempted:
        reasons = list(manual_bypass_reasons or [
            "manual state/mtime/source-file bypass attempt detected (fail closed)"
        ])
        blockers.append(_blocker(BLOCKER_MANUAL_BYPASS, reasons))
        checks["manual_bypass"] = {"block": True, "reasons": reasons}
    else:
        checks["manual_bypass"] = {"block": False, "skipped": False}
    # ── repo/org ─────────────────────────────────────────────────────────────
    if check_repo and resolved_org is not None and resolved_repo is not None:
        repo_assessment = remote_repo_guard.assess_remote_repo_match(
            remote=remote or "",
            resolved_org=resolved_org,
            resolved_repo=resolved_repo,
            local_remote_url=local_remote_url,
            org_explicit=org_explicit,
            repo_explicit=repo_explicit,
        )
        checks["repo"] = {
            "block": bool(repo_assessment.get("block")),
            "reasons": list(repo_assessment.get("reasons") or []),
            "resolved_org": resolved_org,
            "resolved_repo": resolved_repo,
        }
        if repo_assessment.get("block"):
            blockers.append(
                _blocker(
                    BLOCKER_WRONG_REPO,
                    list(repo_assessment.get("reasons") or ["repo/org mismatch"]),
                    detail={
                        "resolved_org": resolved_org,
                        "resolved_repo": resolved_repo,
                        "local_remote_url": local_remote_url,
                    },
                )
            )
    else:
        checks["repo"] = {"block": False, "skipped": True}
    # ── profile/role + capability ────────────────────────────────────────────
    # Role match OR possession of the task's required permission (Blocker A).
    # Reviewer/merger may run issue-comment-class tasks when they hold
    # gitea.issue.comment; unauthorized escalation without the permission
    # still fails closed.
    if check_role and req_role and role:
        authorized = authorization_compatible(
            role,
            req_role,
            required_permission=req_perm,
            allowed_operations=allowed_operations,
        )
        if not authorized:
            perm_clause = (
                f", required_permission='{req_perm}'" if req_perm else ""
            )
            reasons = [
                f"active profile role '{role}' is not authorized for task "
                f"'{task_name or '(unknown)'}' (required_role='{req_role}'"
                f"{perm_clause}; profile={profile_name or '(unknown)'})"
            ]
            checks["role"] = {
                "block": True,
                "reasons": reasons,
                "required_permission": req_perm,
                "capability_authorized": False,
            }
            blockers.append(
                _blocker(
                    BLOCKER_WRONG_ROLE,
                    reasons,
                    detail={
                        "profile_name": profile_name,
                        "profile_role": role,
                        "required_role": req_role,
                        "required_permission": req_perm,
                    },
                )
            )
        else:
            checks["role"] = {
                "block": False,
                "reasons": [],
                "required_permission": req_perm,
                "capability_authorized": bool(
                    req_perm
                    and allowed_operations is not None
                    and req_perm in {
                        str(o).strip() for o in (allowed_operations or [])
                        if o is not None
                    }
                    and not roles_compatible(role, req_role)
                ),
            }
    else:
        checks["role"] = {"block": False, "skipped": True}
    # ── stale runtime (master parity) ────────────────────────────────────────
    if check_stale_runtime and (
        startup_head is not None or current_code_head is not None
    ):
        parity = master_parity_gate.assess_master_parity(
            {"startup_head": startup_head},
            current_code_head,
        )
        stale_reasons = master_parity_gate.parity_block_reasons(parity)
        checks["stale_runtime"] = {
            "block": bool(stale_reasons),
            "reasons": list(stale_reasons),
            "startup_head": startup_head,
            "current_code_head": current_code_head,
            "stale": bool(parity.get("stale")),
        }
        if stale_reasons:
            blockers.append(
                _blocker(
                    BLOCKER_STALE_RUNTIME,
                    list(stale_reasons),
                    detail={
                        "startup_head": startup_head,
                        "current_code_head": current_code_head,
                    },
                )
            )
    else:
        checks["stale_runtime"] = {"block": False, "skipped": True}
    # ── root checkout ────────────────────────────────────────────────────────
    if (
        check_root_checkout
        and workspace_path is not None
        and project_root is not None
        and root_porcelain is not None
    ):
        root_assessment = root_checkout_guard.assess_root_checkout_guard(
            workspace_path=workspace_path,
            canonical_repo_root=project_root,
            current_branch=current_branch,
            head_sha=root_head_sha,
            porcelain_status=root_porcelain,
            remote_master_sha=remote_master_sha,
            resolved_role=req_role or role,
            actual_role=role,
        )
        checks["root_checkout"] = {
            "block": bool(root_assessment.get("block")),
            "reasons": list(root_assessment.get("reasons") or []),
        }
        if root_assessment.get("block"):
            blockers.append(
                _blocker(
                    BLOCKER_ROOT_CHECKOUT,
                    list(root_assessment.get("reasons") or [
                        "control checkout is not clean master"
                    ]),
                    detail={
                        "workspace_path": workspace_path,
                        "project_root": project_root,
                        "current_branch": current_branch,
                    },
                )
            )
    else:
        checks["root_checkout"] = {"block": False, "skipped": True}
    # ── worktree under branches (author) ─────────────────────────────────────
    worktree_role = role or req_role
    if (
        check_worktree
        and workspace_path is not None
        and project_root is not None
        and worktree_role in _BRANCHES_REQUIRED_ROLES
    ):
        wt = author_mutation_worktree.assess_author_mutation_worktree(
            workspace_path=workspace_path,
            project_root=project_root,
            current_branch=current_branch,
        )
        checks["worktree"] = {
            "block": bool(wt.get("block")),
            "reasons": list(wt.get("reasons") or []),
            "under_branches": wt.get("under_branches"),
        }
        if wt.get("block"):
            blockers.append(
                _blocker(
                    BLOCKER_WRONG_WORKTREE,
                    list(wt.get("reasons") or [
                        "mutation worktree is not under branches/"
                    ]),
                    detail={
                        "workspace_path": workspace_path,
                        "project_root": project_root,
                    },
                )
            )
    else:
        checks["worktree"] = {
            "block": False,
            "skipped": worktree_role not in _BRANCHES_REQUIRED_ROLES
            or workspace_path is None,
        }
    # ── foreign lease ────────────────────────────────────────────────────────
    lease_needed = lease_required or (
        task_name.removeprefix("gitea_") in {
            t.removeprefix("gitea_") for t in _LEASE_AWARE_TASKS
        }
        and foreign_lease is not None
    )
    if lease_needed or foreign_lease is True:
        is_foreign = bool(foreign_lease)
        if foreign_lease is None and lease_required:
            # Ownership must be proven when lease_required; missing ownership
            # facts fail closed.
            owner_ok = (
                (not lease_owner_session and not active_session_id)
                or (
                    lease_owner_session
                    and active_session_id
                    and lease_owner_session == active_session_id
                )
            )
            identity_ok = (
                (not lease_owner_identity and not active_identity)
                or (
                    lease_owner_identity
                    and active_identity
                    and lease_owner_identity == active_identity
                )
            )
            if not owner_ok or not identity_ok:
                is_foreign = True
        reasons = list(lease_reasons or [])
        if is_foreign and not reasons:
            reasons = [
                "active lease is owned by a foreign session or identity "
                "(anti-stomp fail closed)"
            ]
        checks["lease"] = {
            "block": is_foreign,
            "reasons": reasons if is_foreign else [],
            "lease_owner_session": lease_owner_session,
            "active_session_id": active_session_id,
        }
        if is_foreign:
            blockers.append(
                _blocker(BLOCKER_FOREIGN_LEASE, reasons)
            )
    else:
        checks["lease"] = {"block": False, "skipped": True}
    # ── terminal lock ────────────────────────────────────────────────────────
    if terminal_lock_blocks is not None:
        reasons = list(terminal_lock_reasons or [])
        if terminal_lock_blocks and not reasons:
            reasons = [
                "terminal review-decision lock is active for this head "
                "(anti-stomp fail closed)"
            ]
        checks["terminal_lock"] = {
            "block": bool(terminal_lock_blocks),
            "reasons": reasons if terminal_lock_blocks else [],
        }
        if terminal_lock_blocks:
            blockers.append(_blocker(BLOCKER_TERMINAL_LOCK, reasons))
    else:
        checks["terminal_lock"] = {"block": False, "skipped": True}
    # ── expected head SHA ────────────────────────────────────────────────────
    if require_head_sha or (
        expected_head_sha is not None and live_head_sha is not None
    ):
        exp = (expected_head_sha or "").strip().lower()
        live = (live_head_sha or "").strip().lower()
        mismatch = False
        reasons: list[str] = []
        if require_head_sha and not exp:
            mismatch = True
            reasons.append(
                "expected_head_sha is required before this mutation "
                "(anti-stomp fail closed)"
            )
        elif exp and live and exp != live:
            mismatch = True
            reasons.append(
                f"expected_head_sha '{exp}' does not match live PR head "
                f"'{live}' (anti-stomp fail closed; stale prompt data)"
            )
        elif require_head_sha and exp and not live:
            mismatch = True
            reasons.append(
                "live PR head SHA could not be determined to corroborate "
                "expected_head_sha (anti-stomp fail closed)"
            )
        checks["head_sha"] = {
            "block": mismatch,
            "reasons": reasons,
            "expected_head_sha": expected_head_sha,
            "live_head_sha": live_head_sha,
        }
        if mismatch:
            blockers.append(_blocker(BLOCKER_HEAD_SHA, reasons))
    else:
        checks["head_sha"] = {"block": False, "skipped": True}
    # ── workflow hash ────────────────────────────────────────────────────────
    if workflow_hash_valid is not None:
        reasons = list(workflow_hash_reasons or [])
        if not workflow_hash_valid and not reasons:
            reasons = [
                "workflow load proof missing or hash stale "
                "(anti-stomp fail closed)"
            ]
        checks["workflow_hash"] = {
            "block": not bool(workflow_hash_valid),
            "reasons": reasons if not workflow_hash_valid else [],
        }
        if not workflow_hash_valid:
            blockers.append(_blocker(BLOCKER_WORKFLOW_HASH, reasons))
    else:
        checks["workflow_hash"] = {"block": False, "skipped": True}
    # ── source contamination ─────────────────────────────────────────────────
    if source_contaminated is not None:
        reasons = list(contamination_reasons or [])
        if source_contaminated and not reasons:
            reasons = [
                "session is source-contaminated (stable-branch push or "
                "contaminated approval path); anti-stomp fail closed"
            ]
        checks["source_contamination"] = {
            "block": bool(source_contaminated),
            "reasons": reasons if source_contaminated else [],
        }
        if source_contaminated:
            blockers.append(
                _blocker(BLOCKER_SOURCE_CONTAMINATION, reasons)
            )
    else:
        checks["source_contamination"] = {"block": False, "skipped": True}
    if blockers:
        return _blocked_result(blockers, checks)
    return _allowed_result(checks)
 def format_anti_stomp_error(assessment: dict[str, Any]) -> str:
    """Single RuntimeError message for MCP mutation gates."""
    kind = assessment.get("blocker_kind") or "unknown"
    reasons = "; ".join(
        assessment.get("reasons")
        or ["anti-stomp preflight blocked the mutation"]
    )
    next_action = assessment.get("exact_next_action") or "stop and diagnose"
    return (
        f"Anti-stomp preflight (#604) blocked mutation "
        f"[{kind}]: {reasons}. "
        f"exact_next_action: {next_action}"
    )
 def block_response(
    assessment: dict[str, Any],
    **extra_fields: Any,
 ) -> dict[str, Any]:
    """Structured tool-return payload for a blocked mutation."""
    payload = {
        "success": False,
        "performed": False,
        "blocked": True,
        "anti_stomp": True,
        "blocker_kind": assessment.get("blocker_kind"),
        "blockers": list(assessment.get("blockers") or []),
        "reasons": list(assessment.get("reasons") or []),
        "exact_next_action": assessment.get("exact_next_action"),
        "checks": assessment.get("checks") or {},
    }
    payload.update(extra_fields)
    return payload
 def contamination_from_stable_marker(
    marker: dict | None,
    *,
    task: str | None,
    actual_role: str | None,
 ) -> tuple[bool, list[str]]:
    """Derive source-contamination facts from a #671 durable marker."""
    if not marker:
        return False, []
    gate = stable_branch_push_guard.assess_contamination_gate(
        marker,
        task=task,
        actual_role=actual_role,
    )
    if gate.get("block"):
        return True, list(gate.get("reasons") or [])
    return False, []
@@ -619,188 +619,6 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
        )
 def _anti_stomp_in_test_mode() -> bool:
    """Whether the #604 anti-stomp gate is skipped under pytest.
    Mirrors remote-repo / stable-contamination test bypasses so the existing
    suite (bare remotes, mocked APIs) stays green. Set
    ``GITEA_TEST_FORCE_ANTI_STOMP=1`` to exercise the live gate under tests.
    """
    if not _preflight_in_test_mode():
        return False
    return not bool(os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP"))
 def _run_anti_stomp_preflight(
    task: str | None,
    *,
    remote: str | None = None,
    worktree_path: str | None = None,
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    expected_head_sha: str | None = None,
    live_head_sha: str | None = None,
    require_head_sha: bool = False,
    lease_required: bool = False,
    foreign_lease: bool | None = None,
    lease_reasons: list[str] | None = None,
    terminal_lock_blocks: bool | None = None,
    terminal_lock_reasons: list[str] | None = None,
    workflow_hash_valid: bool | None = None,
    workflow_hash_reasons: list[str] | None = None,
    raise_on_block: bool = True,
 ) -> dict | None:
    """Shared #604 anti-stomp preflight for MCP mutation entrypoints.
    Gathers live workspace/parity/role/repo facts and invokes the pure
    :func:`anti_stomp_preflight.assess_anti_stomp_preflight` assessor. On
    block, raises ``RuntimeError`` (default) or returns a structured
    :func:`anti_stomp_preflight.block_response` when *raise_on_block* is
    False. Returns ``None`` when the mutation may proceed.
    """
    if _anti_stomp_in_test_mode():
        return None
    # Only enforce when the caller names a known mutation task. Read-only
    # paths and legacy verify_preflight_purity calls without a task stay
    # on the pre-#604 gate chain (whoami/capability/root/worktree).
    if not task or not anti_stomp_preflight.is_mutation_task(task):
        return None
    profile = get_profile()
    profile_name = profile.get("profile_name")
    profile_role = _actual_profile_role()
    allowed_operations = list(profile.get("allowed_operations") or [])
    req_role = None
    req_perm = None
    if task:
        try:
            req_role = task_capability_map.required_role(task)
        except KeyError:
            req_role = _preflight_resolved_role
        try:
            req_perm = task_capability_map.required_permission(task)
        except KeyError:
            req_perm = None
    ctx = _resolve_namespace_mutation_context(worktree_path)
    workspace = ctx["workspace_path"]
    canonical_root = ctx["canonical_repo_root"]
    git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
    remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
    # Repo/org facts (best-effort; explicit org/repo when provided).
    resolved_org = org
    resolved_repo = repo
    local_remote_url = None
    org_explicit = org is not None
    repo_explicit = repo is not None
    if remote:
        try:
            local_remote_url = _local_git_remote_url(remote)
        except Exception:
            local_remote_url = None
        if resolved_org is None or resolved_repo is None:
            try:
                rem = _effective_remote(remote)
                if rem in REMOTES:
                    if resolved_org is None:
                        resolved_org = REMOTES[rem]["org"]
                    if resolved_repo is None:
                        resolved_repo = REMOTES[rem]["repo"]
            except Exception:
                pass
    parity = _current_master_parity()
    startup_head = parity.get("startup_head")
    current_code_head = parity.get("current_head")
    # Source contamination from durable #671 marker (role-aware).
    source_contaminated = None
    contamination_reasons = None
    try:
        marker = _load_stable_contamination_marker(remote)
        contaminated, cont_reasons = anti_stomp_preflight.contamination_from_stable_marker(
            marker,
            task=task,
            actual_role=profile_role,
        )
        if marker is not None:
            source_contaminated = contaminated
            contamination_reasons = cont_reasons
    except Exception:
        # Fail soft on marker I/O: existing #671 enforcer still runs separately.
        source_contaminated = None
    # Workflow-hash facts when the task is review/merge oriented.
    if workflow_hash_valid is None and task in {
        "review_pr",
        "submit_pr_review",
        "approve_pr",
        "request_changes_pr",
        "merge_pr",
    }:
        try:
            wf_reasons = _review_workflow_load_gate_reasons()
            workflow_hash_valid = not bool(wf_reasons)
            workflow_hash_reasons = list(wf_reasons) if wf_reasons else None
        except Exception:
            pass
    # Mirror remote_repo_guard's pytest bypass: under the unit suite bare
    # remote=prgs still defaults to Timesheet, and re-checking here would
    # break happy-path reconciler/author tests that already rely on the
    # #530 gate being opt-in via GITEA_FORCE_REMOTE_REPO_CHECK.
    check_repo = True
    if "pytest" in sys.modules and not os.environ.get(
        "GITEA_FORCE_REMOTE_REPO_CHECK"
    ):
        check_repo = False
    assessment = anti_stomp_preflight.assess_anti_stomp_preflight(
        task=task,
        remote=remote,
        resolved_org=resolved_org,
        resolved_repo=resolved_repo,
        local_remote_url=local_remote_url,
        org_explicit=org_explicit,
        repo_explicit=repo_explicit,
        check_repo=check_repo,
        profile_name=profile_name,
        profile_role=profile_role,
        required_role=req_role or _preflight_resolved_role,
        required_permission=req_perm,
        allowed_operations=allowed_operations,
        workspace_path=workspace,
        project_root=canonical_root,
        current_branch=git_state.get("current_branch"),
        root_head_sha=git_state.get("head_sha"),
        root_porcelain=git_state.get("porcelain_status") or "",
        remote_master_sha=remote_master_sha,
        startup_head=startup_head,
        current_code_head=current_code_head,
        lease_required=lease_required,
        foreign_lease=foreign_lease,
        lease_reasons=lease_reasons,
        terminal_lock_blocks=terminal_lock_blocks,
        terminal_lock_reasons=terminal_lock_reasons,
        require_head_sha=require_head_sha,
        expected_head_sha=expected_head_sha,
        live_head_sha=live_head_sha,
        workflow_hash_valid=workflow_hash_valid,
        workflow_hash_reasons=workflow_hash_reasons,
        source_contaminated=source_contaminated,
        contamination_reasons=contamination_reasons,
        manual_bypass_attempted=False,
    )
    if not assessment.get("block"):
        return None
    if raise_on_block:
        raise RuntimeError(anti_stomp_preflight.format_anti_stomp_error(assessment))
    return anti_stomp_preflight.block_response(assessment)
 def verify_preflight_purity(
    remote: str | None = None,
    worktree_path: str | None = None,
@@ -808,29 +626,13 @@ def verify_preflight_purity(
    *,
    target_issue_number: int | None = None,
    require_author_lock: bool = False,
    expected_head_sha: str | None = None,
    live_head_sha: str | None = None,
    require_head_sha: bool = False,
    lease_required: bool = False,
    foreign_lease: bool | None = None,
    lease_reasons: list[str] | None = None,
    terminal_lock_blocks: bool | None = None,
    terminal_lock_reasons: list[str] | None = None,
    workflow_hash_valid: bool | None = None,
    workflow_hash_reasons: list[str] | None = None,
    org: str | None = None,
    repo: str | None = None,
 ):
-    """Verify identity/capability order, production workspace guards, then anti-stomp.
+    """Verify identity/capability order, then production workspace guards.
    #683: pytest/unittest must not skip production root/branches/scope
    enforcement when force-on signals request production behavior. The
    early return below only skips *preflight-order* purity checks under
    pure unit-test isolation — never when production guards are active.
    #604: also runs the shared anti-stomp preflight for mutation tasks (typed
    blocker + exact next action). Extended lease/head/terminal facts may be
    supplied by review/merge callers.
    """
    global _preflight_reviewer_violation_files
@@ -841,12 +643,7 @@ def verify_preflight_purity(
    # Pure unit-test isolation: skip purity-order unless legacy dirty/porcelain
    # force flags request the dirtiness path. #683 FORCE_PRODUCTION_GUARDS alone
    # runs production root/branches/scope without requiring whoami/capability.
-    force_anti_stomp = bool(os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP"))
+    skip_purity_order = in_test and not workflow_scope_guard.purity_order_forced()
    skip_purity_order = (
        in_test
        and not workflow_scope_guard.purity_order_forced()
        and not force_anti_stomp
    )
    if not skip_purity_order:
        if not _preflight_whoami_called:
@@ -941,25 +738,6 @@ def verify_preflight_purity(
            target_issue_number=target_issue_number,
            require_author_lock=require_author_lock,
        )
        # #604: common anti-stomp preflight after legacy + #683 enforcers.
        _run_anti_stomp_preflight(
            task,
            remote=remote,
            worktree_path=worktree_path,
            org=org,
            repo=repo,
            expected_head_sha=expected_head_sha,
            live_head_sha=live_head_sha,
            require_head_sha=require_head_sha,
            lease_required=lease_required,
            foreign_lease=foreign_lease,
            lease_reasons=lease_reasons,
            terminal_lock_blocks=terminal_lock_blocks,
            terminal_lock_reasons=terminal_lock_reasons,
            workflow_hash_valid=workflow_hash_valid,
            workflow_hash_reasons=workflow_hash_reasons,
            raise_on_block=True,
        )
        _clear_preflight_capability_state()
        return
@@ -974,25 +752,6 @@ def verify_preflight_purity(
            target_issue_number=target_issue_number,
            require_author_lock=require_author_lock,
        )
        if force_anti_stomp:
            _run_anti_stomp_preflight(
                task,
                remote=remote,
                worktree_path=worktree_path,
                org=org,
                repo=repo,
                expected_head_sha=expected_head_sha,
                live_head_sha=live_head_sha,
                require_head_sha=require_head_sha,
                lease_required=lease_required,
                foreign_lease=foreign_lease,
                lease_reasons=lease_reasons,
                terminal_lock_blocks=terminal_lock_blocks,
                terminal_lock_reasons=terminal_lock_reasons,
                workflow_hash_valid=workflow_hash_valid,
                workflow_hash_reasons=workflow_hash_reasons,
                raise_on_block=True,
            )
 def _session_issue_lock_snapshot(
@@ -1360,7 +1119,6 @@ import root_checkout_guard  # noqa: E402
 import workflow_scope_guard  # noqa: E402  # #683 production scope / force-on guards
 import stable_branch_push_guard  # noqa: E402
 import remote_repo_guard  # noqa: E402
 import anti_stomp_preflight  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
@@ -4472,30 +4230,6 @@ def _evaluate_pr_review_submission(
        result["pr_work_lease"] = lease_block
        return result
    if live:
        # #604: common anti-stomp with live head + lease proof (typed blocker).
        anti_task = {
            "approve": "approve_pr",
            "request_changes": "request_changes_pr",
            "comment": "submit_pr_review",
        }.get(action, "submit_pr_review")
        try:
            _run_anti_stomp_preflight(
                anti_task,
                remote=remote,
                worktree_path=worktree_path,
                org=org,
                repo=repo,
                expected_head_sha=pinned_sha,
                live_head_sha=actual_sha,
                require_head_sha=True,
                foreign_lease=False,
                terminal_lock_blocks=False,
            )
        except RuntimeError as exc:
            reasons.append(str(exc))
            return result
    if (body or "").strip():
        gate = _canonical_comment_gate(body, context="pr_review")
        if gate["blocked"]:
@@ -6775,12 +6509,7 @@ def gitea_edit_pr(
        raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
    closing = payload.get("state") == "closed"
-    # Closing uses close_pr; non-closing title/body/base edits use edit_pr so
+    verify_preflight_purity(remote, task="close_pr" if closing else None)
    # the shared #604 anti-stomp inventory stays consistent with wiring.
    if closing:
        verify_preflight_purity(remote, task="close_pr")
    else:
        verify_preflight_purity(remote, task="edit_pr")
    # PR closure is a first-class capability, distinct from retitling or
    # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
@@ -7272,25 +7001,6 @@ def gitea_merge_pr(
        reasons.extend(lease_block.get("reasons") or [])
        result["pr_work_lease"] = lease_block
        return result
    # #604: common anti-stomp with live head + lease proof (typed blocker).
    try:
        _run_anti_stomp_preflight(
            "merge_pr",
            remote=remote,
            worktree_path=worktree_path,
            org=org,
            repo=repo,
            expected_head_sha=expected_head_sha,
            live_head_sha=actual_sha,
            require_head_sha=True,
            foreign_lease=False,
            terminal_lock_blocks=False,
        )
    except RuntimeError as exc:
        reasons.append(str(exc))
        return result
    if expected_head_sha and actual_sha and expected_head_sha != actual_sha:
        reasons.append(
            "expected head SHA does not match current PR head (fail closed)"
@@ -10167,11 +9877,7 @@ def gitea_acquire_reviewer_pr_lease(
            "permission_report": _permission_block_report("gitea.pr.comment"),
        }
-    # task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604
+    _verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
    # anti-stomp for the declared lease-acquire mutation inventory entry.
    _verify_role_mutation_workspace(
        remote, worktree=worktree, task="acquire_reviewer_pr_lease"
    )
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    profile = get_profile()
@@ -10312,7 +10018,6 @@ def gitea_adopt_merger_pr_lease(
            "permission_report": _permission_block_report("gitea.pr.merge"),
        }
    # task=adopt_merger_pr_lease so verify_preflight_purity runs shared #604 anti-stomp.
    _verify_role_mutation_workspace(
        remote, worktree=worktree, task="adopt_merger_pr_lease"
    )
@@ -13954,36 +13659,18 @@ def gitea_route_task_session(
    )
-_restart_triggered = False
+# #685: resolver stale-runtime detection is report-only. Config touch / os._exit
-
+# self-recovery was removed from the read-only path (was _trigger_mcp_auto_restart).
-
+# Recovery is owned exclusively by the IDE/client reconnect path.
 def _trigger_mcp_auto_restart():
    global _restart_triggered
    if _restart_triggered or _preflight_in_test_mode():
        return
    _restart_triggered = True
    config_path = os.environ.get(
        "MCP_CONFIG_PATH",
        os.path.expanduser("~/.gemini/config/mcp_config.json")
    )
    try:
        if os.path.exists(config_path):
            os.utime(config_path, None)
    except Exception:
        pass
    import threading
    import time
    def delayed_exit():
        time.sleep(1.0)
        os._exit(0)
    threading.Thread(target=delayed_exit, daemon=True).start()
 def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> list[str]:
-    """Check running runtimes and return errors if they are missing or stale."""
+    """Read-only: report missing or stale MCP runtimes (no config or process mutation).
    #685: Never touches MCP client config, never spawns recovery threads, never
    calls ``os._exit``. Stale detection remains fail-closed via returned reasons
    only; the IDE/client owns reconnect/reload.
    """
    import subprocess
    import re
    from datetime import datetime
@@ -14063,11 +13750,13 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
            }
    if self_stale:
-        _trigger_mcp_auto_restart()
+        # #685: report-only — no config utime, no thread, no os._exit.
        reasons.append(
-            "stale-runtime: The active Gitea MCP server process is stale (running code from before changes were merged). "
+            "stale-runtime: The active Gitea MCP server process is stale "
-            "Auto-restart has been triggered: touched mcp_config.json to reload the daemon. "
+            "(running code from before changes were merged). "
-            "The current process will cleanly exit shortly."
+            "Reconnect the IDE/client-managed MCP namespace for this profile "
            "so it reloads current master. The resolver does not touch "
            "mcp_config.json, spawn recovery threads, or terminate this process."
        )
    if matching_profiles:
@@ -14099,10 +13788,15 @@ def gitea_resolve_task_capability(
    remote: str = "dadeschools",
    host: str | None = None,
 ) -> dict:
-    """Read-only: Resolve which capability, profile, and namespace is required for a Gitea task.
+    """Read-only / side-effect free: resolve capability, profile, and namespace for a task.
-    Helps the client or LLM determine the correct namespace or profile before acting,
+    Does **not** mutate MCP client configuration, spawn recovery threads, kill
-    and returns exact next action instructions if the current session is not authorized.
+    processes, or trigger daemon reloads (#685). Stale-runtime detection remains
    fail-closed: when the serving process is stale the result includes
    ``blocker_kind=runtime_reconnect_required``, ``restart_required=true``,
    ``stop_required=true``, and ``mutation_performed=false`` with a precise
    ``exact_safe_next_action`` pointing at IDE/client reconnect. Recovery is
    owned by the client reconnect path — never by this resolver.
    Args:
        task: The task/action to check (e.g. review_pr, create_issue).
@@ -14287,31 +13981,32 @@ def gitea_resolve_task_capability(
    configured = len(matching_profiles) > 0
    available_in_session = allowed_in_current_session
    runtime_stale_blocker = False
    if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
        runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
        if runtime_reasons:
            restart_required = True
            runtime_stale_blocker = True
            reason_msg = "; ".join(runtime_reasons)
            next_safe_action = (
                "stale-runtime: Gitea MCP runtime conflict or missing process detected. "
                "Please fully restart the Gitea MCP server and retry."
            )
    if not allowed_in_current_session:
        if configured and switching:
            restart_required = True
            available_in_session = False
-            reason_msg = (
+            if not reason_msg:
-                f"{required_role.capitalize()} profile exists but MCP server "
+                reason_msg = (
-                "was added after session startup and is not attached."
+                    f"{required_role.capitalize()} profile exists but MCP server "
-            )
+                    "was added after session startup and is not attached."
                )
        elif not configured:
-            reason_msg = (
+            if not reason_msg:
-                f"No profile configured with permission '{required_permission}'."
+                reason_msg = (
-            )
+                    f"No profile configured with permission '{required_permission}'."
                )
        elif role_mismatch_reason:
-            reason_msg = role_mismatch_reason
+            if not reason_msg:
                reason_msg = role_mismatch_reason
    different_namespace_required = False
    next_safe_action = "None; ready for operations."
@@ -14337,6 +14032,16 @@ def gitea_resolve_task_capability(
                    "or use the corresponding MCP namespace."
                )
    # #685: stale-runtime typed remediation wins for exact_next_action when the
    # serving process/profile inventory is stale — even if permission is OK.
    if runtime_stale_blocker:
        next_safe_action = (
            "blocker_kind=runtime_reconnect_required: reconnect/restart the "
            "IDE-managed Gitea MCP server for this profile so it reloads current "
            "master. Do not edit mcp_config.json by hand; the resolver does not "
            "touch config, spawn recovery threads, or terminate the process."
        )
    # Task/role alignment guards (#167): the requested task, not the
    # available credential, decides what the session may do. A review/merge
    # task under a non-reviewer profile must stop — not silently degrade
@@ -14398,12 +14103,16 @@ def gitea_resolve_task_capability(
        "configured": configured,
        "restart_required": restart_required,
        "stop_required": stop_required or restart_required,
        # #685: resolver is always side-effect free; never claims mutations.
        "mutation_performed": False,
        "task_role_guidance": task_role_guidance,
        "matching_configured_profile": matching_profiles,
        "runtime_switching_supported": switching,
        "different_mcp_namespace_required": different_namespace_required,
        "exact_safe_next_action": next_safe_action,
    }
    if runtime_stale_blocker:
        result["blocker_kind"] = "runtime_reconnect_required"
    if reason_msg:
        result["reason"] = reason_msg
    if task in ("review_pr", "merge_pr"):
@@ -60,15 +60,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
        "permission": "gitea.pr.close",
        "role": "author",
    },
    # Non-closing PR metadata edits (title/body/base). Closing uses close_pr.
    "edit_pr": {
        "permission": "gitea.pr.create",
        "role": "author",
    },
    "gitea_edit_pr": {
        "permission": "gitea.pr.create",
        "role": "author",
    },
    "address_pr_change_requests": {
        "permission": "gitea.branch.push",
        "role": "author",
@@ -77,22 +68,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
        "permission": "gitea.pr.review",
        "role": "reviewer",
    },
    "submit_pr_review": {
        "permission": "gitea.pr.review",
        "role": "reviewer",
    },
    "merge_pr": {
        "permission": "gitea.pr.merge",
        "role": "merger",
    },
    "acquire_reviewer_pr_lease": {
        "permission": "gitea.pr.comment",
        "role": "reviewer",
    },
    "gitea_acquire_reviewer_pr_lease": {
        "permission": "gitea.pr.comment",
        "role": "reviewer",
    },
    # #695 AC8: controller quarantine of contaminated formal reviews.
    # Apply path posts an append-only forensic audit comment (pr.comment).
    "quarantine_contaminated_review": {
@@ -0,0 +1,326 @@
 """#685: gitea_resolve_task_capability must be side-effect free.
 Stale-runtime detection remains fail-closed, but the resolver must never:
  * touch mcp_config.json (or any MCP client config)
  * spawn recovery threads
  * call os._exit / terminate the serving process
  * claim that an auto-restart was triggered
 Recovery is owned by the IDE/client reconnect path only.
 """
 from __future__ import annotations
 import os
 import tempfile
 import threading
 import unittest
 from datetime import datetime
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 import sys
 ROOT = str(Path(__file__).resolve().parent.parent)
 if ROOT not in sys.path:
    sys.path.insert(0, ROOT)
 import gitea_mcp_server as mcp_server
 ROLE_PROFILES = (
    ("create_issue", "prgs-author", "author"),
    ("review_pr", "prgs-reviewer", "reviewer"),
    ("merge_pr", "prgs-merger", "merger"),
    ("reconciliation_cleanup", "prgs-reconciler", "reconciler"),
 )
 def _stale_self_ps_mocks(profile: str = "prgs-author"):
    """Build subprocess mocks: self PID is stale vs code mtime."""
    mock_getpid = MagicMock(return_value=12345)
    mock_exists = MagicMock(return_value=True)
    code_time = datetime(2026, 7, 8, 14, 0, 0)
    mock_getmtime = MagicMock(return_value=code_time.timestamp())
    ps_output = (
        "  PID LSTART COMMAND\n"
        "12345 Wed Jul  8 13: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 = f"GITEA_MCP_PROFILE={profile}"
    mock_run_git = MagicMock()
    mock_run_git.stdout = "SAME"
    def side_effect(args, **kwargs):
        if args[0] == "ps" and "eww" in args:
            return mock_run_env
        if args[0] == "ps":
            return mock_run_ps
        if args[0] == "git":
            return mock_run_git
        raise ValueError(f"Unexpected subprocess args: {args}")
    mock_run = MagicMock(side_effect=side_effect)
    return mock_getpid, mock_exists, mock_getmtime, mock_run
 class TestIssue685DiagnosticsNoSideEffects(unittest.TestCase):
    def setUp(self):
        mcp_server._process_boot_head_sha = None
    def tearDown(self):
        mcp_server._process_boot_head_sha = None
    @patch.dict(os.environ, {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"}, clear=False)
    @patch("subprocess.run")
    @patch("os.path.getmtime")
    @patch("os.path.exists")
    @patch("os.getpid")
    @patch("os.utime")
    @patch("threading.Thread")
    @patch("os._exit")
    def test_stale_self_does_not_touch_config_or_exit(
        self,
        mock_exit,
        mock_thread,
        mock_utime,
        mock_getpid,
        mock_exists,
        mock_getmtime,
        mock_run,
    ):
        mock_getpid.return_value = 12345
        mock_exists.return_value = True
        mock_getmtime.return_value = datetime(2026, 7, 8, 14, 0, 0).timestamp()
        mock_run.side_effect = _stale_self_ps_mocks("prgs-author")[3].side_effect
        before_threads = threading.active_count()
        reasons = mcp_server._check_mcp_runtimes_diagnostics(
            "create_issue", ["prgs-author"]
        )
        after_threads = threading.active_count()
        self.assertTrue(
            any("stale-runtime" in r and "active Gitea MCP server process is stale" in r
                for r in reasons),
            reasons,
        )
        # Must not claim auto-restart / config touch
        blob = " ".join(reasons)
        self.assertNotIn("Auto-restart has been triggered", blob)
        self.assertNotIn("touched mcp_config", blob)
        self.assertNotIn("will cleanly exit", blob)
        mock_utime.assert_not_called()
        mock_thread.assert_not_called()
        mock_exit.assert_not_called()
        self.assertEqual(before_threads, after_threads)
    @patch.dict(os.environ, {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"}, clear=False)
    @patch("subprocess.run")
    @patch("os.path.getmtime")
    @patch("os.path.exists")
    @patch("os.getpid")
    @patch("os.utime")
    def test_repeated_stale_calls_do_not_trigger_restart_loop(
        self, mock_utime, mock_getpid, mock_exists, mock_getmtime, mock_run
    ):
        mock_getpid.return_value = 12345
        mock_exists.return_value = True
        mock_getmtime.return_value = datetime(2026, 7, 8, 14, 0, 0).timestamp()
        mock_run.side_effect = _stale_self_ps_mocks("prgs-author")[3].side_effect
        for _ in range(5):
            reasons = mcp_server._check_mcp_runtimes_diagnostics(
                "create_issue", ["prgs-author"]
            )
            self.assertTrue(any("stale-runtime" in r for r in reasons))
        mock_utime.assert_not_called()
    def test_trigger_mcp_auto_restart_removed(self):
        """#685 AC: auto-restart helper is removed (unreachable from read-only)."""
        self.assertFalse(hasattr(mcp_server, "_trigger_mcp_auto_restart"))
        self.assertFalse(hasattr(mcp_server, "_restart_triggered"))
 class TestIssue685ResolverTypedBlocker(unittest.TestCase):
    def setUp(self):
        mcp_server._process_boot_head_sha = None
    def tearDown(self):
        mcp_server._process_boot_head_sha = None
        if hasattr(mcp_server, "capability_stop_terminal"):
            mcp_server.capability_stop_terminal.clear()
    def _resolve_with_stale_runtime(self, task: str, profile_name: str, role: str):
        allowed = [
            "gitea.read",
            "gitea.issue.create",
            "gitea.issue.comment",
            "gitea.issue.close",
            "gitea.branch.create",
            "gitea.branch.push",
            "gitea.branch.delete",
            "gitea.pr.create",
            "gitea.pr.comment",
            "gitea.pr.review",
            "gitea.pr.approve",
            "gitea.pr.request_changes",
            "gitea.pr.merge",
            "gitea.pr.close",
            "gitea.repo.commit",
        ]
        profile = {
            "profile_name": profile_name,
            "role": role,
            "allowed_operations": allowed,
            "forbidden_operations": [],
        }
        config = {
            "profiles": {
                profile_name: {
                    "role": role,
                    "allowed_operations": allowed,
                    "forbidden_operations": [],
                }
            }
        }
        mock_getpid, mock_exists, mock_getmtime, mock_run = _stale_self_ps_mocks(
            profile_name
        )
        with patch.dict(
            os.environ,
            {
                "GITEA_FORCE_MCP_RUNTIME_CHECK": "1",
                "GITEA_MCP_PROFILE": profile_name,
            },
            clear=False,
        ), patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
            mcp_server.gitea_config, "load_config", return_value=config
        ), patch.object(
            mcp_server, "_authenticated_username", return_value="test-user"
        ), patch.object(
            mcp_server, "_ensure_matching_profile", return_value=None
        ), patch.object(
            mcp_server, "record_preflight_check", return_value=None
        ), patch.object(
            mcp_server, "record_mutation_authority", return_value=None
        ), patch.object(
            mcp_server, "init_review_decision_lock", return_value=None
        ), patch(
            "subprocess.run", mock_run
        ), patch(
            "os.path.getmtime", mock_getmtime
        ), patch(
            "os.path.exists", mock_exists
        ), patch(
            "os.getpid", mock_getpid
        ), patch(
            "os.utime"
        ) as mock_utime, patch(
            "threading.Thread"
        ) as mock_thread, patch(
            "os._exit"
        ) as mock_exit:
            result = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs")
            return result, mock_utime, mock_thread, mock_exit
    def test_stale_returns_typed_blocker_fields(self):
        result, mock_utime, mock_thread, mock_exit = self._resolve_with_stale_runtime(
            "create_issue", "prgs-author", "author"
        )
        self.assertTrue(result.get("restart_required"), result)
        self.assertTrue(result.get("stop_required"), result)
        self.assertEqual(result.get("blocker_kind"), "runtime_reconnect_required")
        self.assertIs(result.get("mutation_performed"), False)
        action = result.get("exact_safe_next_action") or ""
        self.assertIn("reconnect", action.lower())
        self.assertNotIn("None; ready for operations", action)
        reason = result.get("reason") or ""
        self.assertIn("stale-runtime", reason)
        self.assertNotIn("Auto-restart has been triggered", reason)
        mock_utime.assert_not_called()
        mock_thread.assert_not_called()
        mock_exit.assert_not_called()
    def test_all_four_role_profiles_get_same_side_effect_free_contract(self):
        for task, profile, role in ROLE_PROFILES:
            with self.subTest(task=task, profile=profile):
                # Skip tasks that may be unknown on this branch
                try:
                    import task_capability_map as tcm
                    tcm.required_permission(task)
                except Exception:
                    self.skipTest(f"task {task} not in capability map")
                result, mock_utime, mock_thread, mock_exit = (
                    self._resolve_with_stale_runtime(task, profile, role)
                )
                self.assertTrue(
                    result.get("restart_required") or result.get("stop_required"),
                    result,
                )
                self.assertEqual(
                    result.get("blocker_kind"), "runtime_reconnect_required", result
                )
                self.assertIs(result.get("mutation_performed"), False, result)
                mock_utime.assert_not_called()
                mock_thread.assert_not_called()
                mock_exit.assert_not_called()
    def test_config_mtime_and_contents_unchanged(self):
        with tempfile.TemporaryDirectory() as tmp:
            cfg = os.path.join(tmp, "mcp_config.json")
            original = '{"servers": {"gitea-author": {}}}'
            with open(cfg, "w", encoding="utf-8") as fh:
                fh.write(original)
            mtime_before = os.path.getmtime(cfg)
            result, mock_utime, mock_thread, mock_exit = self._resolve_with_stale_runtime(
                "create_issue", "prgs-author", "author"
            )
            # Force-path also must not use real utime when diagnostics runs
            with open(cfg, encoding="utf-8") as fh:
                after = fh.read()
            self.assertEqual(after, original)
            self.assertEqual(os.path.getmtime(cfg), mtime_before)
            mock_utime.assert_not_called()
            self.assertTrue(result.get("restart_required"), result)
 class TestIssue685MutationGatesStillFailClosed(unittest.TestCase):
    def test_parity_stale_still_reports_restart_required(self):
        """Mutation-facing parity gate remains fail-closed when heads differ."""
        import master_parity_gate as mpg
        out = mpg.assess_master_parity(
            {"startup_head": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
            "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
        )
        self.assertFalse(out.get("in_parity"))
        self.assertTrue(out.get("restart_required") or out.get("stale"))
 class TestIssue685DocstringReadOnlyContract(unittest.TestCase):
    def test_resolve_docstring_declares_side_effect_free(self):
        doc = mcp_server.gitea_resolve_task_capability.__doc__ or ""
        lower = doc.lower()
        self.assertTrue(
            "side-effect" in lower or "read-only" in lower or "does not mutate" in lower,
            doc,
        )
        self.assertNotIn("auto-restart", lower)
        self.assertNotIn("os._exit", lower)
 if __name__ == "__main__":
    unittest.main()
@@ -111,21 +111,13 @@ class TestMcpStaleRuntime(unittest.TestCase):
        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")
+    def test_auto_restart_helper_removed_from_read_only_path(self):
-    @patch("os.utime")
+        """#685: config-touch / os._exit self-recovery is no longer on the server."""
-    @patch("os.path.exists")
+        self.assertFalse(
-    @patch.dict("os.environ", {"MCP_CONFIG_PATH": "/tmp/mcp_config.json"})
+            hasattr(gitea_mcp_server, "_trigger_mcp_auto_restart"),
-    def test_auto_restart_trigger_touches_and_spawns(self, mock_exists, mock_utime, mock_thread):
+            "_trigger_mcp_auto_restart must not remain (side-effect-free resolver)",
-        mock_exists.return_value = True
+        )
-        gitea_mcp_server._restart_triggered = False
+        self.assertFalse(hasattr(gitea_mcp_server, "_restart_triggered"))
        # 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__":