"""Snapshots survive real actor eviction without retaining opaque replay binaries."""

import json
import os
from pathlib import Path
import socket
import subprocess
import unittest

from harness import Albedo, Provider, Reply, ROOT, exclusive


@exclusive
class ContextRetentionTest(unittest.TestCase):
    def test_eviction_releases_replay_while_inspection_survives(self):
        payload = "REPLAY_SENTINEL:" + "x" * (256 * 1024)
        reply = Reply(
            "raw",
            events=[
                {
                    "type": "response.completed",
                    "response": {
                        "id": "retention-response",
                        "status": "completed",
                        "output": [
                            {
                                "type": "reasoning",
                                "id": "retention-reasoning",
                                "summary": [],
                                "encrypted_content": payload,
                            },
                            {
                                "type": "message",
                                "role": "assistant",
                                "content": [{"type": "output_text", "text": "done"}],
                            },
                        ],
                        "usage": {"input_tokens": 31, "output_tokens": 2},
                    },
                }
            ],
        )
        provider = Provider(lambda _: reply)
        self.addCleanup(provider.close)

        def prepare(app):
            app.daemon.env["ALBEDO_INSPECT"] = "1"
            app.write_extensions({"rolling": {"contextWindowTokens": 2_000_000}})

        with Albedo(provider, protocol="responses", prepare=prepare) as app:
            session = app.session()
            for prompt in ("first", "second"):
                app.prompt(session, prompt).close()
                app.idle(session)
            self.assertEqual(len(provider.requests), 2)
            self.assertTrue(
                any(
                    item.get("encrypted_content") == payload
                    for item in provider.requests[1]["request"]["input"]
                ),
                "second request lost opaque replay",
            )
            route = f"/sessions/{session}/context"

            def read(path):
                with app.api(path) as response:
                    return json.load(response)

            self.assertEqual(read(route)["state"], "ready")
            support = ROOT / "test/daemon/albedo_context_snapshot_probe.erl"
            subprocess.run(
                ["erlc", "-o", str(app.root), str(support)], check=True, timeout=30
            )
            cookie = (app.home / "inspect.cookie").read_text().strip()
            node = f"albedo_{app.daemon._pid}@{socket.gethostname().split('.')[0]}"
            beam = Path(app.root) / "albedo_context_snapshot_probe"
            expression = (
                f"Node = '{node}', "
                f"{{ok, Binary}} = file:read_file({json.dumps(str(beam) + '.beam')}), "
                "{module, albedo_context_snapshot_probe} = "
                "rpc:call(Node, code, load_binary, "
                '[albedo_context_snapshot_probe, "probe.erl", Binary]), '
                "io:put_chars(rpc:call(Node, albedo_context_snapshot_probe, "
                f"actor_json, [<<{json.dumps(session)}>>])), halt()."
            )
            result = subprocess.run(
                [
                    "erl",
                    "+S",
                    "2:2",
                    "-sname",
                    "retention_probe",
                    "-setcookie",
                    cookie,
                    "-noshell",
                    "-eval",
                    expression,
                ],
                capture_output=True,
                text=True,
                check=False,
                timeout=30,
            )
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            measured = json.loads(result.stdout)
            artifact = os.environ.get("ALBEDO_INSPECTOR_ARTIFACT")
            if artifact:
                Path(artifact).write_text(json.dumps(measured, indent=2) + "\n")
            self.assertTrue(measured["history_unloaded"])
            self.assertFalse(measured["replay_reachable"])
            self.assertLess(measured["actor_binary_bytes"], len(payload))
            for key in (
                "history_readable",
                "kernel_released",
                "context_cleared",
                "actor_stopped",
                "workers_terminated",
            ):
                self.assertTrue(measured[key], key)
            self.assertEqual(measured["replay_backing_binary_bytes"], 0)
            self.assertEqual(len(provider.requests), 2)