Gitea-Tools/mcp_restart_paths.py

"""Inventory and fail-closed guards for MCP restart/reload/kill paths (#657).

Single source of truth enumerating every code/script/doc path that can
restart, reload, reconnect, kill, or force-recreate an MCP process. Each path
is classified and linked to the guard that constrains it. The companion
human-readable inventory lives in ``docs/mcp-restart-path-inventory.md`` and is
kept in lock-step with this module by ``tests/test_mcp_restart_paths.py``.

Design intent (aligns with #655 restart-coordinator roadmap):

* **No unguarded full restart.** The in-process MCP daemon
  (``gitea_mcp_server.py`` / ``mcp_server.py`` / ``role_session_router.py``)
  must never replace or kill its own process — replacing the process after the
  host wired up the stdio pipes desyncs the JSON-RPC transport (observed with
  Antigravity/Cascade hosts). ``assert_no_daemon_self_replacement`` enforces
  this against the live source tree.
* **No legacy auto-restart helper.** ``_trigger_mcp_auto_restart`` was removed
  when the stale-runtime resolver became side-effect free (#685);
  ``assert_auto_restart_helper_absent`` keeps it removed.
* **Unknown restart attempts fail closed.** LLM tools must route any restart
  intent through a *registered* path. ``assert_restart_attempt_registered``
  raises ``UnknownRestartPathError`` for anything not in this inventory.
* **pkill stays forbidden (#630).** Manual daemon kills are classified as
  contamination by :mod:`runtime_recovery_guard`; this module records that path
  and the test asserts the classification still holds.

This module performs no restarts, spawns no threads, and touches no config or
process state. It is pure inventory + read-only source assertions.
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable

# --- Classifications -------------------------------------------------------

#: A narrow, one-shot recovery that is safe by construction (e.g. a CLI wrapper
#: re-execing into the venv interpreter before importing anything, or an
#: in-process profile switch). Never targets the running MCP daemon process.
CLASS_SANCTIONED_NARROW = "sanctioned_narrow_recovery"

#: The path detects a condition that would require a restart, then *fails
#: closed* on mutations and emits restart/reconnect guidance. It never restarts
#: the process itself (recovery is owned by the host/operator).
CLASS_GUARDED_FAIL_CLOSED = "guarded_fail_closed"

#: The path is forbidden. Attempting it is a workflow-safety violation and,
#: where an LLM tool could invoke it, is marked as contamination.
CLASS_FORBIDDEN = "forbidden"

#: A previously-existing unguarded restart primitive that has been deleted. A
#: regression guard keeps it absent.
CLASS_REMOVED = "removed"

#: Behavior that lives in the host/IDE and is outside this process's control
#: (e.g. a manual ``/mcp reconnect``). Documented, not code-guarded here.
CLASS_HOST_RESIDUAL = "host_residual"

VALID_CLASSIFICATIONS = frozenset(
    {
        CLASS_SANCTIONED_NARROW,
        CLASS_GUARDED_FAIL_CLOSED,
        CLASS_FORBIDDEN,
        CLASS_REMOVED,
        CLASS_HOST_RESIDUAL,
    }
)

#: The in-process MCP daemon modules. These must never self-replace/self-kill.
DAEMON_MODULES = (
    "gitea_mcp_server.py",
    "mcp_server.py",
    "role_session_router.py",
)

#: The legacy auto-restart helper removed in #685. Must stay removed.
LEGACY_AUTO_RESTART_HELPER = "_trigger_mcp_auto_restart"

#: Call patterns that would let the daemon replace or terminate its own
#: process. Matched as calls (trailing ``(``) so prose/docstring mentions such
#: as "we do NOT os.execv() here" or "never calls ``os._exit``" do not trip the
#: scanner (comment lines are stripped first regardless).
DAEMON_SELF_REPLACEMENT_PRIMITIVES = (
    "os.execv(",
    "os.execve(",
    "os.execvp(",
    "os.execvpe(",
    "os.kill(",
    "os.killpg(",
    "os._exit(",
    "os.abort(",
)


@dataclass(frozen=True)
class RestartPath:
    """One classified restart/reload/kill path in the inventory."""

    path_id: str
    title: str
    mechanism: str
    classification: str
    guard: str
    locations: tuple[str, ...]
    references: tuple[str, ...]
    residual_host: bool = False
    notes: str = ""


class UnknownRestartPathError(RuntimeError):
    """Raised when a restart attempt is not a registered, classified path."""


# --- The inventory ---------------------------------------------------------

_RESTART_PATHS: tuple[RestartPath, ...] = (
    RestartPath(
        path_id="cli_venv_bootstrap_execv",
        title="CLI wrapper venv re-exec",
        mechanism=(
            "Standalone CLI scripts re-exec into venv/bin/python3 via os.execv "
            "at import top, guarded by `sys.executable != venv_python`."
        ),
        classification=CLASS_SANCTIONED_NARROW,
        guard=(
            "One-shot, pre-import bootstrap; runs before any MCP transport "
            "exists and only when not already on the venv interpreter, so it "
            "cannot desync a live daemon. Idempotent guard condition prevents "
            "a re-exec loop."
        ),
        locations=(
            "create_pr.py",
            "create_issue.py",
            "close_issue.py",
            "merge_pr.py",
            "review_pr.py",
            "edit_pr.py",
            "delete_branch.py",
            "mark_issue.py",
            "manage_labels.py",
            "list_issues.py",
            "list_prs.py",
        ),
        references=("#657",),
    ),
    RestartPath(
        path_id="daemon_self_replacement",
        title="MCP daemon self-replacement",
        mechanism=(
            "The in-process MCP daemon replacing/terminating its own process "
            "(os.execv/os.kill/os._exit) to reload code."
        ),
        classification=CLASS_FORBIDDEN,
        guard=(
            "Forbidden by design: replacing the process after the host wired "
            "up stdio desyncs JSON-RPC (Antigravity/Cascade). Enforced against "
            "the source tree by assert_no_daemon_self_replacement()."
        ),
        locations=("gitea_mcp_server.py:~155 (decision comment)",) + DAEMON_MODULES,
        references=("#657", "#584"),
    ),
    RestartPath(
        path_id="legacy_auto_restart_helper",
        title="Legacy _trigger_mcp_auto_restart helper",
        mechanism=(
            "A helper that actively restarted the MCP server from the "
            "read-only resolver path."
        ),
        classification=CLASS_REMOVED,
        guard=(
            "Removed in #685 when the resolver became side-effect free. Kept "
            "absent by assert_auto_restart_helper_absent()."
        ),
        locations=("gitea_mcp_server.py", "mcp_server.py"),
        references=("#685", "#657"),
    ),
    RestartPath(
        path_id="config_touch_reload",
        title="MCP client config-touch reload",
        mechanism=(
            "Touching (utime) the MCP client config file to make the host "
            "reload/recreate the server process."
        ),
        classification=CLASS_REMOVED,
        guard=(
            "Removed from the resolver in #685: stale-runtime detection is "
            "report-only and never mutates client config, spawns threads, or "
            "calls os._exit."
        ),
        locations=("gitea_mcp_server.py (resolve_task_capability)",),
        references=("#685", "#657"),
    ),
    RestartPath(
        path_id="master_advance_auto_restart",
        title="Master-advance staleness gate",
        mechanism=(
            "On-disk master advancing past the running code. The master-parity "
            "gate detects it and fails mutations closed with restart guidance."
        ),
        classification=CLASS_GUARDED_FAIL_CLOSED,
        guard=(
            "Detect + fail closed only; the process never self-restarts. "
            "master_parity_gate captures startup parity and blocks mutations "
            "while stale, emitting restart/reconnect guidance."
        ),
        locations=(
            "master_parity_gate.py",
            "gitea_mcp_server.py (gitea_assess_master_parity)",
        ),
        references=("#420", "#591", "#657"),
    ),
    RestartPath(
        path_id="stale_runtime_resolver_reconnect",
        title="Stale-runtime resolver reconnect guidance",
        mechanism=(
            "The capability resolver detecting a stale serving process and "
            "reporting restart_required/stop_required for a client reconnect."
        ),
        classification=CLASS_GUARDED_FAIL_CLOSED,
        guard=(
            "Report-only (#685): returns restart_required/stop_required and an "
            "exact_safe_next_action pointing at IDE/client reconnect; performs "
            "no restart, thread spawn, config touch, or os._exit."
        ),
        locations=("gitea_mcp_server.py (gitea_resolve_task_capability)",),
        references=("#685", "#657"),
    ),
    RestartPath(
        path_id="manual_daemon_kill",
        title="Manual daemon kill (pkill/killall/kill)",
        mechanism=(
            "Shell kills of the MCP daemon: `pkill -f mcp_server.py`, "
            "`killall`, broad `pkill -f python` sweeps, or `kill <pid>` of a "
            "daemon pid."
        ),
        classification=CLASS_FORBIDDEN,
        guard=(
            "Forbidden (#630): runtime_recovery_guard classifies these as "
            "contamination and gitea_record_daemon_process_kill_attempt writes "
            "a durable marker that fails subsequent mutations closed. Operator "
            "maintenance authorization is read only from the environment, not "
            "from a tool argument."
        ),
        locations=(
            "runtime_recovery_guard.py",
            "gitea_mcp_server.py (gitea_record_daemon_process_kill_attempt)",
        ),
        references=("#630", "#657"),
    ),
    RestartPath(
        path_id="conflict_marker_infra_stop",
        title="Startup conflict-marker infra stop",
        mechanism=(
            "The daemon entrypoint scans for unresolved merge-conflict markers "
            "at startup and stops (sys.exit(1)) if found."
        ),
        classification=CLASS_GUARDED_FAIL_CLOSED,
        guard=(
            "Fail-closed startup stop, not a restart: the process exits and "
            "waits for the operator to resolve conflicts and relaunch. Never "
            "self-restarts or loops."
        ),
        locations=("mcp_server.py (check_conflict_markers)",),
        references=("#657",),
    ),
    RestartPath(
        path_id="ide_client_reconnect",
        title="Host/IDE MCP reconnect",
        mechanism=(
            "A manual `/mcp reconnect` (or equivalent host action) that the "
            "IDE performs to recreate the MCP client connection."
        ),
        classification=CLASS_HOST_RESIDUAL,
        guard=(
            "Outside this process's control. It is the sanctioned recovery the "
            "gates point operators toward; documented as residual host "
            "behavior. No in-process code initiates it."
        ),
        locations=("host/IDE",),
        references=("#584", "#656", "#657"),
        residual_host=True,
    ),
    RestartPath(
        path_id="profile_switch_runtime",
        title="Runtime profile switch",
        mechanism=(
            "Switching the active execution profile at runtime "
            "(dynamic-profile mode)."
        ),
        classification=CLASS_SANCTIONED_NARROW,
        guard=(
            "In-process and restart-free: runtime_switching_supported is true, "
            "so a profile switch rebinds capability without recreating the "
            "process. No restart primitive is invoked."
        ),
        locations=("gitea_mcp_server.py (gitea_activate_profile)",),
        references=("#656", "#657"),
    ),
)

_BY_ID: dict[str, RestartPath] = {p.path_id: p for p in _RESTART_PATHS}


# --- Read-only accessors ---------------------------------------------------


def iter_restart_paths() -> tuple[RestartPath, ...]:
    """Return the full inventory as an immutable tuple."""

    return _RESTART_PATHS


def restart_path_ids() -> frozenset[str]:
    """Return the set of registered path ids."""

    return frozenset(_BY_ID)


def get_restart_path(path_id: str) -> RestartPath:
    """Return the registered path, or raise :class:`UnknownRestartPathError`."""

    try:
        return _BY_ID[path_id]
    except KeyError as exc:
        raise UnknownRestartPathError(
            f"unknown restart path id {path_id!r}; not in the #657 inventory"
        ) from exc


def paths_by_classification(classification: str) -> tuple[RestartPath, ...]:
    """Return all registered paths with the given classification."""

    if classification not in VALID_CLASSIFICATIONS:
        raise ValueError(f"unknown classification {classification!r}")
    return tuple(p for p in _RESTART_PATHS if p.classification == classification)


def assert_restart_attempt_registered(path_id: str) -> RestartPath:
    """Fail closed unless ``path_id`` is a registered, classified restart path.

    LLM tools that intend to trigger any restart/reload/reconnect must name a
    registered path so an unknown/novel restart primitive cannot slip through
    silently. Forbidden and removed paths are registered too — this only
    asserts the attempt is *known*, not that it is *permitted*; callers must
    still honor the classification.
    """

    return get_restart_path(path_id)


def assert_registry_wellformed() -> None:
    """Validate the inventory's own invariants (fail closed on drift)."""

    seen: set[str] = set()
    for path in _RESTART_PATHS:
        if path.path_id in seen:
            raise ValueError(f"duplicate restart path id {path.path_id!r}")
        seen.add(path.path_id)
        if path.classification not in VALID_CLASSIFICATIONS:
            raise ValueError(
                f"{path.path_id!r} has invalid classification "
                f"{path.classification!r}"
            )
        if not path.guard.strip():
            raise ValueError(f"{path.path_id!r} is missing a guard description")
        if not path.references:
            raise ValueError(f"{path.path_id!r} is missing references")
        if not path.locations:
            raise ValueError(f"{path.path_id!r} is missing locations")
        if path.classification == CLASS_HOST_RESIDUAL and not path.residual_host:
            raise ValueError(
                f"{path.path_id!r} is host_residual but residual_host is False"
            )


