Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263"""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 afresh Python interpreter per stop, a burst of stops competed with the kernelboots for the CPU, missed its deadline, and left the session refusing to open("previous kernel stop is unconfirmed"). Only a real daemon with real kernelsunder CPU load shows that, so this runs through an inspected daemon."""
import jsonimport osimport subprocessimport sysimport unittest
from harness import ROOT, Albedo, Provider, exclusive, python, textfrom 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)))
@exclusiveclass 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)