fix(mcp): block manual daemon killing as workflow recovery (Closes #630)

A session that ran `pkill -f mcp_server.py`, waited for the IDE to respawn
the daemons, and then closed an issue left no trace distinguishing that
closure from one performed over a sanctioned runtime. Detection existed only
as advisory strings (`native_mcp_preference.classify_command_path`,
`review_workflow_boundary`) and never failed closed on the mutations that
followed.

Adds `runtime_recovery_guard.py`, mirroring the #671 stable-branch
contamination model so the two cannot drift apart: classification of
kill/pkill/killall commands and known-pid kills, a durable
`runtime_recovery_contamination` marker, a fail-closed pre-flight gate over
the shared review/merge/close/completion task set, and reconciler-only
clearing. `comment_issue` and `lock_issue` stay allowed so a contaminated
worker can still post its audit comment and hand off. The marker is
recovery-critical, so contamination cannot expire into cleanliness with the
session-state TTL.

Read-only inspection (`ps aux | grep mcp_server`), sanctioned reconnects,
and process management unrelated to the daemons are never flagged; a bare
`kill` of an unknown pid is reported as ambiguous rather than contaminating,
so ordinary subprocess work is not false-blocked. A pattern broad enough to
sweep unrelated namespaces (`pkill -f python`) is contamination even when it
never names MCP.

Operator-authorized host maintenance stays permitted, but the authorization
is read from GITEA_OPERATOR_DAEMON_MAINTENANCE_AUTHORIZATION in the process
environment only and never from a tool argument, so a session cannot
authorize itself.

Final-report rules reject clean-session claims and require the contaminated
recovery to be surfaced. Two tools are registered (inventory 112 to 114) and
the sanctioned-versus-forbidden contrast is documented in the namespace
recovery doc and the workflow skill.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

runtime_recovery_guard.py

@@ -0,0 +1,499 @@
+"""Fail-closed guard against manual MCP daemon process killing (#630).
+
+Workflow recovery must use sanctioned reconnect/restart paths only: host
+auto-reconnect, an operator-owned restart, or the documented client relaunch. A
+session that instead runs ``pkill -f mcp_server.py`` has manipulated the very
+host processes its own proof depends on.
+
+Incident origin: a session ran ``ps aux | grep mcp_server``, then
+``pkill -f mcp_server.py``, waited for the IDE to respawn the daemons, called
+MCP tools, and closed issue #601. Nothing distinguished that closure from one
+performed over a sanctioned runtime, and unrelated namespaces may have been
+killed as collateral damage.
+
+Partial detection already existed — ``native_mcp_preference.classify_command_path``
+flags ``kill``/``pkill`` near ``mcp_server`` as an MCP-server touch, and
+``review_workflow_boundary`` classifies a pre-review ``pkill`` as MCP repair
+activity — but neither wrote a durable marker nor failed closed on the
+review / merge / close mutations that followed.
+
+This module mirrors ``stable_branch_push_guard`` (#671) deliberately: same
+contamination-marker shape, same gated-task set, same reconciler-only clear.
+Like that guard it is **pure** — callers gather the raw facts (the proposed
+command line, known MCP pids, the durable marker, the process environment) and
+pass them in, so one implementation serves prompts, MCP gates and tests.
+Nothing here kills, spawns or inspects a process, performs I/O, or reads
+durable state.
+
+Design rules honoured (from the #630 acceptance criteria):
+
+* Detect ``pkill -f mcp_server.py``, ``pkill -f gitea_mcp_server``, broad
+  ``pkill -f mcp``, ``killall`` equivalents, and ``kill <pid>`` of a known MCP
+  daemon pid.
+* Detect a pattern broad enough to take unrelated namespaces as collateral
+  damage (``pkill -f python``) even when it never names MCP.
+* Never flag read-only inspection (``ps aux | grep mcp_server``), a sanctioned
+  client reconnect, or process management unrelated to the daemons. A bare
+  ``kill <pid>`` with no MCP linkage is reported as *ambiguous*, never as
+  contamination, so ordinary subprocess work is not false-blocked.
+* Never accept operator authorization from a tool argument. Authorization is
+  read from the process environment only — which an in-session worker cannot
+  set for an already-running daemon. A self-assertable ``operator_authorized``
+  argument was rejected in the PR #710 review (finding F1) and is not
+  reintroduced here.
+* Contamination is never clearable by the same worker session; only a
+  reconciler (audit) role may clear it or bypass the gate.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from typing import Any, Iterable
+
+# Single source of truth for both the redactor and the gated-mutation set: the
+# #671 guard already owns them, so the two contamination models can never drift
+# apart on which mutations a contaminated session may still perform.
+from stable_branch_push_guard import (  # noqa: F401  (CONTAMINATION_GATED_TASKS re-exported)
+    CONTAMINATION_GATED_TASKS,
+    redact_command,
+)
+
+CONTAMINATION_KIND = "manual_daemon_kill"
+
+#: The session killed (or pattern-matched) an MCP daemon process directly.
+REASON_MANUAL_DAEMON_KILL = "manual_daemon_kill"
+#: The pattern was broad enough to sweep unrelated MCP namespaces.
+REASON_BROAD_PROCESS_KILL = "broad_process_kill"
+
+#: Operator authorization is read from this environment variable ONLY. It is
+#: set outside the workflow session by the operator who owns host maintenance;
+#: an in-session worker cannot set it for an already-running daemon. The value
+#: is an audit reference (ticket, change id, or operator note) and is recorded
+#: on the marker. Never accept this from a tool argument (#710 finding F1).
+OPERATOR_AUTHORIZATION_ENV = "GITEA_OPERATOR_DAEMON_MAINTENANCE_AUTHORIZATION"
+
+REMEDIATION = (
+    "Manual MCP daemon process killing is not a sanctioned workflow recovery. "
+    "Stop, leave the host processes alone, and recover through the client "
+    "reconnect / relaunch path (see docs/mcp-namespace-eof-recovery.md) or an "
+    "operator-owned restart. This session is workflow-contaminated until a "
+    "reconciler audits it; review, merge, close and completion mutations fail "
+    "closed until then."
+)
+
+# ── command tokenising ────────────────────────────────────────────────────────
+
+# Split a compound command line into simple commands on shell separators so
+# ``ps aux | grep mcp_server`` is analysed segment by segment and its harmless
+# inspection half never reaches the kill classifier.
+_SEGMENT_SPLIT_RE = re.compile(r"(?:\|\||&&|\||;|\n)")
+
+_KILL_VERBS = frozenset({"kill", "pkill", "killall"})
+
+# Tokens that may legitimately precede the kill verb in command position.
+_COMMAND_PREFIXES = frozenset({
+    "sudo", "command", "exec", "time", "nohup", "env", "builtin",
+})
+
+# Matches the daemon process names: ``mcp_server``/``mcp-server`` (optionally
+# ``gitea_``-prefixed, optionally ``.py``) or a standalone ``mcp`` token.
+# ``mcpfoo`` deliberately does not match.
+_MCP_TARGET_RE = re.compile(
+    r"(?:gitea[_-])?mcp[_-]?server|(?<![\w-])mcp(?![\w-])",
+    re.IGNORECASE,
+)
+
+# Patterns broad enough that matching them would kill unrelated MCP namespaces
+# (and unrelated tooling) as collateral damage.
+_BROAD_PATTERN_RE = re.compile(
+    r"^(?:python[\d.]*|node|uv|venv|java|ruby|perl|\.|\.\*|\*|%)$",
+    re.IGNORECASE,
+)
+
+# ``pkill``/``killall`` flags that consume the following token as their value,
+# so it is not mistaken for a process pattern.
+_VALUE_FLAGS = frozenset({
+    "-u", "-U", "-g", "-G", "-P", "-t", "-s", "-F", "-M", "-N", "-r",
+    "--signal", "--uid", "--euid", "--group", "--parent", "--session",
+    "--terminal", "--ns", "--nslist", "--pidfile", "--older",
+})
+
+# Sanctioned recovery language — informational only. Its presence never
+# suppresses a detected kill; a session that describes a reconnect *and* runs
+# ``pkill`` is still contaminated.
+_SANCTIONED_RECOVERY_RE = re.compile(
+    r"/mcp\s+reconnect|client\s+reconnect|reconnect\s+the\s+(?:ide|client)|"
+    r"relaunch\s+the\s+(?:ide|client)|ide\s+restart|operator[- ]owned\s+restart",
+    re.IGNORECASE,
+)
+
+
+def _clean(value: str | None) -> str:
+    return (value or "").strip()
+
+
+def _split_segments(command: str) -> list[str]:
+    return [seg for seg in _SEGMENT_SPLIT_RE.split(command) if seg.strip()]
+
+
+def is_sanctioned_recovery(text: str | None) -> bool:
+    """True when *text* describes a sanctioned reconnect/restart path.
+
+    Informational only: this never downgrades a detected process kill.
+    """
+    return bool(_SANCTIONED_RECOVERY_RE.search(_clean(text)))
+
+
+# ── kill classification ───────────────────────────────────────────────────────
+
+def _analyse_kill_segment(
+    segment: str,
+    *,
+    mcp_pids: frozenset[str],
+) -> dict[str, Any] | None:
+    """Classify one command segment, or return None when it is not a kill."""
+    tokens = segment.split()
+    idx = 0
+    # Skip env assignments and harmless command prefixes (``sudo pkill ...``).
+    while idx < len(tokens) and (tokens[idx] in _COMMAND_PREFIXES or "=" in tokens[idx]):
+        idx += 1
+    if idx >= len(tokens):
+        return None
+
+    verb = os.path.basename(tokens[idx]).lower()
+    if verb not in _KILL_VERBS:
+        return None
+
+    operands: list[str] = []
+    skip_next = False
+    for token in tokens[idx + 1:]:
+        if skip_next:
+            skip_next = False
+            continue
+        if token.startswith("-"):
+            if token in _VALUE_FLAGS:
+                skip_next = True
+            continue
+        operands.append(token)
+
+    names_mcp = bool(_MCP_TARGET_RE.search(segment))
+
+    def _result(
+        *,
+        reason_class: str | None,
+        contamination: bool,
+        ambiguous: bool,
+        reason: str,
+    ) -> dict[str, Any]:
+        return {
+            "verb": verb,
+            "operands": operands,
+            "reason_class": reason_class,
+            "contamination": contamination,
+            "ambiguous": ambiguous,
+            "reason": reason,
+        }
+
+    if verb in {"pkill", "killall"}:
+        if names_mcp:
+            return _result(
+                reason_class=REASON_MANUAL_DAEMON_KILL,
+                contamination=True,
+                ambiguous=False,
+                reason=(
+                    f"'{verb}' targets the MCP daemon process pattern; this is "
+                    "manual daemon killing, not a sanctioned recovery"
+                ),
+            )
+        broad = [op for op in operands if _BROAD_PATTERN_RE.match(op)]
+        if broad:
+            return _result(
+                reason_class=REASON_BROAD_PROCESS_KILL,
+                contamination=True,
+                ambiguous=False,
+                reason=(
+                    f"'{verb}' pattern {broad[0]!r} is broad enough to kill "
+                    "unrelated MCP namespaces as collateral damage"
+                ),
+            )
+        if not operands:
+            return _result(
+                reason_class=None,
+                contamination=False,
+                ambiguous=True,
+                reason=f"'{verb}' with no resolvable pattern; target unknown",
+            )
+        return _result(
+            reason_class=None,
+            contamination=False,
+            ambiguous=False,
+            reason=(
+                f"'{verb}' targets {operands!r}, which does not name an MCP "
+                "daemon or a broad pattern"
+            ),
+        )
+
+    # ``kill`` — pid-addressed.
+    pids = [op for op in operands if op.isdigit()]
+    hits = sorted(set(pids) & mcp_pids, key=int)
+    if hits:
+        return _result(
+            reason_class=REASON_MANUAL_DAEMON_KILL,
+            contamination=True,
+            ambiguous=False,
+            reason=(
+                "'kill' targets known MCP daemon pid(s) "
+                f"{', '.join(hits)}; this is manual daemon killing"
+            ),
+        )
+    if names_mcp:
+        return _result(
+            reason_class=REASON_MANUAL_DAEMON_KILL,
+            contamination=True,
+            ambiguous=False,
+            reason="'kill' resolves its target from an MCP daemon process lookup",
+        )
+    if not pids:
+        return _result(
+            reason_class=None,
+            contamination=False,
+            ambiguous=True,
+            reason="'kill' with no resolvable numeric pid; target unknown",
+        )
+    return _result(
+        reason_class=None,
+        contamination=False,
+        ambiguous=True,
+        reason=(
+            f"'kill' targets pid(s) {', '.join(pids)}, which are not known MCP "
+            "daemon pids; pass mcp_pids to resolve the ambiguity"
+        ),
+    )
+
+
+def classify_recovery_command(
+    command: str | None = None,
+    *,
+    mcp_pids: Iterable[Any] | None = None,
+) -> dict[str, Any]:
+    """Classify a proposed command for manual MCP daemon kill intent (#630).
+
+    Pure classification; operator authorization is applied separately by
+    :func:`assess_recovery_command`.
+    """
+    text = _clean(command)
+    pid_set = frozenset(
+        str(pid).strip() for pid in (mcp_pids or []) if str(pid).strip()
+    )
+
+    segments: list[dict[str, Any]] = []
+    for raw_segment in _split_segments(text):
+        analysed = _analyse_kill_segment(raw_segment, mcp_pids=pid_set)
+        if analysed is not None:
+            analysed["segment"] = redact_command(raw_segment)
+            segments.append(analysed)
+
+    contaminating = [seg for seg in segments if seg["contamination"]]
+    return {
+        "command_present": bool(text),
+        "redacted_command": redact_command(text),
+        "process_kill": bool(segments),
+        "contamination": bool(contaminating),
+        "reason_class": contaminating[0]["reason_class"] if contaminating else None,
+        "ambiguous": bool(
+            not contaminating and any(seg["ambiguous"] for seg in segments)
+        ),
+        "sanctioned_recovery": is_sanctioned_recovery(text),
+        "segments": segments,
+        "reasons": [seg["reason"] for seg in segments],
+        "known_mcp_pids": sorted(pid_set, key=lambda p: int(p) if p.isdigit() else 0),
+    }
+
+
+# ── operator authorization ────────────────────────────────────────────────────
+
+def operator_authorization(env: dict[str, str] | None = None) -> dict[str, Any]:
+    """Read operator authorization for host daemon maintenance (#630 non-goal 1).
+
+    Authorization comes from :data:`OPERATOR_AUTHORIZATION_ENV` in the process
+    environment and from nowhere else. A worker session cannot set an
+    environment variable for an already-running daemon, so this cannot be
+    self-asserted the way a tool argument could be (#710 finding F1).
+    """
+    source = env if env is not None else os.environ
+    reference = _clean(source.get(OPERATOR_AUTHORIZATION_ENV))
+    return {
+        "authorized": bool(reference),
+        "reference": reference or None,
+        "source": OPERATOR_AUTHORIZATION_ENV if reference else None,
+        "self_assertable": False,
+    }
+
+
+def assess_recovery_command(
+    command: str | None = None,
+    *,
+    mcp_pids: Iterable[Any] | None = None,
+    env: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """Classify *command* and apply operator authorization (#630 AC1/AC2)."""
+    classification = classify_recovery_command(command, mcp_pids=mcp_pids)
+    authorization = operator_authorization(env)
+    detected = classification["contamination"]
+    contaminated = detected and not authorization["authorized"]
+    return {
+        "classification": classification,
+        "authorization": authorization,
+        "contaminated": contaminated,
+        "authorized_bypass": bool(detected and authorization["authorized"]),
+        "remediation": REMEDIATION if contaminated else None,
+    }
+
+
+# ── contamination record + gate ───────────────────────────────────────────────
+
+def build_contamination_record(
+    *,
+    reason_class: str,
+    command_redacted: str | None = None,
+    session_id: str | None = None,
+    remote: str | None = None,
+    role: str | None = None,
+    detail: str | None = None,
+    authorization_reference: str | None = None,
+) -> dict[str, Any]:
+    """Build the durable contamination marker payload (redacted, audit-safe).
+
+    ``reason_class`` is :data:`REASON_MANUAL_DAEMON_KILL` or
+    :data:`REASON_BROAD_PROCESS_KILL`. The command is stored already redacted;
+    secrets never persist on the marker.
+    """
+    return {
+        "kind": CONTAMINATION_KIND,
+        "reason_class": _clean(reason_class) or REASON_MANUAL_DAEMON_KILL,
+        "command_summary": redact_command(command_redacted),
+        "session_id": _clean(session_id) or None,
+        "remote": _clean(remote) or None,
+        "role": _clean(role) or None,
+        "detail": _clean(detail) or None,
+        "authorization_reference": _clean(authorization_reference) or None,
+        "cleared_by_reconciler": False,
+    }
+
+
+def assess_contamination_gate(
+    marker: dict[str, Any] | None,
+    *,
+    task: str | None,
+    actual_role: str | None,
+) -> dict[str, Any]:
+    """Fail closed on gated mutations while a contamination marker is live (#630 AC3).
+
+    * No marker → allowed.
+    * Reconciler (audit) role → allowed (the sanctioned inspect/clear path).
+    * Marker present + ``task`` in :data:`CONTAMINATION_GATED_TASKS` → blocked.
+    * Marker present + non-gated task (``comment_issue``, ``lock_issue``) →
+      allowed, so the contaminated worker can still post the durable audit
+      comment and hand off.
+    """
+    if not marker or marker.get("cleared_by_reconciler"):
+        return {"block": False, "reasons": [], "task": task}
+
+    role = _clean(actual_role).lower()
+    if role == "reconciler":
+        return {
+            "block": False,
+            "reasons": [],
+            "task": task,
+            "detail": "reconciler audit path is exempt from the contamination gate",
+        }
+
+    task_name = _clean(task)
+    if task_name and task_name in CONTAMINATION_GATED_TASKS:
+        summary = marker.get("command_summary") or marker.get("detail") or "(no summary)"
+        reason_class = marker.get("reason_class") or REASON_MANUAL_DAEMON_KILL
+        return {
+            "block": True,
+            "reasons": [
+                f"session is workflow-contaminated ({reason_class}): {summary}. "
+                f"'{task_name}' is blocked until a reconciler audits and clears "
+                "the contamination. " + REMEDIATION
+            ],
+            "task": task_name,
+        }
+
+    return {"block": False, "reasons": [], "task": task_name or None}
+
+
+def format_contamination_gate_error(gate: dict[str, Any]) -> str:
+    """Single RuntimeError message for MCP mutation gates."""
+    reasons = "; ".join(gate.get("reasons") or ["session workflow-contaminated"])
+    return f"Runtime-recovery contamination gate (#630): {reasons}"
+
+
+# ── final-report rules ────────────────────────────────────────────────────────
+
+# Claims that assert a clean session. While a marker is live these are false and
+# must be rejected rather than merely downgraded.
+_CLEAN_CLAIM_RE = re.compile(
+    r"\bclean\s+session\b|\bsession\s+(?:is|was|remains)\s+clean\b|"
+    r"\bno\s+contamination\b|\buncontaminated\b|\bcontamination\s*[:=]\s*none\b|"
+    r"\bworkflow[- ]clean\b|\bno\s+workflow\s+contamination\b",
+    re.IGNORECASE,
+)
+
+# Language that actually surfaces the contamination to a reader.
+_SURFACED_RE = re.compile(
+    r"manual[_ ]daemon[_ ]kill|broad[_ ]process[_ ]kill|daemon\s+process\s+kill|"
+    r"contaminated\s+recovery|runtime[- ]recovery\s+contamination|"
+    r"workflow[- ]contaminated",
+    re.IGNORECASE,
+)
+
+
+def assess_final_report_claim(
+    report_text: str | None,
+    marker: dict[str, Any] | None,
+) -> dict[str, Any]:
+    """Reject clean-session claims while contaminated (#630 scope item 4).
+
+    A live marker imposes two obligations on the final report: it must surface
+    the contaminated recovery explicitly, and it must not claim the session is
+    clean. Either failure blocks.
+    """
+    if not marker or marker.get("cleared_by_reconciler"):
+        return {
+            "block": False,
+            "reasons": [],
+            "contaminated": False,
+            "surfaced": None,
+            "clean_claim": False,
+        }
+
+    text = _clean(report_text)
+    surfaced = bool(_SURFACED_RE.search(text))
+    clean_claim = bool(_CLEAN_CLAIM_RE.search(text))
+
+    reasons: list[str] = []
+    if clean_claim:
+        reasons.append(
+            "final report claims a clean session while a live "
+            f"{marker.get('reason_class') or CONTAMINATION_KIND} contamination "
+            "marker exists; the claim is false and must be removed"
+        )
+    if not surfaced:
+        reasons.append(
+            "final report does not surface the contaminated runtime recovery; "
+            "the report must state that MCP daemon processes were manually "
+            "killed and that the session awaits a reconciler audit"
+        )
+
+    return {
+        "block": bool(reasons),
+        "reasons": reasons,
+        "contaminated": True,
+        "surfaced": surfaced,
+        "clean_claim": clean_claim,
+        "reason_class": marker.get("reason_class"),
+    }