# --- Source-tree guards ----------------------------------------------------


def _repo_root(root: str | os.PathLike[str] | None = None) -> Path:
    if root is not None:
        return Path(root)
    return Path(__file__).resolve().parent


def _iter_code_lines(text: str) -> Iterable[tuple[int, str]]:
    """Yield (1-based lineno, line) for lines that are not full-line comments."""

    for lineno, line in enumerate(text.splitlines(), start=1):
        if line.lstrip().startswith("#"):
            continue
        yield lineno, line


def scan_daemon_self_replacement(
    root: str | os.PathLike[str] | None = None,
) -> list[dict[str, object]]:
    """Return violations where a daemon module could self-replace/self-kill.

    Scans :data:`DAEMON_MODULES` for calls in
    :data:`DAEMON_SELF_REPLACEMENT_PRIMITIVES`. Full-line comments are ignored,
    and only call forms (with a trailing ``(``) match, so decision comments and
    docstrings that merely mention the primitives do not produce false hits.
    """

    repo = _repo_root(root)
    violations: list[dict[str, object]] = []
    for module in DAEMON_MODULES:
        path = repo / module
        if not path.exists():
            continue
        text = path.read_text(encoding="utf-8", errors="replace")
        for lineno, line in _iter_code_lines(text):
            for primitive in DAEMON_SELF_REPLACEMENT_PRIMITIVES:
                if primitive in line:
                    violations.append(
                        {
                            "module": module,
                            "line": lineno,
                            "primitive": primitive,
                            "text": line.strip(),
                        }
                    )
    return violations


