Gitea-Tools/mcp_client_reconnect.py

"""Sanctioned MCP client reconnect request surface for Codex/LLM sessions (#678).

Codex and other agent hosts can detect stale or closed Gitea MCP runtimes, but
the host owns the transport. This module never restarts, kills, or reloads a
daemon. It builds:

1. A **callable reconnect request** result agents can invoke via
   ``gitea_request_mcp_reconnect`` (report-only, side-effect free).
2. A **typed blocker** with exact operator UI steps when recovery must be
   performed by the host/operator.

Forbidden recovery paths (must never be recommended):

* ``pkill`` / ``kill`` / ``killall`` of MCP daemons
* ``touch`` / mtime config reload hacks
* ``.env`` or MCP config edits as recovery
* session-state file edits
* raw Gitea API / direct server-import fallbacks

After the operator reconnects, workflows restart from identity / runtime /
capability preflight (``gitea_whoami`` → ``gitea_resolve_task_capability`` →
task).
"""

from __future__ import annotations

from typing import Any, Mapping

# --- Reason vocabulary -------------------------------------------------------

REASON_STALE_RUNTIME = "stale-runtime"
REASON_TRANSPORT_EOF = "transport_eof"
REASON_MISSING_NAMESPACE = "missing_namespace"
REASON_NOT_REQUIRED = "not_required"
REASON_UNSPECIFIED = "unspecified"

VALID_REASONS = frozenset(
    {
        REASON_STALE_RUNTIME,
        REASON_TRANSPORT_EOF,
        REASON_MISSING_NAMESPACE,
        REASON_NOT_REQUIRED,
        REASON_UNSPECIFIED,
    }
)

# Boundary statuses reported to callers (match review_workflow_boundary style).
BOUNDARY_CLEAN = "clean"
BOUNDARY_MISMATCH = "mismatch"
BOUNDARY_STALE = "stale"
BOUNDARY_UNKNOWN = "unknown"

# Typed blocker kinds
BLOCKER_OPERATOR_RECONNECT = "operator_mcp_reconnect_required"
BLOCKER_NONE = "none"

FORBIDDEN_RECOVERY_PATHS: tuple[str, ...] = (
    "pkill / kill / killall of mcp_server.py, gitea_mcp_server, or broad python sweeps",
    "touch / mtime-based MCP config reload hacks",
    ".env edits as recovery",
    "MCP config file edits as recovery",
    "session-state file edits as recovery",
    "raw Gitea API or direct MCP server-import fallbacks",
)

# Client-specific operator UI steps. Keep Codex first (issue title surface).
OPERATOR_UI_STEPS: dict[str, tuple[str, ...]] = {
    "codex": (
        "In Codex, open the MCP / Developer tools panel for this workspace.",
        "Locate the named Gitea MCP server entry (namespace) that needs reconnect "
        "(e.g. gitea-author, gitea-reviewer, gitea-merger, gitea-tools, "
        "gitea-controller, gitea-reconciler).",
        "Click 'Reload Developer Tools' or the server reconnect/reload control "
        "for that entry so the client spawns a fresh MCP subprocess.",
        "If per-server reconnect is unavailable, fully restart the Codex client "
        "(quit and relaunch) so all MCP namespaces reattach.",
        "After reconnect, rerun the blocked workflow from preflight: "
        "gitea_whoami → gitea_resolve_task_capability → the original task. "
        "Do not resume mid-mutation.",
    ),
    "claude_code": (
        "Run `/mcp` (or open the MCP servers UI) in Claude Code.",
        "Reconnect the affected gitea-* server entry so the client reopens stdio.",
        "If reconnect fails, relaunch the Claude Code session entirely.",
        "After reconnect, restart the workflow from gitea_whoami → "
        "gitea_resolve_task_capability → task.",
    ),
    "generic": (
        "Use the host/IDE MCP reconnect or reload control for the named namespace.",
        "If no per-namespace control exists, restart the MCP client/editor.",
        "After reconnect, restart the workflow from identity/capability preflight.",
    ),
}

#: What an *unidentified* client gets. #948: this is deliberately the
#: host-agnostic step set rather than a specific product. Defaulting to one
#: vendor emitted Codex UI steps to a Gemini/Antigravity operator, who then had
#: no reachable recovery path — the guidance named a panel they do not have.
DEFAULT_CLIENT = "generic"

