Merge pull request 'fix(mcp): block manual daemon killing as workflow recovery (Closes #630)' (#786) from fix/issue-630-daemon-kill-contamination into master

docs/mcp-namespace-eof-recovery.md

@@ -110,8 +110,49 @@ healthy. See `docs/mcp-namespace-health.md`.
 - Do **not** kill MCP PIDs or touch config mtimes as a substitute for client
   reconnect.
 
+## Sanctioned recovery vs forbidden process manipulation (#630)
+
+Both restore a working namespace. Only one leaves the session trustworthy.
+
+**Sanctioned — the runtime is repaired by whoever owns it:**
+
+- IDE/host auto-reconnect, or an explicit client reconnect (`/mcp reconnect`).
+- Relaunching the IDE/client so it respawns the daemons it started.
+- An operator-owned restart performed outside the workflow session.
+
+**Forbidden — the session manipulates the processes its own proof depends on:**
+
+- `pkill -f mcp_server.py`, `pkill -f gitea_mcp_server`, broad `pkill -f mcp`.
+- `killall` of a daemon, or `kill <pid>` of an MCP daemon pid.
+- Any pattern broad enough to take unrelated namespaces with it
+  (`pkill -f python`), even when it never names MCP.
+
+Read-only inspection (`ps aux | grep mcp_server`) is neither: it proves nothing
+and breaks nothing. A `kill` of some unrelated pid is reported as *ambiguous*
+rather than contaminating, so ordinary subprocess work is never false-blocked.
+
+**What happens on a detected attempt.** `gitea_record_daemon_process_kill_attempt`
+classifies a proposed command and, when it is a manual daemon kill, writes a
+durable contamination marker for the active profile identity. While that marker
+is live every review / merge / close / completion mutation fails closed;
+`comment_issue` and `lock_issue` stay allowed so the contaminated worker can
+still post its audit comment and hand off. The final report must surface the
+contaminated recovery and must not claim a clean session.
+
+Contamination is **not self-clearable**. Only
+`gitea_audit_runtime_recovery_contamination` with `action=clear`, run under a
+reconciler profile, removes it. The marker is recovery-critical, so it does not
+expire into cleanliness when the session-state TTL lapses.
+
+**Operator-authorized host maintenance stays permitted.** Authorization is read
+from the `GITEA_OPERATOR_DAEMON_MAINTENANCE_AUTHORIZATION` environment variable
+and from nowhere else — set outside the session by the operator who owns the
+host, and recorded as an audit reference on the assessment. It is deliberately
+not a tool argument: a session must never be able to authorize itself.
+
 ## Related
 
+- #630 — manual daemon killing as contaminated recovery (this contrast, enforced).
 - #531 / #544 — stale-runtime detection (`ps`-based); sibling failure mode.
 - #558 / `docs/mcp-daemon-import-guard.md` — why shell imports are not a repair.
 - `docs/mcp-client-registration.md` — per-server registration contract.

docs/mcp-tool-inventory.md

@@ -64,6 +64,7 @@ that gates each call, not which tools exist.
 - `gitea_assess_work_issue_duplicate`
 - `gitea_assess_worktree_cleanup_integrity`
 - `gitea_audit_config`
+- `gitea_audit_runtime_recovery_contamination`
 - `gitea_audit_stable_branch_contamination`
 - `gitea_audit_worktree_cleanup`
 - `gitea_authorize_reconciliation_cleanup_phase`
@@ -125,6 +126,7 @@ that gates each call, not which tools exist.
 - `gitea_reconcile_issue_claims`
 - `gitea_reconcile_merged_cleanups`
 - `gitea_reconcile_superseded_by_merged_pr`
+- `gitea_record_daemon_process_kill_attempt`
 - `gitea_record_irrecoverable_decision_lock_provenance`
 - `gitea_record_pre_review_command`
 - `gitea_record_shell_spawn_outcome`

final_report_validator.py

@@ -16,6 +16,7 @@ import issue_acceptance_gate
 import issue_lock_provenance
 import merger_lease_adoption
 import reviewer_handoff_consistency
+import runtime_recovery_guard
 import thread_state_ledger_validator
 from mcp_native_cleanup_proof import assess_mcp_native_cleanup_proof
 from post_merge_cleanup_proof import assess_post_merge_cleanup_proof
@@ -1870,6 +1871,7 @@ def assess_final_report_validator(
     validation_session: dict | None = None,
     reconciler_close_lock: dict | None = None,
     mutation_attempt_ledger: list[dict] | None = None,
+    runtime_recovery_marker: dict | None = None,
 ) -> dict[str, Any]:
     """Validate final-report text against task-specific proof rules (#327).
 
@@ -1908,6 +1910,28 @@ def assess_final_report_validator(
     action_log = sanitized_action_log
     findings.extend(action_log_findings)
 
+    # #630 scope item 4: while a manual daemon-kill contamination marker is
+    # live, the report must surface it and must not claim a clean session.
+    if runtime_recovery_marker:
+        runtime_recovery = runtime_recovery_guard.assess_final_report_claim(
+            report_text,
+            runtime_recovery_marker,
+        )
+        checks["runtime_recovery_contamination"] = runtime_recovery
+        if runtime_recovery.get("block"):
+            findings.extend(
+                _findings_from_reasons(
+                    "shared.runtime_recovery_contamination",
+                    runtime_recovery.get("reasons") or [],
+                    field="Runtime recovery",
+                    severity="block",
+                    safe_next_action=(
+                        "state the manual daemon kill and the pending reconciler "
+                        "audit in the report; remove any clean-session claim"
+                    ),
+                )
+            )
+
     if normalized_kind == "issue_filing" and issue_filing_lock is not None:
         checks["issue_filing"] = assess_issue_filing_final_report(
             report_text,

gitea_mcp_server.py

@@ -1410,6 +1410,10 @@ def verify_preflight_purity(
         # contaminated by a direct stable-branch push attempt (reconciler-exempt).
         _enforce_stable_branch_contamination_gate(task, remote)
 
+        # #630: block the same mutation classes while the session is
+        # contaminated by manual MCP daemon process killing (reconciler-exempt).
+        _enforce_runtime_recovery_contamination_gate(task, remote)
+
         ctx = _resolve_namespace_mutation_context(worktree_path)
         workspace = ctx["workspace_path"]
         canonical_root = ctx["canonical_repo_root"]
@@ -1874,6 +1878,74 @@ def _clear_stable_contamination_marker(
     )
 
 
+def _runtime_recovery_profile_identity() -> str:
+    return mcp_session_state.current_profile_identity(
+        profile_name=get_profile().get("profile_name"),
+    )
+
+
+def _load_runtime_recovery_marker(remote: str | None = None) -> dict | None:
+    return mcp_session_state.load_state(
+        kind=mcp_session_state.KIND_RUNTIME_RECOVERY_CONTAMINATION,
+        remote=remote,
+        profile_identity=_runtime_recovery_profile_identity(),
+    )
+
+
+def _save_runtime_recovery_marker(
+    record: dict,
+    *,
+    remote: str | None = None,
+) -> dict | None:
+    return mcp_session_state.save_state(
+        kind=mcp_session_state.KIND_RUNTIME_RECOVERY_CONTAMINATION,
+        payload=record,
+        remote=remote,
+        profile_identity=_runtime_recovery_profile_identity(),
+    )
+
+
+def _clear_runtime_recovery_marker(
+    *,
+    remote: str | None = None,
+    profile_identity: str | None = None,
+) -> None:
+    mcp_session_state.clear_state(
+        kind=mcp_session_state.KIND_RUNTIME_RECOVERY_CONTAMINATION,
+        remote=remote,
+        profile_identity=profile_identity or _runtime_recovery_profile_identity(),
+    )
+
+
+def _enforce_runtime_recovery_contamination_gate(
+    task: str | None,
+    remote: str | None = None,
+) -> None:
+    """#630 AC3: fail closed on gated mutations after a manual daemon kill.
+
+    Mirrors the #671 stable-branch gate: reconciler role is exempt (the
+    sanctioned audit/clear path), and non-gated tasks (comment_issue,
+    lock_issue) stay allowed so a contaminated worker can still post the
+    durable audit comment and hand off.
+    """
+    if _preflight_in_test_mode() and not os.environ.get(
+        "GITEA_TEST_FORCE_RUNTIME_CONTAMINATION"
+    ):
+        return
+    marker = _load_runtime_recovery_marker(remote)
+    if not marker:
+        return
+    gate = runtime_recovery_guard.assess_contamination_gate(
+        marker,
+        task=task,
+        actual_role=_actual_profile_role(),
+    )
+    if gate["block"]:
+        raise RuntimeError(
+            runtime_recovery_guard.format_contamination_gate_error(gate)
+        )
+
+
 def _enforce_stable_branch_contamination_gate(
     task: str | None,
     remote: str | None = None,
@@ -1954,6 +2026,7 @@ import author_mutation_worktree  # noqa: E402
 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 runtime_recovery_guard  # noqa: E402  # #630 manual daemon-kill contamination
 import remote_repo_guard  # noqa: E402
 import anti_stomp_preflight  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
@@ -15871,6 +15944,146 @@ def gitea_audit_stable_branch_contamination(
     }
 
 
+@mcp.tool()
+def gitea_record_daemon_process_kill_attempt(
+    command: str | None = None,
+    remote: str = "dadeschools",
+    mcp_pids: list[str] | None = None,
+    session_id: str | None = None,
+    mark: bool = True,
+) -> dict:
+    """Classify a proposed command for manual MCP daemon kill intent (#630).
+
+    Workflow recovery must use sanctioned reconnect/restart paths only. This
+    tool detects ``pkill -f mcp_server.py`` and its equivalents (``killall``,
+    ``pkill -f gitea_mcp_server``, broad ``pkill -f python`` sweeps that would
+    take unrelated namespaces as collateral damage, and ``kill <pid>`` of a pid
+    listed in ``mcp_pids``).
+
+    Read-only inspection (``ps aux | grep mcp_server``) and a sanctioned client
+    reconnect are never flagged. A bare ``kill <pid>`` with no MCP linkage is
+    reported as *ambiguous* rather than contaminating, so ordinary subprocess
+    management is not false-blocked.
+
+    Operator authorization for host maintenance is read from the process
+    environment (``GITEA_OPERATOR_DAEMON_MAINTENANCE_AUTHORIZATION``) and never
+    from a tool argument, so it cannot be self-asserted by the session being
+    judged. When authorization is present the attempt is reported as an
+    authorized bypass and no marker is written.
+
+    When ``mark`` is true and contamination is detected without authorization, a
+    durable ``runtime_recovery_contamination`` marker is written for the active
+    profile identity. Subsequent review/merge/close/completion mutations then
+    fail closed until a reconciler audits and clears it. The stored command
+    summary is redacted; secrets never persist.
+    """
+    assessment = runtime_recovery_guard.assess_recovery_command(
+        command,
+        mcp_pids=mcp_pids,
+    )
+    classification = assessment["classification"]
+    authorization = assessment["authorization"]
+    contaminated = bool(assessment["contaminated"])
+
+    marker = None
+    marked = False
+    if contaminated and mark:
+        record = runtime_recovery_guard.build_contamination_record(
+            reason_class=classification.get("reason_class")
+            or runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL,
+            command_redacted=classification.get("redacted_command"),
+            session_id=session_id,
+            remote=remote,
+            role=_actual_profile_role(),
+            detail="; ".join(classification.get("reasons") or []),
+            authorization_reference=authorization.get("reference"),
+        )
+        marker = _save_runtime_recovery_marker(record, remote=remote)
+        marked = marker is not None
+
+    return {
+        "classification": classification,
+        "authorization": authorization,
+        "contaminated": contaminated,
+        "authorized_bypass": bool(assessment["authorized_bypass"]),
+        "marked": marked,
+        "marker": marker,
+        "profile_identity": _runtime_recovery_profile_identity(),
+        "remediation": assessment["remediation"],
+    }
+
+
+@mcp.tool()
+def gitea_audit_runtime_recovery_contamination(
+    action: str = "inspect",
+    remote: str = "dadeschools",
+    profile_identity: str | None = None,
+) -> dict:
+    """Reconciler audit of a manual daemon-kill contamination marker (#630).
+
+    ``action='inspect'`` (default, read-only) loads the durable marker for the
+    given ``profile_identity`` (or the active session's) and reports it.
+
+    ``action='clear'`` removes the marker so the contaminated session may
+    resume gated mutations. Clearing is the reconciler audit path and requires
+    an active reconciler profile — a worker session must never self-clear.
+    ``profile_identity`` targets the contaminated worker's marker (a reconciler
+    runs under its own identity).
+    """
+    act = (action or "inspect").strip().lower()
+    target_identity = (
+        (profile_identity or "").strip() or _runtime_recovery_profile_identity()
+    )
+
+    marker = mcp_session_state.load_state(
+        kind=mcp_session_state.KIND_RUNTIME_RECOVERY_CONTAMINATION,
+        remote=remote,
+        profile_identity=target_identity,
+    )
+
+    if act == "inspect":
+        return {
+            "action": "inspect",
+            "profile_identity": target_identity,
+            "contaminated": marker is not None,
+            "marker": marker,
+            "read_only": True,
+        }
+
+    if act == "clear":
+        role = _actual_profile_role()
+        if role != "reconciler":
+            return {
+                "action": "clear",
+                "success": False,
+                "performed": False,
+                "profile_identity": target_identity,
+                "reasons": [
+                    "runtime-recovery contamination may only be cleared by a "
+                    f"reconciler audit; active role is '{role}' (fail closed). "
+                    "The contaminated worker session must not self-clear."
+                ],
+            }
+        _clear_runtime_recovery_marker(
+            remote=remote,
+            profile_identity=target_identity,
+        )
+        return {
+            "action": "clear",
+            "success": True,
+            "performed": True,
+            "profile_identity": target_identity,
+            "was_contaminated": marker is not None,
+        }
+
+    return {
+        "action": act,
+        "success": False,
+        "performed": False,
+        "reasons": [f"unknown action '{act}'; use 'inspect' or 'clear'"],
+    }
+
+
 @mcp.tool()
 def gitea_validate_review_final_report(
     report_text: str,

mcp_session_state.py

@@ -36,6 +36,11 @@ KIND_REVIEW_DRAFT = "review_draft"
 # other session proofs; a contaminated session fails closed on gated mutations
 # until a reconciler audits and clears it.
 KIND_STABLE_BRANCH_CONTAMINATION = "stable_branch_contamination"
+# Durable marker set when a worker session manually kills MCP daemon processes
+# instead of using a sanctioned reconnect/restart path (#630). Same shape and
+# same reconciler-only clear as the #671 marker above; kept as its own kind so
+# an audit can tell the two contamination classes apart.
+KIND_RUNTIME_RECOVERY_CONTAMINATION = "runtime_recovery_contamination"
 # Durable shadow of the in-memory reviewer session lease (#702). Written on
 # every sanctioned record/heartbeat and removed on sanctioned clear, so a
 # daemon that dies without teardown leaves provable orphan evidence (owner
@@ -71,6 +76,10 @@ RECOVERY_CRITICAL_KINDS = frozenset(
         # #702 crash-orphan evidence (must outlive TTL; F4)
         KIND_REVIEWER_SESSION_LEASE,
         KIND_STALE_BINDING_RECOVERY,
+        # #630: contamination must not expire into cleanliness. A TTL-bound
+        # marker would let a contaminated session self-clear by waiting, which
+        # defeats the reconciler-only clear the gate depends on.
+        KIND_RUNTIME_RECOVERY_CONTAMINATION,
     }
 )
 

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"),
+    }

skills/llm-project-workflow/SKILL.md

@@ -168,6 +168,42 @@ Tooling: call `gitea_record_stable_branch_push_attempt` to classify/record a
 proposed push before running it; `gitea_audit_stable_branch_contamination` to
 inspect or (reconciler-only) clear the marker.
 
+## Runtime Recovery Protection (#630)
+
+MCP connectivity is recovered through **sanctioned reconnect/restart only**:
+host auto-reconnect, an explicit client reconnect, an IDE/client relaunch, or an
+operator-owned restart. Worker sessions must never kill the daemons their own
+proof depends on.
+
+**Forbidden for author/reviewer/merger sessions:**
+
+- `pkill -f mcp_server.py`, `pkill -f gitea_mcp_server`, broad `pkill -f mcp`.
+- `killall` of a daemon, or `kill <pid>` of an MCP daemon pid.
+- Any pattern broad enough to sweep unrelated namespaces (`pkill -f python`),
+  even when it never names MCP.
+
+**Allowed (never blocked):** read-only inspection (`ps aux | grep mcp_server`),
+and process management unrelated to the daemons — a `kill` of some other pid is
+reported as *ambiguous*, not as contamination.
+
+**What happens on a detected attempt:** the session is marked
+workflow-contaminated (durable marker, redacted command summary + session id +
+remote + role). While contaminated, all review / merge / close / completion
+mutations fail closed. `comment_issue` and `lock_issue` remain allowed so the
+contaminated worker can post the durable audit comment and hand off.
+Contamination **cannot be self-cleared** — only a reconciler audit may clear it,
+and it does not expire with the session-state TTL. The final report must surface
+the contaminated recovery and must not claim a clean session.
+
+Operator-authorized host maintenance stays permitted, but the authorization is
+read from the operator's environment, never from a tool argument: a session must
+not be able to authorize itself.
+
+Tooling: call `gitea_record_daemon_process_kill_attempt` to classify/record a
+proposed command before running it; `gitea_audit_runtime_recovery_contamination`
+to inspect or (reconciler-only) clear the marker. Full contrast in
+`docs/mcp-namespace-eof-recovery.md`.
+
 ## Shell Spawn Hard-Stop Rule
 
 `exit_code: -1` with empty stdout/stderr means the shell failed to spawn — not a

tests/test_issue_630_runtime_recovery_guard.py

@@ -0,0 +1,491 @@
+"""Manual MCP daemon-kill contamination guard (#630).
+
+Covers the four scenarios the acceptance criteria name — manual process kill,
+sanctioned reconnect, stale-runtime restart, and a contaminated post-restart
+mutation — across the pure guard, the durable marker, the MCP tools, the
+pre-flight enforcement gate, and the final-report rules.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import patch
+
+import final_report_validator
+import mcp_session_state
+import runtime_recovery_guard as guard
+import gitea_mcp_server as srv
+
+
+AUTH_ENV = guard.OPERATOR_AUTHORIZATION_ENV
+
+
+def _clear_marker(remote="prgs"):
+    srv._clear_runtime_recovery_marker(remote=remote)
+
+
+def teardown_function():
+    _clear_marker()
+
+
+def _marker(reason_class=guard.REASON_MANUAL_DAEMON_KILL, **overrides):
+    record = guard.build_contamination_record(
+        reason_class=reason_class,
+        command_redacted="pkill -f mcp_server.py",
+        session_id="prgs-author-1234-abcd",
+        remote="prgs",
+        role="author",
+        detail="manual daemon kill",
+    )
+    record.update(overrides)
+    return record
+
+
+# ── AC1/AC2: manual process kill is detected and classified ──────────────────
+
+def test_pkill_mcp_server_py_is_contamination():
+    result = guard.classify_recovery_command("pkill -f mcp_server.py")
+    assert result["process_kill"] is True
+    assert result["contamination"] is True
+    assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+    assert result["ambiguous"] is False
+
+
+def test_equivalent_kill_forms_are_contamination():
+    for command in (
+        "pkill -f gitea_mcp_server",
+        "pkill -f mcp",
+        "pkill -9 -f mcp_server.py",
+        "killall mcp_server",
+        "sudo pkill -f mcp_server.py",
+        "killall -9 mcp-server",
+    ):
+        result = guard.classify_recovery_command(command)
+        assert result["contamination"] is True, command
+        assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL, command
+
+
+def test_broad_pattern_is_collateral_damage_contamination():
+    result = guard.classify_recovery_command("pkill -f python")
+    assert result["contamination"] is True
+    assert result["reason_class"] == guard.REASON_BROAD_PROCESS_KILL
+    assert "collateral" in " ".join(result["reasons"])
+
+
+def test_kill_of_known_mcp_pid_is_contamination():
+    result = guard.classify_recovery_command("kill -9 4242", mcp_pids=[4242, 99])
+    assert result["contamination"] is True
+    assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+    assert "4242" in " ".join(result["reasons"])
+
+
+def test_kill_resolved_from_mcp_lookup_is_contamination():
+    result = guard.classify_recovery_command("kill $(pgrep -f mcp_server.py)")
+    assert result["contamination"] is True
+
+
+def test_compound_command_detects_the_kill_half():
+    result = guard.classify_recovery_command(
+        "ps aux | grep mcp_server && pkill -f mcp_server.py"
+    )
+    assert result["contamination"] is True
+
+
+# ── no false positives ───────────────────────────────────────────────────────
+
+def test_read_only_inspection_is_not_a_kill():
+    result = guard.classify_recovery_command("ps aux | grep mcp_server")
+    assert result["process_kill"] is False
+    assert result["contamination"] is False
+
+
+def test_grepping_for_pkill_is_not_a_kill():
+    result = guard.classify_recovery_command('grep -rn "pkill" native_mcp_preference.py')
+    assert result["process_kill"] is False
+    assert result["contamination"] is False
+
+
+def test_unrelated_pkill_target_is_not_contamination():
+    result = guard.classify_recovery_command("pkill -f my-dev-server")
+    assert result["process_kill"] is True
+    assert result["contamination"] is False
+    assert result["ambiguous"] is False
+
+
+def test_bare_kill_of_unknown_pid_is_ambiguous_not_contamination():
+    result = guard.classify_recovery_command("kill 31337")
+    assert result["contamination"] is False
+    assert result["ambiguous"] is True
+    assert "not known MCP" in " ".join(result["reasons"])
+
+
+def test_kill_without_pid_is_ambiguous():
+    result = guard.classify_recovery_command("kill")
+    assert result["contamination"] is False
+    assert result["ambiguous"] is True
+
+
+def test_empty_command_is_inert():
+    result = guard.classify_recovery_command(None)
+    assert result["command_present"] is False
+    assert result["process_kill"] is False
+    assert result["contamination"] is False
+
+
+# ── sanctioned reconnect / restart ───────────────────────────────────────────
+
+def test_sanctioned_reconnect_is_not_contamination():
+    result = guard.classify_recovery_command(
+        "/mcp reconnect then re-run gitea_whoami"
+    )
+    assert result["sanctioned_recovery"] is True
+    assert result["contamination"] is False
+    assert result["process_kill"] is False
+
+
+def test_stale_runtime_restart_language_is_not_contamination():
+    result = guard.classify_recovery_command(
+        "runtime is stale against master; relaunch the IDE client so the "
+        "namespaces restart"
+    )
+    assert result["sanctioned_recovery"] is True
+    assert result["contamination"] is False
+
+
+def test_sanctioned_language_never_excuses_an_actual_kill():
+    result = guard.classify_recovery_command(
+        "client reconnect did not help; pkill -f mcp_server.py"
+    )
+    assert result["sanctioned_recovery"] is True
+    assert result["contamination"] is True
+
+
+# ── operator authorization (env-only, never self-assertable) ─────────────────
+
+def test_operator_authorization_absent_by_default():
+    auth = guard.operator_authorization(env={})
+    assert auth["authorized"] is False
+    assert auth["reference"] is None
+    assert auth["self_assertable"] is False
+
+
+def test_operator_authorization_read_from_env_only():
+    auth = guard.operator_authorization(env={AUTH_ENV: "CHG-4471 host maintenance"})
+    assert auth["authorized"] is True
+    assert auth["reference"] == "CHG-4471 host maintenance"
+    assert auth["source"] == AUTH_ENV
+
+
+def test_authorized_maintenance_is_not_contamination():
+    assessment = guard.assess_recovery_command(
+        "pkill -f mcp_server.py",
+        env={AUTH_ENV: "CHG-4471"},
+    )
+    assert assessment["classification"]["contamination"] is True
+    assert assessment["contaminated"] is False
+    assert assessment["authorized_bypass"] is True
+    assert assessment["remediation"] is None
+
+
+def test_unauthorized_kill_is_contamination():
+    assessment = guard.assess_recovery_command("pkill -f mcp_server.py", env={})
+    assert assessment["contaminated"] is True
+    assert assessment["authorized_bypass"] is False
+    assert assessment["remediation"]
+
+
+# ── redaction ────────────────────────────────────────────────────────────────
+
+def test_marker_and_classification_redact_secrets():
+    command = "GITEA_TOKEN=supersecretvalue pkill -f mcp_server.py"
+    result = guard.classify_recovery_command(command)
+    assert "supersecretvalue" not in result["redacted_command"]
+    assert "GITEA_TOKEN=***" in result["redacted_command"]
+    record = guard.build_contamination_record(
+        reason_class=guard.REASON_MANUAL_DAEMON_KILL,
+        command_redacted=result["redacted_command"],
+    )
+    assert "supersecretvalue" not in record["command_summary"]
+    assert record["cleared_by_reconciler"] is False
+
+
+# ── AC3: gate over the gated mutation set ────────────────────────────────────
+
+def test_gate_blocks_gated_tasks():
+    marker = _marker()
+    for task in ("merge_pr", "review_pr", "close_issue", "create_pr", "submit_pr_review"):
+        gate = guard.assess_contamination_gate(marker, task=task, actual_role="author")
+        assert gate["block"] is True, task
+
+
+def test_gate_allows_handoff_tasks():
+    marker = _marker()
+    for task in ("comment_issue", "lock_issue"):
+        gate = guard.assess_contamination_gate(marker, task=task, actual_role="author")
+        assert gate["block"] is False, task
+
+
+def test_gate_exempts_reconciler():
+    gate = guard.assess_contamination_gate(
+        _marker(), task="merge_pr", actual_role="reconciler"
+    )
+    assert gate["block"] is False
+
+
+def test_gate_allows_when_no_marker_or_cleared():
+    assert guard.assess_contamination_gate(
+        None, task="merge_pr", actual_role="author"
+    )["block"] is False
+    cleared = _marker(cleared_by_reconciler=True)
+    assert guard.assess_contamination_gate(
+        cleared, task="merge_pr", actual_role="author"
+    )["block"] is False
+
+
+def test_gate_error_message_names_the_issue():
+    gate = guard.assess_contamination_gate(
+        _marker(), task="merge_pr", actual_role="author"
+    )
+    assert "#630" in guard.format_contamination_gate_error(gate)
+
+
+# ── scope item 4: final-report rules ─────────────────────────────────────────
+
+def test_final_report_clean_claim_is_rejected():
+    result = guard.assess_final_report_claim(
+        "Runtime recovery: manual daemon kill occurred. Otherwise a clean session.",
+        _marker(),
+    )
+    assert result["block"] is True
+    assert result["clean_claim"] is True
+
+
+def test_final_report_must_surface_the_contamination():
+    result = guard.assess_final_report_claim(
+        "All acceptance criteria met; tests pass.", _marker()
+    )
+    assert result["block"] is True
+    assert result["surfaced"] is False
+
+
+def test_final_report_that_surfaces_and_claims_nothing_clean_passes():
+    result = guard.assess_final_report_claim(
+        "This session performed a manual daemon kill of the MCP processes and "
+        "is workflow-contaminated pending a reconciler audit.",
+        _marker(),
+    )
+    assert result["block"] is False
+    assert result["surfaced"] is True
+
+
+def test_final_report_unconstrained_without_marker():
+    result = guard.assess_final_report_claim("clean session", None)
+    assert result["block"] is False
+    assert result["contaminated"] is False
+
+
+def test_validator_blocks_clean_claim_while_contaminated():
+    out = final_report_validator.assess_final_report_validator(
+        "Merged the PR. No contamination in this session.",
+        "merge_pr",
+        runtime_recovery_marker=_marker(),
+    )
+    assert out["blocked"] is True
+    assert any(
+        finding["rule_id"] == "shared.runtime_recovery_contamination"
+        for finding in out["findings"]
+    )
+
+
+def test_validator_default_is_unchanged_without_marker():
+    out = final_report_validator.assess_final_report_validator(
+        "Merged the PR. No contamination in this session.",
+        "merge_pr",
+    )
+    assert "runtime_recovery_contamination" not in out["checks"]
+    assert not any(
+        finding["rule_id"] == "shared.runtime_recovery_contamination"
+        for finding in out["findings"]
+    )
+
+
+# ── durable marker must outlive the session TTL ──────────────────────────────
+
+def test_contamination_marker_is_recovery_critical():
+    assert (
+        mcp_session_state.KIND_RUNTIME_RECOVERY_CONTAMINATION
+        in mcp_session_state.RECOVERY_CRITICAL_KINDS
+    )
+
+
+# ── server wiring: record tool ───────────────────────────────────────────────
+
+def test_record_tool_marks_manual_daemon_kill():
+    _clear_marker()
+    res = srv.gitea_record_daemon_process_kill_attempt(
+        command="pkill -f mcp_server.py", remote="prgs"
+    )
+    assert res["contaminated"] is True
+    assert res["marked"] is True
+    assert res["marker"]["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+    loaded = srv._load_runtime_recovery_marker("prgs")
+    assert loaded is not None
+    assert "mcp_server.py" in loaded["command_summary"]
+
+
+def test_record_tool_marks_broad_sweep():
+    _clear_marker()
+    res = srv.gitea_record_daemon_process_kill_attempt(
+        command="pkill -f python", remote="prgs"
+    )
+    assert res["contaminated"] is True
+    assert res["marker"]["reason_class"] == guard.REASON_BROAD_PROCESS_KILL
+
+
+def test_record_tool_marks_known_pid_kill():
+    _clear_marker()
+    res = srv.gitea_record_daemon_process_kill_attempt(
+        command="kill -9 4242", mcp_pids=["4242"], remote="prgs"
+    )
+    assert res["contaminated"] is True
+    assert res["marked"] is True
+
+
+def test_record_tool_does_not_mark_inspection():
+    _clear_marker()
+    res = srv.gitea_record_daemon_process_kill_attempt(
+        command="ps aux | grep mcp_server", remote="prgs"
+    )
+    assert res["contaminated"] is False
+    assert res["marked"] is False
+    assert srv._load_runtime_recovery_marker("prgs") is None
+
+
+def test_record_tool_does_not_mark_sanctioned_reconnect():
+    _clear_marker()
+    res = srv.gitea_record_daemon_process_kill_attempt(
+        command="/mcp reconnect", remote="prgs"
+    )
+    assert res["contaminated"] is False
+    assert res["marked"] is False
+    assert srv._load_runtime_recovery_marker("prgs") is None
+
+
+def test_record_tool_mark_false_is_read_only():
+    _clear_marker()
+    res = srv.gitea_record_daemon_process_kill_attempt(
+        command="pkill -f mcp_server.py", remote="prgs", mark=False
+    )
+    assert res["contaminated"] is True
+    assert res["marked"] is False
+    assert srv._load_runtime_recovery_marker("prgs") is None
+
+
+def test_record_tool_honours_operator_authorization():
+    _clear_marker()
+    with patch.dict(os.environ, {AUTH_ENV: "CHG-4471"}):
+        res = srv.gitea_record_daemon_process_kill_attempt(
+            command="pkill -f mcp_server.py", remote="prgs"
+        )
+    assert res["authorized_bypass"] is True
+    assert res["contaminated"] is False
+    assert res["marked"] is False
+    assert srv._load_runtime_recovery_marker("prgs") is None
+
+
+# ── server wiring: audit tool ────────────────────────────────────────────────
+
+def test_audit_inspect_reports_marker():
+    _clear_marker()
+    srv.gitea_record_daemon_process_kill_attempt(
+        command="pkill -f mcp_server.py", remote="prgs"
+    )
+    out = srv.gitea_audit_runtime_recovery_contamination(action="inspect", remote="prgs")
+    assert out["contaminated"] is True
+    assert out["read_only"] is True
+
+
+def test_audit_clear_refused_for_non_reconciler():
+    _clear_marker()
+    srv.gitea_record_daemon_process_kill_attempt(
+        command="pkill -f mcp_server.py", remote="prgs"
+    )
+    with patch.object(srv, "_actual_profile_role", return_value="author"):
+        out = srv.gitea_audit_runtime_recovery_contamination(
+            action="clear", remote="prgs"
+        )
+    assert out["success"] is False
+    assert out["reasons"]
+    assert srv._load_runtime_recovery_marker("prgs") is not None
+
+
+def test_audit_clear_allowed_for_reconciler():
+    _clear_marker()
+    srv.gitea_record_daemon_process_kill_attempt(
+        command="pkill -f mcp_server.py", remote="prgs"
+    )
+    identity = srv._runtime_recovery_profile_identity()
+    with patch.object(srv, "_actual_profile_role", return_value="reconciler"):
+        out = srv.gitea_audit_runtime_recovery_contamination(
+            action="clear", remote="prgs", profile_identity=identity
+        )
+    assert out["success"] is True
+    assert srv._load_runtime_recovery_marker("prgs") is None
+
+
+def test_audit_unknown_action_fails_closed():
+    out = srv.gitea_audit_runtime_recovery_contamination(action="nuke", remote="prgs")
+    assert out["success"] is False
+    assert out["performed"] is False
+
+
+# ── AC3/AC4: contaminated post-restart mutation fails closed ─────────────────
+
+def _force_gate_env():
+    return patch.dict(os.environ, {"GITEA_TEST_FORCE_RUNTIME_CONTAMINATION": "1"})
+
+
+def test_gate_blocks_mutations_after_manual_kill_and_restart():
+    _clear_marker()
+    # The session kills the daemons, the IDE respawns them, and the session then
+    # attempts the mutations #601 was closed with.
+    srv.gitea_record_daemon_process_kill_attempt(
+        command="pkill -f mcp_server.py", remote="prgs"
+    )
+    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="author"):
+        for task in ("merge_pr", "review_pr", "close_issue", "create_pr"):
+            try:
+                srv._enforce_runtime_recovery_contamination_gate(task, "prgs")
+                raised = False
+            except RuntimeError as exc:
+                raised = True
+                assert "#630" in str(exc)
+            assert raised, task
+
+
+def test_gate_allows_handoff_comment_when_contaminated():
+    _clear_marker()
+    srv.gitea_record_daemon_process_kill_attempt(
+        command="pkill -f mcp_server.py", remote="prgs"
+    )
+    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="author"):
+        srv._enforce_runtime_recovery_contamination_gate("comment_issue", "prgs")
+        srv._enforce_runtime_recovery_contamination_gate("lock_issue", "prgs")
+
+
+def test_gate_exempts_reconciler_audit():
+    _clear_marker()
+    srv.gitea_record_daemon_process_kill_attempt(
+        command="pkill -f mcp_server.py", remote="prgs"
+    )
+    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="reconciler"):
+        srv._enforce_runtime_recovery_contamination_gate("merge_pr", "prgs")
+
+
+def test_gate_noop_after_sanctioned_restart_only():
+    _clear_marker()
+    srv.gitea_record_daemon_process_kill_attempt(
+        command="/mcp reconnect", remote="prgs"
+    )
+    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="author"):
+        srv._enforce_runtime_recovery_contamination_gate("merge_pr", "prgs")