def assert_no_daemon_self_replacement(
    root: str | os.PathLike[str] | None = None,
) -> None:
    """Fail closed if any daemon module can restart/kill its own process."""

    violations = scan_daemon_self_replacement(root)
    if violations:
        rendered = "; ".join(
            f"{v['module']}:{v['line']} {v['primitive']}" for v in violations
        )
        raise AssertionError(
            "MCP daemon must never self-replace/self-kill (#657); found: "
            f"{rendered}"
        )


def scan_auto_restart_helper(
    root: str | os.PathLike[str] | None = None,
) -> list[dict[str, object]]:
    """Return occurrences of a *definition* of the legacy auto-restart helper."""

    repo = _repo_root(root)
    needle = f"def {LEGACY_AUTO_RESTART_HELPER}"
    hits: list[dict[str, object]] = []
    for module in DAEMON_MODULES:
        path = repo / module
        if not path.exists():
            continue
        text = path.read_text(encoding="utf-8", errors="replace")
        for lineno, line in _iter_code_lines(text):
            if needle in line:
                hits.append({"module": module, "line": lineno})
    return hits


def assert_auto_restart_helper_absent(
    root: str | os.PathLike[str] | None = None,
) -> None:
    """Fail closed if the removed ``_trigger_mcp_auto_restart`` reappears."""

    hits = scan_auto_restart_helper(root)
    if hits:
        rendered = "; ".join(f"{h['module']}:{h['line']}" for h in hits)
        raise AssertionError(
            f"{LEGACY_AUTO_RESTART_HELPER} was removed in #685 and must not "
            f"return (#657); found definition at: {rendered}"
        )