#: The historical default, kept addressable by name so Codex callers still get
#: Codex steps, without it silently becoming the fallback for unknown clients.
LEGACY_DEFAULT_CLIENT = "codex"


def normalize_reason(reason: str | None) -> str:
    """Map free-form reason strings onto the closed vocabulary."""
    raw = (reason or "").strip().lower()
    if not raw:
        return REASON_UNSPECIFIED
    if raw in VALID_REASONS:
        return raw
    text = raw.replace(" ", "_").replace("-", "_")
    aliases = {
        "stale_runtime": REASON_STALE_RUNTIME,
        "staleruntime": REASON_STALE_RUNTIME,
        "runtime_stale": REASON_STALE_RUNTIME,
        "stale": REASON_STALE_RUNTIME,
        "transport_eof": REASON_TRANSPORT_EOF,
        "transport_closed": REASON_TRANSPORT_EOF,
        "eof": REASON_TRANSPORT_EOF,
        "client_is_closing": REASON_TRANSPORT_EOF,
        "missing_namespace": REASON_MISSING_NAMESPACE,
        "namespace_missing": REASON_MISSING_NAMESPACE,
        "not_required": REASON_NOT_REQUIRED,
        "healthy": REASON_NOT_REQUIRED,
        "ok": REASON_NOT_REQUIRED,
        "unspecified": REASON_UNSPECIFIED,
    }
    if text in aliases:
        return aliases[text]
    hyphenated = text.replace("_", "-")
    if hyphenated in VALID_REASONS:
        return hyphenated
    return REASON_UNSPECIFIED


def normalize_client(client: str | None) -> str:
    """Return the UI-step key for a client.

    #948: alias resolution is shared with ``mcp_worker_identity`` so a client
    name means the same thing wherever it is read. A name we recognise but have
    no bespoke steps for — Gemini, Antigravity, Grok — resolves to the generic
    host-agnostic steps rather than to another vendor's panel.
    """
    import mcp_worker_identity

    canonical = mcp_worker_identity.normalize_client_name(client)
    if canonical in OPERATOR_UI_STEPS:
        return canonical
    return DEFAULT_CLIENT


def classify_boundary_status(
    *,
    startup_sha: str | None,
    current_master_sha: str | None,
    live_stale: bool | None = None,
    in_parity: bool | None = None,
) -> str:
    """Derive boundary_status from parity evidence."""
    if live_stale is True or in_parity is False:
        return BOUNDARY_STALE
    start = (startup_sha or "").strip().lower()
    current = (current_master_sha or "").strip().lower()
    if start and current and start != current:
        return BOUNDARY_MISMATCH
    if start and current and start == current:
        return BOUNDARY_CLEAN
    if in_parity is True:
        return BOUNDARY_CLEAN
    return BOUNDARY_UNKNOWN


def operator_ui_steps(client: str | None, *, namespace: str | None = None) -> list[str]:
    """Exact operator UI steps for the named client."""
    key = normalize_client(client)
    steps = list(OPERATOR_UI_STEPS.get(key) or OPERATOR_UI_STEPS[DEFAULT_CLIENT])
    ns = (namespace or "").strip()
    if ns:
        steps = [
            s.replace("named Gitea MCP server entry (namespace)", f"namespace '{ns}'")
            .replace("affected gitea-* server entry", f"server entry '{ns}'")
            .replace("named namespace", f"namespace '{ns}'")
            for s in steps
        ]
    return steps


