"""Releasing many kernels at once on a saturated machine confirms every stop.

A stop ends the kernel's process groups through a helper. When the helper was a
fresh Python interpreter per stop, a burst of stops competed with the kernel
boots for the CPU, missed its deadline, and left the session refusing to open
("previous kernel stop is unconfirmed"). Only a real daemon with real kernels
under CPU load shows that, so this runs through an inspected daemon.
"""

import json
import os
import subprocess
import sys
import unittest

from harness import ROOT, Albedo, Provider, exclusive, python, text
from inspect_support import Inspection

SESSIONS = int(os.environ.get("STOP_LOAD_SESSIONS", "24"))
BURNERS = int(os.environ.get("STOP_LOAD_BURNERS", str(os.cpu_count() or 4)))


@exclusive
class StopUnderLoadTests(unittest.TestCase):
    def test_a_burst_of_stops_on_a_busy_machine_confirms_every_kernel(self):
        def script(request):
            if request["messages"][-1].get("role") == "tool":
                return text("ready")
            return python("sentinel = 1")

        provider = Provider(script)
        self.addCleanup(provider.close)

        def prepare(app):
            app.daemon.env["ALBEDO_INSPECT"] = "1"

        with Albedo(provider, prepare=prepare) as app:
            sessions = [app.session() for _ in range(SESSIONS)]
            for session in sessions:
                app.prompt(session, "boot").close()
            for session in sessions:
                app.idle(session)
            burners = [
                subprocess.Popen([sys.executable, "-c", "while True: pass"])
                for _ in range(BURNERS)
            ]
            try:
                probe = Inspection(app, ROOT / "test/e2e/albedo_stop_load_probe.erl")
                ids = ",".join(f'<<"{session}">>' for session in sessions)
                report = json.loads(
                    probe.evaluate(
                        "io:put_chars(Call(albedo_stop_load_probe, mass_json, "
                        f'[[{ids}], <<"{app.workspace}">>]))',
                        timeout=180,
                    )
                )
            finally:
                for burner in burners:
                    burner.kill()
            self.assertEqual(report["failed"], 0, report)
            log = (app.home / "daemon.log").read_text(errors="replace")
            self.assertNotIn("termination helper", log)