Gitea-Tools/tests/test_issue_630_runtime_recovery_guard.py

"""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


# ── #787: background separator and subshell forms reach the classifier ───────

def test_background_separator_kill_is_contamination():
    result = guard.classify_recovery_command("sleep 1 & 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_subshell_wrapped_kill_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_further_background_and_subshell_forms_are_contamination():
    for command in (
        "pkill -f mcp_server.py &",
        "( sudo pkill -f mcp_server.py )",
        "((pkill -f gitea_mcp_server))",
        "sleep 1 & killall mcp_server",
        "(ps aux | grep mcp_server) & pkill -f mcp_server.py",
    ):
        result = guard.classify_recovery_command(command)
        assert result["contamination"] is True, command
        assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL, command


def test_logical_operators_are_not_split_into_single_characters():
    # ``&&``/``||`` must still be consumed whole by the widened alternation.
    assert guard._split_segments("a && b || c") == ["a", "b", "c"]
    assert guard._split_segments("a & b") == ["a", "b"]
    assert guard._split_segments("(a)") == ["a"]


# ── 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_user_scoped_pkill_of_unrelated_app_is_not_contamination():
    # ``-u`` consumes ``mcpuser``; the surviving operand names no daemon (#787).
    result = guard.classify_recovery_command("pkill -u mcpuser -f myapp")
    assert result["process_kill"] is True
    assert result["contamination"] is False
    assert result["ambiguous"] is False


def test_commit_message_quoting_the_kill_string_is_not_a_kill():
    result = guard.classify_recovery_command(
        'git commit -m "block pkill -f mcp_server.py as workflow recovery"'
    )
    assert result["process_kill"] is False
    assert result["contamination"] 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_background_separator_kill():
    _clear_marker()
    res = srv.gitea_record_daemon_process_kill_attempt(
        command="sleep 1 & 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_subshell_wrapped_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")