"""Static and safety tests for the guarded live OOS runner.""" from __future__ import annotations import sys from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) import run_leo_m3taversal_oos_handler_suite as suite # noqa: E402 import working_leo_m3taversal_oos_protocol as protocol_lib # noqa: E402 def protocol() -> dict: return protocol_lib.freeze_protocol( "runner-test-seed", created_at_utc="2026-07-15T00:00:00+00:00", ) @pytest.mark.parametrize("grounding_mode", suite.GROUNDING_MODES) def test_guarded_remote_script_compiles_and_installs_preexecution_gate(grounding_mode: str) -> None: script = suite.build_guarded_remote_script( "cafef00d", prompts=[{"id": "P1", "dimension": "demo", "message": "Read the database only"}], suite_mode="test-mode", report_prefix="oos-test-report", prompt_note="frozen test", grounding_mode=grounding_mode, leakage_scan={"pass": True, "exact_prompt_matches": []}, ) compile(script, "", "exec") assert "OOS_PROMPT_LEAKAGE_SCAN = {'pass': True" in script assert 'OOS_PROMPT_LEAKAGE_SCAN = {"pass": true' not in script assert "install_read_only_tool_surface" in script assert "trace_payload_sha256" in script assert 'tool_trace_module.__file__ = ""' in script assert "LEO_TOOL_TRACE_SOURCE_SHA256" in script assert "from leo_tool_trace import extract_kb_tool_trace" not in script assert '"tool_trace_source_sha256": LEO_TOOL_TRACE_SOURCE_SHA256' in script assert 'report["preexecution_safety_gate"]' in script assert 'report["remote_temp_profile_prompt_leakage_scan"]' in script assert '"errors": remote_scan_errors' in script assert "scan_path.resolve(strict=True)" in script assert "2_000_000" not in script assert 'raise RuntimeError("OOS pre-execution safety gate failed")' in script assert "send_message_tool_enabled" in script assert '"scope": "full_model_visible_temp_profile_excluding_sessions_state_memories_and_venv"' in script assert "Telegram transport is denied in the OOS no-post harness" in script assert "runner.adapters.clear()" in script assert 'report["executed_behavior_manifest"] = build_behavior_manifest(temp_profile)' in script assert script.index('report["executed_behavior_manifest"] = build_behavior_manifest(temp_profile)') < script.index( "source = SessionSource(" ) assert "timeout=180" in script assert "timeout=300" not in script if grounding_mode == "db_tool_ablated": assert "shutil.rmtree(db_context_dir)" in script def test_prompt_leakage_scan_uses_partial_markers_and_runtime_roots( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: frozen = protocol() marker = frozen["trials"][0]["prompts"][0]["leakage_markers"][1] runtime_root = tmp_path / "skills" runtime_root.mkdir() (runtime_root / "leaked.md").write_text(f"prefix {marker} suffix", encoding="utf-8") monkeypatch.setattr(suite, "RUNTIME_PROMPT_SCAN_ROOTS", (runtime_root,)) scan = suite.prompt_leakage_scan(frozen) assert scan["pass"] is False assert scan["exact_prompt_matches"][0]["prompt_id"] == frozen["trials"][0]["prompts"][0]["id"] assert scan["exact_prompt_matches"][0]["marker_sha256"] def test_current_repo_runtime_has_no_frozen_prompt_marker_leakage() -> None: scan = suite.prompt_leakage_scan(protocol()) assert scan["pass"] is True, scan["exact_prompt_matches"] assert scan["scanned_files"] > 0 @pytest.mark.parametrize("root_state", ("missing", "empty", "symlink")) def test_prompt_leakage_scan_fails_closed_without_real_coverage( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, root_state: str ) -> None: runtime_root = tmp_path / "runtime" if root_state == "empty": runtime_root.mkdir() elif root_state == "symlink": target = tmp_path / "target" target.mkdir() (target / "runtime.py").write_text("safe runtime text", encoding="utf-8") runtime_root.symlink_to(target, target_is_directory=True) monkeypatch.setattr(suite, "RUNTIME_PROMPT_SCAN_ROOTS", (runtime_root,)) scan = suite.prompt_leakage_scan(protocol()) assert scan["pass"] is False assert scan["expected_roots"][0]["exists"] is (root_state != "missing") assert scan["scanned_files"] == 0 def test_prompt_leakage_scan_fails_closed_on_nested_symlink(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: runtime_root = tmp_path / "runtime" runtime_root.mkdir() (runtime_root / "runtime.py").write_text("safe runtime text", encoding="utf-8") hidden = tmp_path / "hidden" hidden.mkdir() (hidden / "prompt.md").write_text("unscanned prompt text", encoding="utf-8") (runtime_root / "linked").symlink_to(hidden, target_is_directory=True) monkeypatch.setattr(suite, "RUNTIME_PROMPT_SCAN_ROOTS", (runtime_root,)) scan = suite.prompt_leakage_scan(protocol()) assert scan["pass"] is False assert scan["scanned_files"] == 1 assert len(scan["symlink_entries"]) == 1 def test_restart_probe_requires_full_fingerprint_guard_and_cleanup() -> None: counts = {"public.claims": 2} fingerprint = {"status": "ok", "fingerprint_sha256": "a" * 64} report = { "remote_returncode": 0, "pass_runtime": True, "posted_to_telegram": False, "db_counts_changed": False, "db_counts_before": counts, "db_counts_after": counts, "db_fingerprint_unchanged": True, "db_fingerprint_before": fingerprint, "db_fingerprint_after": fingerprint, "preexecution_safety_gate": {"status": "pass"}, "telegram_transport_deny": { "enabled": True, "attempt_count": 0, "runner_adapters_empty": True, }, "temp_profile_removed": True, "post_run_orphan_readback": {"no_matching_processes": True}, } assert suite._restart_probe_passes(report) is True for path in ( ("db_fingerprint_unchanged",), ("preexecution_safety_gate", "status"), ("post_run_orphan_readback", "no_matching_processes"), ("telegram_transport_deny", "enabled"), ): broken = {**report} if len(path) == 1: broken[path[0]] = False else: broken[path[0]] = {**report[path[0]], path[1]: False} assert suite._restart_probe_passes(broken) is False def test_portable_artifact_paths_are_repo_relative() -> None: assert suite._portable_artifact_path(ROOT / "pyproject.toml") == "pyproject.toml"