"""A replay cursor cannot skip durable events from a replacement session actor."""

import json
import socket
import sqlite3
import subprocess
import unittest

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


def restart_actor(app, session):
    support = ROOT / "test/e2e/albedo_stream_replay_probe.erl"
    subprocess.run(
        ["erlc", "-Werror", "-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 = app.root / "albedo_stream_replay_probe.beam"
    expression = (
        f"Node = '{node}', "
        f"{{ok, Binary}} = file:read_file({json.dumps(str(beam))}), "
        "{module, albedo_stream_replay_probe} = "
        "rpc:call(Node, code, load_binary, "
        '[albedo_stream_replay_probe, "probe.erl", Binary]), '
        '<<"actor_stopped">> = rpc:call(Node, albedo_stream_replay_probe, '
        f"restart, [<<{json.dumps(session)}>>]), "
        'io:put_chars("actor_stopped"), halt().'
    )
    result = subprocess.run(
        [
            "erl",
            "+S",
            "2:2",
            "-sname",
            "stream_replay_probe",
            "-setcookie",
            cookie,
            "-noshell",
            "-eval",
            expression,
        ],
        capture_output=True,
        text=True,
        check=True,
        timeout=30,
    )
    if result.stdout != "actor_stopped":
        raise AssertionError(result.stdout + result.stderr)


class StreamReplayTest(unittest.TestCase):
    def setUp(self):
        self.provider = Provider(lambda _: text("answer"))
        self.addCleanup(self.provider.close)

    def turn(self, app, session, prompt):
        app.prompt(session, prompt).close()
        app.idle(session)

    def assert_reset_history(self, page, prompts):
        self.assertEqual(page["events"][0]["type"], "reset")
        self.assertEqual(
            [event["text"] for event in page["events"] if event["type"] == "user"],
            prompts,
        )

    @exclusive
    def test_old_actor_cursor_inside_replacement_window_resets_durable_history(self):
        def prepare(app):
            app.daemon.env["ALBEDO_INSPECT"] = "1"

        with Albedo(self.provider, prepare=prepare) as app:
            session = app.session()
            self.turn(app, session, "before actor restart")
            old = app.stream_page(session)
            restart_actor(app, session)
            self.turn(app, session, "after actor restart")
            current = app.stream_page(session)
            self.assertGreaterEqual(current["cursor"], old["cursor"])
            replay = app.stream_page(session, old)
            self.assert_reset_history(
                replay, ["before actor restart", "after actor restart"]
            )
            self.assertNotEqual(replay["generation"], old["generation"])
            self.assertEqual(replay["generation"], current["generation"])

    @exclusive
    def test_old_daemon_cursor_resets_durable_history(self):
        with Albedo(self.provider) as app:
            session = app.session()
            self.turn(app, session, "before daemon restart")
            old = app.stream_page(session)
            app.restart(crash=True)
            self.turn(app, session, "after daemon restart")
            current = app.stream_page(session)
            self.assertGreaterEqual(current["cursor"], old["cursor"])
            replay = app.stream_page(session, old)
            self.assert_reset_history(
                replay, ["before daemon restart", "after daemon restart"]
            )
            self.assertNotEqual(replay["generation"], old["generation"])

    def test_same_generation_replays_only_subsequent_events_in_order(self):
        with Albedo(self.provider) as app:
            session = app.session()
            self.turn(app, session, "already consumed")
            old = app.stream_page(session)
            self.turn(app, session, "next prompt")
            self.turn(app, session, "last prompt")
            replay = app.stream_page(session, old)
            self.assertEqual(replay["generation"], old["generation"])
            self.assertGreater(replay["cursor"], old["cursor"])
            self.assertNotIn("reset", [event["type"] for event in replay["events"]])
            self.assertEqual(
                [
                    event["text"]
                    for event in replay["events"]
                    if event["type"] == "user"
                ],
                ["next prompt", "last prompt"],
            )
            keepalive = app.stream_page(session, replay)
            self.assertEqual(keepalive["events"], [])
            self.assertEqual(keepalive["generation"], replay["generation"])
            self.assertEqual(keepalive["cursor"], replay["cursor"])

    @exclusive
    def test_failed_transcript_read_emits_failure_without_replacement_cursor(self):
        with Albedo(self.provider) as app:
            session = app.session()
            self.turn(app, session, "committed history")
            old = app.stream_page(session)
            with sqlite3.connect(app.home / "albedo.sqlite") as database:
                database.execute(
                    "ALTER TABLE transcript RENAME TO unavailable_transcript"
                )
            with app.api(
                f"/sessions/{session}/stream?after_generation=other&after_seq={old['cursor']}&tail=2"
            ) as response:
                lines = list(response)
            self.assertIn(b"event: error\n", lines)
            batches = [
                json.loads(line[6:]) for line in lines if line.startswith(b"data: ")
            ]
            self.assertEqual(len(batches), 1)
            self.assertIn("could not load transcript", batches[0]["error"])
            self.assertNotIn("generation", batches[0])
            self.assertNotIn("cursor", batches[0])

    def test_evicted_cursor_resets_in_same_generation_and_initial_tail_pages_history(
        self,
    ):
        with Albedo(self.provider) as app:
            session = app.session()
            self.turn(app, session, "oldest prompt")
            old = app.stream_page(session)
            # A long streamed answer evicts the count-limited replay window.
            self.provider.script = lambda _: text("long answer " * 600)
            self.turn(app, session, "newest prompt")
            replay = app.stream_page(session, old)
            self.assertEqual(replay["generation"], old["generation"])
            self.assert_reset_history(replay, ["oldest prompt", "newest prompt"])
            initial = app.stream_page(session, tail=2)
            self.assert_reset_history(initial, ["newest prompt"])
            reset = initial["events"][0]
            self.assertTrue(reset["more"])
            with app.api(
                f"/sessions/{session}/history?rows=2&before={reset['before']}"
            ) as response:
                older = json.load(response)
            self.assertEqual(
                [event["text"] for event in older["events"] if event["type"] == "user"],
                ["oldest prompt"],
            )