"""Slow reader admission and kernel teardown must not stall the session actor."""

import json
import time
import threading
import unittest

from harness import (
    no_spares,
    ROOT,
    Albedo,
    Provider,
    exclusive,
    latest_user,
    operation_id,
    python,
    text,
)
from inspect_support import Inspection


@exclusive
class SessionActorStallTests(unittest.TestCase):
    def prepare(self, app):
        app.daemon.env["ALBEDO_INSPECT"] = "1"

    def binary(self, value):
        return "<<" + json.dumps(str(value)) + ">>"

    def probe(self, app):
        return Inspection(app, ROOT / "test/e2e/albedo_runtime_stall_probe.erl")

    def wait_file(self, path):
        deadline = time.monotonic() + 20
        while not path.exists():
            self.assertLess(time.monotonic(), deadline, str(path))
            time.sleep(0.02)
        return path.read_text()

    def test_capture_waiting_for_a_reader_keeps_the_actor_responsive(self):
        with Albedo(prepare=self.prepare) as app:
            probe = self.probe(app)
            storage = Inspection(
                app, ROOT / "test/e2e/albedo_storage_concurrency_probe.erl"
            )
            session = app.session()
            # Load the actor before occupying the pool.
            with app.api(f"/sessions/{session}") as response:
                response.read()
            marker = app.root / "captured"
            storage.call("hold_readers", f"[{self.binary(app.root / 'held')}]")
            try:
                probe.call(
                    "start", f"[{self.binary(session)}, capture, {self.binary(marker)}]"
                )
                self.assertEqual(probe.call("wait_capture"), "queued")
                self.assertEqual(
                    probe.call_json("summary_json", f"[{self.binary(session)}]"),
                    {"responsive": True},
                )
            finally:
                storage.call("release")
            self.assertTrue(self.wait_file(marker).startswith("{ok,"))

    # no spares: the test stalls the runtime and releases a kernel it opened itself
    @no_spares
    def test_release_does_not_wait_for_runtime_and_reset_precedes_next_open(self):
        def script(request):
            if request["messages"][-1].get("role") == "tool":
                return text("done")
            if "restore" in latest_user(request):
                return python("assert saved_value == 42\nprint('restored')")
            return python("saved_value = 42")

        provider = Provider(script)
        self.addCleanup(provider.close)
        with Albedo(provider, prepare=self.prepare) as app:
            probe = self.probe(app)
            session = app.session()
            app.prompt(session, "save").close()
            app.idle(session)
            with app.api(f"/sessions/{session}") as response:
                original_kernel = json.load(response)["kernel"]["instance_id"]
            marker = app.root / "released"
            probe.call_json("suspend_json")
            try:
                probe.call(
                    "start", f"[{self.binary(session)}, release, {self.binary(marker)}]"
                )
                self.assertEqual(probe.call("wait_queued", "[reset]"), "queued")
                self.assertEqual(
                    probe.call_json("summary_json", f"[{self.binary(session)}]"),
                    {"responsive": True},
                )
                self.assertEqual(self.wait_file(marker), "true")
                # Admission may queue a boot, but must never overtake Reset.
                app.prompt(session, "restore").close()
            finally:
                probe.call_json("resume_json")
            app.idle(session)
            history = app.history(session)["items"]
            self.assertNotIn("NameError", json.dumps(history))
            self.assertNotIn("AssertionError", json.dumps(history))
            self.assertIn("restored", json.dumps(history))
            with app.api(f"/sessions/{session}") as response:
                self.assertNotEqual(
                    json.load(response)["kernel"]["instance_id"], original_kernel
                )

    def test_awaited_observation_keeps_the_actor_responsive(self):
        with Albedo(prepare=self.prepare) as app:
            probe = self.probe(app)
            session = app.session()
            with app.api(f"/sessions/{session}") as response:
                response.read()
            marker = app.root / "awaited"
            probe.call_json("suspend_json")
            try:
                probe.call(
                    "start", f"[{self.binary(session)}, awaited, {self.binary(marker)}]"
                )
                self.assertEqual(
                    probe.call("wait_queued", "[observe_loaded]"), "queued"
                )
                self.assertEqual(
                    probe.call_json("summary_json", f"[{self.binary(session)}]"),
                    {"responsive": True},
                )
            finally:
                probe.call_json("resume_json")
            self.assertEqual(self.wait_file(marker), "false")

    def test_late_kernel_announcement_does_not_query_runtime(self):
        with Albedo(prepare=self.prepare) as app:
            probe = self.probe(app)
            session = app.session()
            with app.api(f"/sessions/{session}") as response:
                response.read()
            probe.call_json("suspend_json")
            try:
                probe.call("announce", f"[{self.binary(session)}]")
                self.assertEqual(
                    probe.call_json("summary_json", f"[{self.binary(session)}]"),
                    {"responsive": True},
                )
            finally:
                probe.call_json("resume_json")

    def test_move_waits_for_teardown_without_blocking_admission(self):
        def script(request):
            if request["messages"][-1].get("role") == "tool":
                return text("done")
            if "after move" in latest_user(request):
                return python(
                    "assert 'old_workspace_value' not in globals()\nprint(os.getcwd())"
                )
            return python("old_workspace_value = 42")

        provider = Provider(script)
        self.addCleanup(provider.close)
        with Albedo(provider, prepare=self.prepare) as app:
            probe = self.probe(app)
            session = app.session()
            app.prompt(session, "before move").close()
            app.idle(session)
            destination = app.root / "destination"
            destination.mkdir()
            family = app.family_revision(session)
            with app.api(f"/sessions/{session}?view=configuration") as response:
                response.read()
                etag = response.headers["ETag"]
            moved = []

            def move():
                try:
                    with app.api(
                        f"/sessions/{session}?view=configuration",
                        {"workspace": str(destination), "family_revision": family},
                        method="PATCH",
                        headers={"If-Match": etag},
                    ) as response:
                        moved.append(json.load(response))
                except Exception as error:
                    moved.append(error)

            probe.call_json("suspend_json")
            worker = threading.Thread(target=move)
            try:
                worker.start()
                self.assertEqual(probe.call("wait_queued", "[delete]"), "queued")
                self.assertEqual(
                    probe.call_json("summary_json", f"[{self.binary(session)}]"),
                    {"responsive": True},
                )
                app.prompt(session, "after move").close()
            finally:
                probe.call_json("resume_json")
                worker.join(30)
            self.assertEqual(len(moved), 1, moved)
            self.assertIsInstance(moved[0], dict, moved)
            app.idle(session)
            history = json.dumps(app.history(session)["items"])
            self.assertNotIn("AssertionError", history)
            self.assertIn(str(destination), history)
            with app.api(f"/sessions/{session}") as response:
                self.assertEqual(json.load(response)["workspace"], str(destination))

    def test_unload_checks_children_without_blocking_the_parent(self):
        provider = Provider(lambda _: text("done"))
        self.addCleanup(provider.close)
        with Albedo(provider, prepare=self.prepare) as app:
            probe = self.probe(app)
            parent = app.session()
            child = operation_id()
            with app.api(
                f"/sessions/{child}",
                {
                    "kind": "child",
                    "parent_id": parent,
                    "address": "worker",
                    "name": "worker",
                    "initial_input_id": operation_id(),
                    "task": "done",
                },
                method="PUT",
                headers={"If-None-Match": "*"},
            ) as response:
                response.read()
            app.idle(child)
            app.idle(parent)
            marker = app.root / "unloaded"
            probe.call_json("suspend_json")
            try:
                probe.call(
                    "start", f"[{self.binary(parent)}, unload, {self.binary(marker)}]"
                )
                self.assertEqual(
                    probe.call("wait_queued", "[observe_loaded]"), "queued"
                )
                self.assertEqual(
                    probe.call_json("summary_json", f"[{self.binary(parent)}]"),
                    {"responsive": True},
                )
            finally:
                probe.call_json("resume_json")
            self.assertEqual(self.wait_file(marker), "true")

    def test_capture_loans_cannot_exhaust_the_actors_reader_pool(self):
        with Albedo(prepare=self.prepare) as app:
            probe = self.probe(app)
            session = app.session()
            with app.api(f"/sessions/{session}") as response:
                response.read()
            self.assertEqual(
                probe.call(
                    "capture_lock_order",
                    f"[{self.binary(session)}, {self.binary(app.root / 'lock-order')}]",
                ),
                "captured",
            )