diff --git a/docs/mcp-namespace-eof-recovery.md b/docs/mcp-namespace-eof-recovery.md index 3bd3e96..7fd8ef2 100644 --- a/docs/mcp-namespace-eof-recovery.md +++ b/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 ` 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. diff --git a/docs/mcp-tool-inventory.md b/docs/mcp-tool-inventory.md index 61dadf0..a44a6c7 100644 --- a/docs/mcp-tool-inventory.md +++ b/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` diff --git a/final_report_validator.py b/final_report_validator.py index 904ccf0..f00515a 100644 --- a/final_report_validator.py +++ b/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, diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index ffea9ef..3756a4a 100644 --- a/gitea_mcp_server.py +++ b/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 `` 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 `` 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, diff --git a/mcp_session_state.py b/mcp_session_state.py index 1f1a28d..e4c015f 100644 --- a/mcp_session_state.py +++ b/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, } ) diff --git a/runtime_recovery_guard.py b/runtime_recovery_guard.py new file mode 100644 index 0000000..f941403 --- /dev/null +++ b/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 `` 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 `` 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|(? 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"), + } diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index 44cb731..58762c0 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/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 ` 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 diff --git a/tests/test_issue_630_runtime_recovery_guard.py b/tests/test_issue_630_runtime_recovery_guard.py new file mode 100644 index 0000000..4978328 --- /dev/null +++ b/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")