from __future__ import annotations import json import pytest from scripts import run_leo_agent_challenger_loop as runner def test_remote_script_embeds_stateless_challenger_read_only_guard_and_bounded_context() -> None: script = runner.build_remote_script("abc123", challenger_max_tokens=512) assert "stateless_openrouter_agent" in script assert "gatewayrunner_temp_profile_no_model_tools" in script assert "blocked by isolated challenger-loop read-only guard" in script context_source = runner._patched_db_context_source() assert '"--limit",\n "4",' in context_source assert '"--context-limit",\n "6",' in context_source assert "CHALLENGER_MAX_TOKENS = 512" in script assert "posted_to_telegram" in script assert "db_fingerprint_unchanged" in script assert '"temperature": 0.0' in script assert "isolated_challenger_final_revision" in script assert '"use_database_context": step < 3' in script assert "LEO_CHALLENGER_CONTEXT_ONLY" in script assert "Follow the challenger's requested labels exactly" in script assert "challenger_generation_audit" in script assert "evaluate_challenger_objection" in script assert "build_challenger_retry_prompt" in script assert "exhausted semantic regeneration attempts before Leo advance" in script assert "challenger_semantic_gate_passed_before_leo_advance" in script def test_remote_script_rejects_unbounded_tokens_and_unsafe_report_prefix() -> None: with pytest.raises(ValueError, match="between 128 and 2048"): runner.build_remote_script("abc123", challenger_max_tokens=4096) with pytest.raises(ValueError, match="report_prefix"): runner.build_remote_script("abc123", report_prefix="bad/path") def test_remote_report_fetch_retries_timeout_without_deleting_receipt(monkeypatch: pytest.MonkeyPatch) -> None: calls = [] outcomes = iter( ( runner.subprocess.TimeoutExpired("ssh", runner.REMOTE_REPORT_FETCH_TIMEOUT_SECONDS), runner.subprocess.CompletedProcess( args=["ssh"], returncode=0, stdout=json.dumps({"run_id": "abc123"}), stderr="", ), ) ) def fake_run(*args, **kwargs): calls.append((args, kwargs)) outcome = next(outcomes) if isinstance(outcome, Exception): raise outcome return outcome monkeypatch.setattr(runner.subprocess, "run", fake_run) report = runner.fetch_remote_report("abc123") assert report == {"run_id": "abc123"} assert len(calls) == 2 assert "p.unlink()" not in calls[0][1]["input"] assert calls[0][1]["timeout"] == runner.REMOTE_REPORT_FETCH_TIMEOUT_SECONDS def test_remote_report_unlink_retries_and_requires_absence_readback(monkeypatch: pytest.MonkeyPatch) -> None: calls = [] outcomes = iter( ( runner.subprocess.TimeoutExpired("ssh", runner.REMOTE_REPORT_FETCH_TIMEOUT_SECONDS), runner.subprocess.CompletedProcess( args=["ssh"], returncode=0, stdout="absent\n", stderr="", ), ) ) def fake_run(*args, **kwargs): calls.append((args, kwargs)) outcome = next(outcomes) if isinstance(outcome, Exception): raise outcome return outcome monkeypatch.setattr(runner.subprocess, "run", fake_run) unlinked = runner.unlink_remote_report("abc123") assert unlinked is True assert len(calls) == 2 assert "p.unlink()" in calls[0][1]["input"] assert "not p.exists()" in calls[0][1]["input"] def test_run_remote_does_not_claim_cleanup_from_fetch_without_unlink_ack( monkeypatch: pytest.MonkeyPatch, ) -> None: def fake_run(args, **_kwargs): if args[:2] == ["git", "rev-parse"]: return runner.subprocess.CompletedProcess(args=args, returncode=0, stdout="a" * 40 + "\n", stderr="") if args[:2] == ["git", "status"]: return runner.subprocess.CompletedProcess(args=args, returncode=0, stdout="", stderr="") return runner.subprocess.CompletedProcess( args=args, returncode=0, stdout=json.dumps({"run_id": "remote"}) + "\n", stderr="", ) monkeypatch.setattr(runner.subprocess, "run", fake_run) monkeypatch.setattr(runner, "fetch_remote_report", lambda *_args, **_kwargs: {"run_id": "remote"}) monkeypatch.setattr(runner, "unlink_remote_report", lambda *_args, **_kwargs: False) result = runner.run_remote() assert result["execution_transport"]["remote_report_fetched_and_unlinked"] is False def test_retention_projection_removes_runtime_paths_without_losing_hash_proof() -> None: manifest = { "schema": "livingip.leoBehaviorManifest.v1", "generated_at_utc": "2026-07-15T00:00:00+00:00", "behavior_sha256": "a" * 64, "profile_root": "/home/teleo/.hermes/profiles/leoclean", "hermes_root": "/home/teleo/.hermes/hermes-agent", "deployment_root": "/opt/teleo-eval/workspaces/deploy-infra", "model_runtime": {"status": "observed"}, "canonical_database": {"included": False}, "components": { "conversation_sessions": { "role": "session state", "mutability": "dynamic", "replayability": "hashable", "content": { "files": [{"path": "/home/teleo/private/session.json", "sha256": "b" * 64}], "missing": [], "symlinks": [], "file_count": 1, "total_bytes": 42, "sha256": "c" * 64, }, } }, "hermes_runtime": { "git_head": "d" * 40, "git_state": {"worktree_clean": True}, "source_tree": { "files": [{"path": "/home/teleo/private/runtime.py", "sha256": "e" * 64}], "missing": [], "file_count": 1, "total_bytes": 84, "sha256": "f" * 64, }, }, "teleo_infrastructure_runtime": { "git_head": "1" * 40, "git_state": {"worktree_clean": True}, "source_tree": { "files": [{"path": "/opt/teleo-eval/private/runner.py", "sha256": "2" * 64}], "missing": [], "file_count": 1, "total_bytes": 126, "sha256": "3" * 64, }, }, } role = { "role": "leo", "runtime_kind": "gatewayrunner_temp_profile_no_model_tools", "profile_realpath": "/tmp/leo-agent-challenger-loop-test/leo-profile", } role["profile_realpath_sha256"] = runner._sha256_text(role["profile_realpath"]) role["contract_sha256"] = runner.protocol.compute_runtime_role_contract_sha256(role) original_contract = role["contract_sha256"] raw = { "remote_report_path": "/tmp/leo-agent-challenger-loop-test.json", "traceback_tail": [ ' File "/home/teleo/private/runtime.py", line 42, in exchange', "RuntimeError: semantic generation failed", ], "execution_transport": { "recovery": { "remote_path": "/tmp/leo-agent-challenger-loop-test.json", "recovered_report_sha256": "5" * 64, } }, "isolation": {"roles": {"leo": role}}, "live_behavior_manifest_before": manifest, "live_behavior_manifest_after": manifest, } retained = runner.sanitize_report_for_retention(raw) encoded = json.dumps(retained, sort_keys=True) assert "/home/" not in encoded assert "/opt/" not in encoded assert "/tmp/" not in encoded assert raw["remote_report_path"].startswith("/tmp/") assert retained["live_behavior_manifest_before"]["components"]["conversation_sessions"]["content"] == { "file_count": 1, "missing_count": 0, "sha256": "c" * 64, "symlink_count": 0, "total_bytes": 42, } assert retained["live_behavior_manifest_before"]["retention_projection"]["removed_file_entries"] == 3 retained_role = retained["isolation"]["roles"]["leo"] assert retained_role["profile_realpath"] is None assert retained_role["profile_realpath_sha256"] == runner._sha256_text( "/tmp/leo-agent-challenger-loop-test/leo-profile" ) assert retained_role["contract_sha256"] == runner.protocol.compute_runtime_role_contract_sha256(retained_role) assert retained["retention_projection"]["original_runtime_role_contract_sha256"]["leo"] == (original_contract) assert retained["retention_projection"]["raw_remote_report_sha256"] == "5" * 64 assert retained["traceback_tail"] == [ ' File "[absolute-runtime-path-redacted]", line 42, in exchange', "RuntimeError: semantic generation failed", ] assert retained["retention_projection"]["traceback_tail"] == { "raw_traceback_tail_sha256": runner.protocol.canonical_sha256(raw["traceback_tail"]), "line_count": 2, "absolute_runtime_path_occurrence_count": 1, } assert retained["retention_projection"]["profile_realpath_hash_verified"] == {"leo": True} assert runner.sanitize_report_for_retention(retained) == retained retained["remote_stderr"] = "traceback from /home/teleo/private/runtime.py" with pytest.raises(ValueError, match="forbidden absolute runtime path"): runner.sanitize_report_for_retention(retained) def test_retention_projection_fails_closed_on_mismatched_path_hash_or_unexpected_path() -> None: mismatched_role = { "role": "leo", "profile_realpath": "/tmp/leo-profile", "profile_realpath_sha256": "0" * 64, } mismatched_role["contract_sha256"] = runner.protocol.compute_runtime_role_contract_sha256(mismatched_role) with pytest.raises(ValueError, match="profile realpath hash does not match"): runner.sanitize_report_for_retention({"isolation": {"roles": {"leo": mismatched_role}}}) with pytest.raises(ValueError, match="forbidden absolute runtime path"): runner.sanitize_report_for_retention({"remote_stderr": "traceback from /home/teleo/private/runtime.py"}) def test_write_outputs_retains_private_raw_checkpoint_when_projection_fails( monkeypatch: pytest.MonkeyPatch, tmp_path ) -> None: recovery_dir = tmp_path / "recovery" monkeypatch.setattr(runner, "LOCAL_RECOVERY_DIR", recovery_dir) remote = { "run_id": "abc123abc123", "parsed": {"unexpected_runtime_path": "/home/teleo/private/runtime.py"}, "stderr": "remote stderr", } with pytest.raises(ValueError, match="forbidden absolute runtime path"): runner.write_outputs(remote) checkpoints = list(recovery_dir.glob("*.json")) assert len(checkpoints) == 1 assert checkpoints[0].stat().st_mode & 0o777 == 0o600 assert json.loads(checkpoints[0].read_text(encoding="utf-8")) == remote def test_write_outputs_removes_checkpoint_and_retains_only_stderr_hash( monkeypatch: pytest.MonkeyPatch, tmp_path ) -> None: recovery_dir = tmp_path / "recovery" report_dir = tmp_path / "reports" monkeypatch.setattr(runner, "LOCAL_RECOVERY_DIR", recovery_dir) monkeypatch.setattr(runner, "REPORT_DIR", report_dir) monkeypatch.setattr(runner, "OUTPUT_JSON", report_dir / "source.json") monkeypatch.setattr(runner, "RECEIPT_JSON", report_dir / "receipt.json") monkeypatch.setattr(runner, "NEGATIVE_JSON", report_dir / "negative.json") monkeypatch.setattr(runner, "LEDGER_JSONL", report_dir / "ledger.jsonl") monkeypatch.setattr( runner.protocol, "verify_report", lambda *_args, **_kwargs: { "status": "pass", "required_tier": "T2", "current_tier": "T2", "proposal_packet_sha256": "a" * 64, "negative_controls": [{"status": "caught"}], }, ) stderr = "traceback from /home/teleo/private/runtime.py" report, receipt = runner.write_outputs({"run_id": "abc123abc123", "parsed": {}, "returncode": 0, "stderr": stderr}) assert receipt["status"] == "pass" assert "remote_stderr" not in report assert report["remote_stderr_receipt"] == { "present": True, "sha256": runner._sha256_text(stderr), "bytes": len(stderr.encode("utf-8")), "line_count": 1, } assert not recovery_dir.exists()