"""Unread background job results wake idle sessions once, even across idle reaping."""

import time
import unittest

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

# A daemon under this limit sweeps every half second, so the detached job
# below outlives several sweeps.
IDLE_SECONDS = 2


def user_text(request):
    return next(
        (
            item.get("content", "")
            for item in reversed(request["messages"])
            if item.get("role") == "user"
        ),
        "",
    )


def wait_for(predicate, timeout=40):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        result = predicate()
        if result:
            return result
        time.sleep(0.1)
    raise AssertionError("background job did not wake session")


class JobWakeCase(unittest.TestCase):
    idle_seconds = None

    def setUp(self):
        def script(request):
            last = request["messages"][-1]
            user = user_text(request)
            if last.get("role") == "user" and "start a slow job" in user:
                seconds = IDLE_SECONDS * 2.5 if "detached" in user else 1.2
                code = (
                    "import sys\njob = run(sys.executable, '-c', "
                    f"\"import time; time.sleep({seconds}); print('wake-done')\")\njob.id"
                )
                return python(code)
            return text("finished")

        def prepare(app):
            if self.idle_seconds:
                app.env["ALBEDO_IDLE_SECONDS"] = str(self.idle_seconds)

        self.provider = Provider(script)
        self.addCleanup(self.provider.close)
        self.app = Albedo(self.provider, prepare=prepare)
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)

    def users(self):
        return [user_text(record["request"]) for record in self.provider.requests]


class JobWakeTests(JobWakeCase):
    def test_unread_job_wakes_once_with_live_source_and_durable_turn(self):
        session = self.app.session()
        self.app.prompt(session, "start a slow job").close()
        self.app.idle(session)
        self.assertEqual(len(self.users()), 2)
        before = self.app.stream_page(session)
        wait_for(lambda: len(self.users()) >= 3)
        self.app.idle(session)
        self.assertEqual(len(self.users()), 3)
        wake = self.users()[2]
        self.assertIn("background job finished", wake)
        self.assertIn("jobs[", wake)
        self.assertIn("output.read", wake)
        live = self.app.stream_page(session, before)["events"]
        notices = [
            event
            for event in live
            if event.get("type") == "user" and "job finished" in event.get("text", "")
        ]
        self.assertEqual(len(notices), 1)
        self.assertEqual(notices[0]["source"], "job")
        self.assertEqual(notices[0]["clientId"], "job")
        self.assertIn("exit_code=0", notices[0]["text"])
        durable = [
            event
            for event in self.app.events(session)
            if event.get("type") == "user" and "job finished" in event.get("text", "")
        ]
        self.assertEqual(len(durable), 1)
        time.sleep(2.5)
        self.assertEqual(len(self.users()), 3)


@exclusive
class IdleSweepTests(JobWakeCase):
    idle_seconds = IDLE_SECONDS

    def test_live_job_survives_idle_kernel_sweep(self):
        session = self.app.session()
        self.app.prompt(session, "start a slow job, detached").close()
        self.app.idle(session)
        self.assertEqual(len(self.users()), 2)
        # No daemon contact while detached: a live job must pin its kernel.
        wait_for(lambda: len(self.users()) >= 3)
        self.app.idle(session)
        self.assertEqual(len(self.users()), 3)
        self.assertIn("background job finished", self.users()[2])