def build_reconnect_request(
    *,
    namespace: str,
    profile: str | None = None,
    pid: int | str | None = None,
    session_id: str | None = None,
    startup_sha: str | None = None,
    current_master_sha: str | None = None,
    boundary_status: str | None = None,
    reason: str | None = None,
    client: str | None = DEFAULT_CLIENT,
    live_stale: bool | None = None,
    in_parity: bool | None = None,
    restart_required: bool | None = None,
    stop_required: bool | None = None,
    extra: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Build the structured reconnect-request / typed-blocker payload (#678).

    Never mutates process, config, or session state. Always side-effect free.
    """
    ns = (namespace or "").strip() or "unknown"
    normalized_reason = normalize_reason(reason)
    boundary = (boundary_status or "").strip() or classify_boundary_status(
        startup_sha=startup_sha,
        current_master_sha=current_master_sha,
        live_stale=live_stale,
        in_parity=in_parity,
    )

    reconnect_needed = True
    if normalized_reason == REASON_NOT_REQUIRED and boundary == BOUNDARY_CLEAN:
        reconnect_needed = False
    if restart_required is False and stop_required is False and boundary == BOUNDARY_CLEAN:
        # Explicit healthy probe
        if normalized_reason in (REASON_NOT_REQUIRED, REASON_UNSPECIFIED):
            reconnect_needed = False
            normalized_reason = REASON_NOT_REQUIRED

    if restart_required is True or stop_required is True:
        reconnect_needed = True
        if normalized_reason in (REASON_NOT_REQUIRED, REASON_UNSPECIFIED):
            normalized_reason = REASON_STALE_RUNTIME

    client_key = normalize_client(client)
    steps = operator_ui_steps(client_key, namespace=ns)

    result: dict[str, Any] = {
        "success": True,
        "read_only": True,
        "reconnect_performed": False,
        "mutation_performed": False,
        "reconnect_needed": reconnect_needed,
        "namespace": ns,
        "profile": (profile or "").strip() or None,
        "pid": pid,
        "session_id": (session_id or "").strip() or None,
        "startup_sha": (startup_sha or "").strip() or None,
        "current_master_sha": (current_master_sha or "").strip() or None,
        "boundary_status": boundary,
        "reason": normalized_reason,
        "client": client_key,
        "forbidden_recovery_paths": list(FORBIDDEN_RECOVERY_PATHS),
        "post_reconnect_preflight": [
            "gitea_whoami",
            "gitea_resolve_task_capability",
            "original_task",
        ],
        "exact_safe_next_action": None,
        "blocker_kind": BLOCKER_NONE,
        "operator_ui_steps": steps,
        "typed_blocker": None,
    }

    if reconnect_needed:
        result["blocker_kind"] = BLOCKER_OPERATOR_RECONNECT
        result["stop_required"] = True
        result["restart_required"] = True
        result["exact_safe_next_action"] = (
            f"blocker_kind={BLOCKER_OPERATOR_RECONNECT}: operator must reconnect "
            f"MCP namespace '{ns}' via the host UI (client={client_key}). "
            "Do not pkill, touch configs, edit session state, or use raw API. "
            "After reconnect, restart from gitea_whoami → "
            "gitea_resolve_task_capability → task."
        )
        result["typed_blocker"] = {
            "blocker_kind": BLOCKER_OPERATOR_RECONNECT,
            "namespaces": [ns],
            "why_reconnect_required": normalized_reason,
            "operator_ui_steps": steps,
            "client": client_key,
            "forbidden_recovery_paths": list(FORBIDDEN_RECOVERY_PATHS),
            "instruction_after_reconnect": (
                "Rerun the blocked workflow from preflight "
                "(gitea_whoami → gitea_resolve_task_capability → task). "
                "Do not continue mid-mutation from pre-reconnect state."
            ),
        }
    else:
        result["stop_required"] = False
        result["restart_required"] = False
        result["exact_safe_next_action"] = (
            f"Reconnect not required for namespace '{ns}' "
            f"(boundary_status={boundary}). Proceed with the original task."
        )

    if extra:
        for key, value in extra.items():
            if key not in result:
                result[key] = value

    return result


def reasons_never_suggest_forbidden(text: str) -> bool:
    """Return True when *text* does not recommend a forbidden recovery path.

    Mentions that *ban* a path (e.g. ``Do not pkill`` / ``never edit session
    state``) are allowed. Positive recommendations such as ``use pkill`` or
    ``run killall`` fail.
    """
    import re

    lowered = (text or "").lower()
    # Strip common ban prefixes so "do not pkill" does not trip positive checks.
    scrubbed = re.sub(
        r"\b(?:do not|don't|never|must not|forbid(?:den)?|ban(?:ned)?)\b"
        r"[^.!;\n]{0,80}",
        " ",
        lowered,
    )
    # Positive imperative / advisory forms that would tell an agent to do harm.
    positive_suggestions = (
        "use pkill",
        "run pkill",
        "try pkill",
        "pkill -f",
        "use killall",
        "run killall",
        "killall mcp",
        "use kill ",
        "run kill ",
        "touch the mcp",
        "touch mcp config",
        "utime(",
        "edit the mcp config to recover",
        "edit .env to recover",
        "import gitea_mcp_server",
        "python -c 'import gitea_mcp",
    )
    return not any(frag in scrubbed for frag in positive_suggestions)