diff --git a/scripts/run_leo_agent_challenger_loop.py b/scripts/run_leo_agent_challenger_loop.py index ef53574..b385c2a 100644 --- a/scripts/run_leo_agent_challenger_loop.py +++ b/scripts/run_leo_agent_challenger_loop.py @@ -7,8 +7,10 @@ import argparse import copy import hashlib import json +import os import re import subprocess +import tempfile import uuid from datetime import datetime, timezone from pathlib import Path @@ -32,6 +34,7 @@ DEFAULT_MAX_TOKENS = 768 DEFAULT_REPORT_PREFIX = "leo-agent-challenger-loop" REMOTE_REPORT_FETCH_ATTEMPTS = 3 REMOTE_REPORT_FETCH_TIMEOUT_SECONDS = 45 +LOCAL_RECOVERY_DIR = Path(tempfile.gettempdir()) / "teleo-agent-challenger-loop-recovery" RETENTION_PROJECTION_SCHEMA = "livingip.leoAgentChallengerRetentionProjection.v1" BEHAVIOR_RETENTION_SCHEMA = "livingip.leoBehaviorManifestRetentionProjection.v1" RETENTION_FORBIDDEN_PATTERNS = ( @@ -1567,14 +1570,52 @@ def assert_report_safe_for_retention(report: dict[str, Any]) -> None: raise ValueError(f"retained report contains forbidden {label}") +def write_local_recovery_checkpoint(remote: dict[str, Any]) -> Path: + """Persist the fetched raw report privately until durable projection succeeds.""" + + LOCAL_RECOVERY_DIR.mkdir(mode=0o700, parents=True, exist_ok=True) + LOCAL_RECOVERY_DIR.chmod(0o700) + run_id = str(remote.get("run_id") or "") + safe_run_id = run_id if re.fullmatch(r"[0-9a-f]{12}", run_id) else _sha256_text(run_id)[:12] + checkpoint = LOCAL_RECOVERY_DIR / f"{DEFAULT_REPORT_PREFIX}-raw-recovery-{safe_run_id}.json" + descriptor = os.open(checkpoint, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump(remote, handle, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + checkpoint.chmod(0o600) + except BaseException: + checkpoint.unlink(missing_ok=True) + raise + return checkpoint + + +def remove_local_recovery_checkpoint(checkpoint: Path) -> None: + checkpoint.unlink() + try: + LOCAL_RECOVERY_DIR.rmdir() + except OSError: + # Preserve any checkpoint from a different run or concurrent process. + pass + + def write_outputs(remote: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: + recovery_checkpoint = write_local_recovery_checkpoint(remote) REPORT_DIR.mkdir(parents=True, exist_ok=True) - report = remote.get("parsed") if isinstance(remote.get("parsed"), dict) else {} + report = copy.deepcopy(remote.get("parsed")) if isinstance(remote.get("parsed"), dict) else {} report["remote_returncode"] = remote.get("returncode") report["remote_run_id"] = remote.get("run_id") report["execution_transport"] = remote.get("execution_transport") if remote.get("stderr"): - report["remote_stderr"] = remote["stderr"] + remote_stderr = str(remote["stderr"]) + report["remote_stderr_receipt"] = { + "present": True, + "sha256": _sha256_text(remote_stderr), + "bytes": len(remote_stderr.encode("utf-8")), + "line_count": len(remote_stderr.splitlines()), + } report = sanitize_report_for_retention(report) OUTPUT_JSON.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") source_sha = hashlib.sha256(OUTPUT_JSON.read_bytes()).hexdigest() @@ -1607,6 +1648,7 @@ def write_outputs(remote: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any } with LEDGER_JSONL.open("a", encoding="utf-8") as handle: handle.write(json.dumps(ledger_event, sort_keys=True) + "\n") + remove_local_recovery_checkpoint(recovery_checkpoint) return report, receipt diff --git a/tests/test_run_leo_agent_challenger_loop.py b/tests/test_run_leo_agent_challenger_loop.py index f49d6d3..849ac47 100644 --- a/tests/test_run_leo_agent_challenger_loop.py +++ b/tests/test_run_leo_agent_challenger_loop.py @@ -236,3 +236,60 @@ def test_retention_projection_fails_closed_on_mismatched_path_hash_or_unexpected 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()