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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621"""Fresh spare kernels are single-use, bounded, and never adopted across a bundle edit."""
import astimport jsonimport osimport sqlite3from contextlib import closingfrom pathlib import Pathimport socketimport subprocessimport sysimport timeimport unittestimport urllib.parse
from harness import ( Albedo, Provider, alive, daemon_resource, exclusive, exclusive_class, python, text, wait_until,)from inspect_support import Inspectionfrom kernel_support import MARKER, ModelCells, editable_daemon, process_name
# An ssh that runs its command here with no environment of its own. Host# `oldhost` reaches a python too old for a kernel, so remote.connect degrades# to command mode: a connection that holds no process or descriptor.FAKE_SSH = """#!/bin/shwhile [ $# -gt 0 ]; do case "$1" in -o) shift 2 ;; -*) shift ;; *) break ;; esacdonetarget=$1shiftpath={path}[ "$target" = oldhost ] && path={old}:$pathexec env -i HOME={home} PATH="$path" SHELL={old}/loginsh /bin/sh -c "$*""""
# exclusive: boots an editable daemon copy with its spare pool configured@exclusive_classclass KernelPoolTests(ModelCells, unittest.TestCase): bundle: Path
def boot(self, *, spares=1, ssh=False, zygote=None, scale=None): self.cells = []
def script(request): if request["input"][-1].get("role") == "user" and self.cells: return python(self.cells.pop(0)) return text("ok")
def prepare(app): launcher, self.bundle = daemon_resource( "editable daemon", lambda: editable_daemon(app.root) ) app.env.update( ALBEDO_DAEMON=str(launcher), ALBEDO_INSPECT="1", ALBEDO_KERNEL_SPARES=str(spares), ) if spares: # A zero memory budget is constant pressure: every idle kernel # is released, and released sessions waking keep spares warm. app.env["ALBEDO_KERNEL_BUDGET_MB"] = "0" if zygote is not None: app.env["ALBEDO_KERNEL_ZYGOTE"] = zygote if scale is not None: app.env["ALBEDO_TEST_TIME_SCALE"] = scale if ssh: # On the daemon's PATH, so a kernel's environment is unchanged # by connecting and only the connection itself can keep it. local, old, home = (app.root / name for name in ("bin", "old", "rhome")) for directory in (local, old, home): directory.mkdir() path = f"{app.daemon.env['PATH']}" (local / "ssh").write_text( FAKE_SSH.format(path=path, old=old, home=home) ) (old / "python3").write_text("#!/bin/sh\necho 3.9\n") # A login shell that sources no profile, which could reset PATH. (old / "loginsh").write_text( '#!/bin/sh\n[ "$1" = -l ] && shift\nexec /bin/sh "$@"\n' ) for script in (local / "ssh", old / "python3", old / "loginsh"): script.chmod(0o755) app.daemon.env["PATH"] = f"{local}{os.pathsep}{path}"
self.provider = Provider(script) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses", prepare=prepare) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) # A released session waking warms the pool: in a session of its own, # a cell, its release (the saved state), and a second cell. self.session = self.app.session() if spares: self.cell("warm = 1") state = self.app.home / "kernels" / f"{self.session}.state" wait_until(state.exists, 30, "the warming session was not released") self.cell("warm") self.session = self.app.session() self.inspection = Inspection(self.app)
def spare_pids(self): """Pids of ready spares: the runtime's live spare custodians label themselves with their process. An adopted kernel's custodian is gone.""" listed = self.inspection.evaluate( "io:write([Call(erlang, binary_to_integer, [Id]) || P <- Call(erlang, processes, []), " "{albedo_spare, Id} <- [Call(proc_lib, get_label, [P])]])" ) return {int(pid) for pid in ast.literal_eval(listed or "[]") if alive(int(pid))}
def ready(self, *, excluding=()): found = set()
def available(): nonlocal found found = self.spare_pids() - set(excluding) if not found: return False return ( self.inspect_runtime( "io:write(Call('albedo@harness@runtime', spare_available, [Runtime]))" ) == "true" )
wait_until(available, 15, "the runtime did not refill its fresh spare") return found
def inspect_runtime(self, expression): return self.inspection.evaluate( "Names = Call(erlang, registered, []), " '[Name] = [N || N <- Names, lists:prefix("albedo_registry", atom_to_list(N))], ' "Subject = Call('gleam@erlang@process', named_subject, [Name]), " "{ok, Runtime} = Call('albedo@daemon@registry', host, [Subject]), " + expression )
def value(self, code): result, _ = self.cell(code) self.assertEqual(result["status"], "ok", result) return ast.literal_eval(result["value"])
# exclusive: boots with the zygote off @exclusive def test_spare_process_name_changes_on_adoption(self): self.assert_adopted_name("0")
# exclusive: boots with the zygote on @exclusive @unittest.skipUnless(sys.platform == "linux", "fork names use Linux prctl") def test_forked_process_name_changes_on_adoption(self): self.assert_adopted_name("1")
def assert_adopted_name(self, zygote): self.boot(zygote=zygote) spares = self.ready() expected = "albedo-spare" if sys.platform == "linux" else "albedo-kernel" for pid in spares: self.assertEqual(process_name(pid), expected) pid = self.value("hold = run('sleep', '120')\nos.getpid()") self.assertIn(pid, spares) self.assertEqual(process_name(pid), "albedo-kernel")
def test_changed_environment_keeps_its_kernel_without_saving_secrets(self): self.boot() self.ready() pid = self.value( "import os\nos.environ['DENSITY_EPHEMERAL_SECRET']='private'\nos.getpid()" ) time.sleep(1.2) self.assertTrue(alive(pid), "environment changes were lost at turn end") self.assertEqual( self.value("(os.getpid(), os.environ['DENSITY_EPHEMERAL_SECRET'])"), (pid, "private"), ) state = self.app.home / "kernels" / f"{self.session}.state" self.assertFalse(state.exists(), "environment values must not be saved") self.value("del os.environ['DENSITY_EPHEMERAL_SECRET']\n42") wait_until( lambda: not alive(pid), 15, "restoring the environment did not unpin" )
def test_user_thread_survives_turn_end(self): self.boot() self.ready() pid = self.value( "import os, threading\n_gate=threading.Event()\n_worker=threading.Thread(target=_gate.wait, daemon=True)\n_worker.start()\nos.getpid()" ) time.sleep(1.2) self.assertTrue(alive(pid), "a running user thread was terminated") self.assertEqual(self.value("(os.getpid(), _worker.is_alive())"), (pid, True)) self.value("_gate.set()\n_worker.join()\ndel _gate, _worker\n42") wait_until( lambda: not alive(pid), 15, "a finished user thread pinned its kernel" )
def test_listening_socket_survives_turn_end(self): self.boot() self.ready() pid, port = self.value( "import os, socket\n_server=socket.socket()\n_server.bind(('127.0.0.1', 0))\n_server.listen()\n(os.getpid(), _server.getsockname()[1])" ) time.sleep(1.2) self.assertTrue(alive(pid), "a listening server was terminated") with socket.create_connection(("127.0.0.1", port), timeout=2): pass self.assertEqual(self.value("os.getpid()"), pid) self.value("_server.close()\ndel _server\n42") wait_until(lambda: not alive(pid), 15, "a closed server pinned its kernel")
def test_user_timer_survives_turn_end(self): self.boot() self.ready() pid = self.value( "import os\n_timer=asyncio.get_running_loop().call_later(300, lambda: None)\nos.getpid()" ) time.sleep(1.2) self.assertTrue(alive(pid), "a scheduled user callback was discarded") self.assertEqual(self.value("os.getpid()"), pid) self.value("_timer.cancel()\ndel _timer\n42") wait_until( lambda: not alive(pid), 15, "a cancelled user timer pinned its kernel" )
def test_user_async_task_survives_turn_end(self): self.boot() self.ready() pid = self.value( "import os\n_task=asyncio.create_task(asyncio.sleep(300))\nos.getpid()" ) time.sleep(1.2) self.assertTrue(alive(pid), "a user asyncio task was cancelled") self.assertEqual(self.value("(os.getpid(), _task.done())"), (pid, False)) self.value( "_task.cancel()\nawait asyncio.gather(_task, return_exceptions=True)\ndel _task\n42" ) wait_until(lambda: not alive(pid), 15, "a finished user task pinned its kernel")
def test_owned_cua_driver_survives_turn_end(self): from cua_test import FAKE_DRIVER
self.boot() route = f"/sessions/{self.session}?view=configuration" with self.app.api(route) as response: response.read() etag = response.headers["ETag"] self.app.api( route, {"selection": {"extensions": {"cua": True}}}, method="PATCH", headers={"If-Match": etag}, ).close() self.app.api(f"/sessions/{self.session}/reload", {"target": "session"}).close() self.ready() driver = self.app.workspace / "fake-cua-driver" driver.write_text(FAKE_DRIVER) driver.chmod(0o755) pid = self.value( f"cua.configure(executable={str(driver)!r}, socket='/tmp/fake-density-cua.sock')\nawait cua.health()\nos.getpid()" ) time.sleep(1.2) self.assertTrue(alive(pid), "an owned desktop driver was ended") self.assertEqual(self.value("await cua.health()\nos.getpid()"), pid) self.value("await cua.close()\n42") wait_until( lambda: not alive(pid), 15, "a closed desktop driver pinned its kernel" )
def test_owned_browser_survives_turn_end(self): self.boot() self.ready() pid = self.value( "_browser=await browser.spawn()\n_page=await _browser.new_page()\nos.getpid()" ) time.sleep(1.2) self.assertTrue(alive(pid), "an owned browser was ended") self.assertEqual(self.value("await _page.evaluate('1+1')"), 2) self.value("await _browser.close()\ndel _page, _browser\n42") wait_until(lambda: not alive(pid), 15, "a closed browser pinned its kernel")
# exclusive: boots with a fake ssh on its PATH @exclusive def test_open_remote_connection_survives_turn_end(self): self.boot(ssh=True) self.ready() pid = self.value( "_rem = await remote.connect('oldhost')\nassert _rem._stdin is None\nos.getpid()" ) time.sleep(1.2) self.assertTrue(alive(pid), "an open remote connection was ended") self.assertEqual( self.value("(os.getpid(), len(remote.connections()))"), (pid, 1) ) self.value("await _rem.close()\ndel _rem\n42") wait_until( lambda: not alive(pid), 15, "a closed remote connection pinned its kernel" )
def jobs_api(self, path, **query): suffix = ("?" + urllib.parse.urlencode(query)) if query else "" route = f"/extensions/run/sessions/{self.session}/jobs/{path}{suffix}" with self.app.api(route) as response: return json.load(response)
def test_release_keeps_retained_output_and_the_job_book(self): self.boot() self.ready() # A cell whose output spills past the preview (its result is stored # by reference, so it is not read back here), then one naming it. self.cells.append("print('é' * 70000 + 'end')") self.app.prompt(self.session, "run it").close() self.app.idle(self.session) cell, pid, job_id, text = self.value( "import os, sys\n" "job = run(sys.executable, '-c', 'print(\"first \\\\u00e9\"); print(\"second\")')\n" "await job\n" "([c['id'] for c in output.list() if c['bytes'] > 100000][0], os.getpid(), job.id, job.read())" ) self.assertEqual(text, "first é\nsecond\n") wait_until(lambda: not alive(pid), 15, "the resting kernel was not released") # No kernel now: the routes answer from the saved book and wake none. [job] = self.jobs_api("recent")["items"] self.assertEqual( (job["id"], job["state"], job["exit_code"]), (job_id, "exited", 0) ) self.assertEqual(job["bytes"], 16) page = self.jobs_api(f"{job_id}/output", offset=6, limit=4) self.assertEqual( (page["offset"], page["next_offset"], page["text"]), (6, 10, "é\ns") ) self.assertEqual(self.jobs_api(f"{job_id}/output")["text"], text) self.assertEqual(self.spare_pids() & {pid}, set()) restored = self.value( f"(output.read({cell!r}, offset=139990, limit=100), jobs[{job_id!r}].tail(lines=1), " f"jobs[{job_id!r}].exit_code, job.read())" ) self.assertEqual(restored, ("ééééé" + "end\n", "second\n", 0, text))
def test_spare_is_adopted_released_once_and_refilled(self): self.boot() before = self.ready() first = self.value("import os\nsentinel = 41\nos.getpid()") self.assertIn(first, before, "the first cell did not adopt a spare") wait_until(lambda: not alive(first), 15, "the resting kernel was not released") next_spares = self.ready() self.assertTrue(before.isdisjoint(next_spares)) second, sentinel = self.value("(os.getpid(), sentinel)") self.assertIn(second, next_spares) self.assertNotEqual(first, second) self.assertEqual(sentinel, 41) wait_until( lambda: not alive(second), 15, "the adopted kernel returned to the pool" ) self.assertTrue(self.ready().isdisjoint({first, second}))
# exclusive: runs the daemon's timers faster @exclusive def test_ready_spares_exit_after_a_scaled_quiet_minute(self): # Establish a ready spare before starting the accelerated quiet window. # Under load, boot and inspection can consume the entire five seconds. self.boot() self.ready() self.inspect_runtime( "Call(persistent_term, put, [{albedo_clock, scale}, 12]), " "Call('albedo@harness@runtime', want_spares, [Runtime])" ) wait_until( lambda: not self.spare_pids(), 20, "ready spares outlived the quiet minute" )
def test_live_background_work_keeps_the_adopted_kernel(self): self.boot() self.ready() gate = self.app.workspace / "background-gate" program = f"import pathlib, time; p = pathlib.Path({str(gate)!r}); exec('while not p.exists(): time.sleep(0.01)')" pid = self.value( f"import os, sys\njob = run(sys.executable, '-c', {program!r})\nos.getpid()" ) self.assertTrue(alive(pid)) self.assertEqual(self.value("(os.getpid(), job.exit_code)"), (pid, None)) gate.touch() self.value("await job\n42") wait_until(lambda: not alive(pid), 15, "a finished job pinned its kernel")
def zygote_of(self, pid): """The zygote that forked this spare, or None.""" parent = subprocess.check_output( ["ps", "-o", "ppid=", "-p", str(pid)], text=True ).strip() command = subprocess.run( ["ps", "-o", "command=", "-p", parent], capture_output=True, text=True ).stdout return int(parent) if "albedo_zygote.py" in command else None
# exclusive: boots with the zygote on and edits the bundle @exclusive def test_zygote_forks_spares_and_is_recycled_after_a_bundle_edit(self): self.boot(zygote="1") [spare] = self.ready() zygote = self.zygote_of(spare) self.assertIsNotNone(zygote, "the spare was not forked by a zygote") self.assertEqual(self.value("import os\nos.getpid()"), spare) with (self.bundle / "albedo_bundle.py").open("a") as source: source.write(MARKER) # A new session's kernel must run the edited code, not the zygote's. self.session = self.app.session() self.assertEqual( self.value("import albedo_bundle\nalbedo_bundle.MARKER"), "new bundle" ) wait_until( lambda: not alive(zygote), 15, "the stale zygote outlived its bundle" ) # Refusing a fork for a changed bundle is the expected fallback. log = (self.app.home / "daemon.log").read_text(errors="replace") self.assertNotIn("zygote_fork_failed", log)
# exclusive: boots two spares with the zygote on @exclusive def test_spares_warming_together_fork_from_one_zygote(self): self.boot(spares=2, zygote="1") spares = set()
def both(): nonlocal spares spares = self.spare_pids() return len(spares) == 2
wait_until(both, 15, "the runtime did not warm two spares") parents = {self.zygote_of(pid) for pid in spares} self.assertEqual(len(parents), 1, parents) self.assertNotIn(None, parents, "a spare was started without the zygote")
# exclusive: boots with the zygote on and a scaled clock @exclusive def test_zygote_exits_after_a_scaled_idle_minute(self): # At scale 6 its minute without a fork is 10 s; unscaled it would # outlive the 40 s wait. self.boot(zygote="1", scale="6") [spare] = self.ready() zygote = self.zygote_of(spare) self.assertIsNotNone(zygote, "the spare was not forked by a zygote") wait_until(lambda: not alive(zygote), 40, "the idle zygote outlived its minute")
# exclusive: boots with the zygote on @exclusive def test_opens_past_the_spares_fork_from_the_zygote(self): self.boot(zygote="1") [spare] = self.ready() zygote = self.zygote_of(spare) sessions = [self.app.session() for _ in range(3)] for session in sessions: # The kernel may be released once its turn ends, so it names its # parent itself. self.cells.append("import os\nos.getppid()") self.app.prompt(session, "run it").close() pids = [] for session in sessions: self.session = session self.app.idle(session) [result] = [ json.loads(part["value"]) for entry in self.app.history(session)["items"] if entry["kind"] == "tool_result" for part in entry["content"] if part["kind"] == "json" and part["field"] == "result" ] pids.append(ast.literal_eval(result["value"])) self.assertEqual(set(pids), {zygote})
# exclusive: boots with no spares @exclusive def test_an_empty_pool_uses_a_cold_kernel(self): self.boot(spares=0) pid = self.value("import os\nos.getpid()") self.assertTrue(alive(pid)) self.assertEqual(self.spare_pids(), set())
# exclusive: edits the bundle @exclusive def test_a_changed_bundle_is_not_adopted(self): self.boot() before = self.ready() with (self.bundle / "albedo_bundle.py").open("a") as source: source.write(MARKER) pid, marker = self.value( "import os, albedo_bundle\n(os.getpid(), albedo_bundle.MARKER)" ) self.assertNotIn(pid, before) self.assertEqual(marker, "new bundle") wait_until( lambda: all(not alive(old) for old in before), 15, "the stale spare survived its rejected adoption", )
# exclusive: boots with the zygote off and restarts @exclusive def test_replacing_a_record_keeps_the_old_kernel_owned_until_reaped(self): self.boot(zygote="0") self.ready() adopted = self.value("hold = run('sleep', '120')\nos.getpid()") self.app.restart( prepare=lambda app: app.env.update( ALBEDO_TEST_TIME_SCALE="40", ALBEDO_IDLE_SECONDS="400" ) ) probe = Inspection( self.app, Path(__file__).with_name("albedo_kernel_spare_probe.erl") ) probe.call_json("replace_json", f"[<<{json.dumps(self.session)}>>]") with closing(sqlite3.connect(self.app.home / "albedo.sqlite")) as db: orphans = db.execute("SELECT pid FROM kernel_orphans").fetchall() self.assertIn((adopted,), orphans, "replacement lost the old process record") self.assertTrue(alive(adopted), "the orphan ended before the idle limit") wait_until( lambda: not alive(adopted), 20, "the replaced kernel was never reaped" )
# exclusive: boots with no spares and restarts @exclusive def test_startup_recovers_only_unrecorded_matching_kernel_identities(self): self.boot(spares=0) orphaned_session = self.session # The cell can end before the job spawns, and only the spawn writes its # group into identity.json: wait for that, not for the cell. orphaned, orphaned_dir = self.value( "import albedo_cells\nhold = run('sleep', '120')\nawait hold._spawn()\n" "(os.getpid(), albedo_cells.RUN_DIR)" ) identity_path = Path(orphaned_dir) / "identity.json" self.assertEqual(identity_path.stat().st_mode & 0o777, 0o600) [job] = json.loads(identity_path.read_text())["groups"].values() orphaned_job = job["pid"] self.session = self.app.session() stale_session = self.session stale, stale_dir = self.value( "import albedo_cells\nhold = run('sleep', '120')\n" "(os.getpid(), albedo_cells.RUN_DIR)" ) self.session = self.app.session() linked = self.value("sentinel = 41\nhold = run('sleep', '120')\nos.getpid()")
def prepare(app): with closing(sqlite3.connect(app.home / "albedo.sqlite")) as db: db.execute( "DELETE FROM kernel_links WHERE session IN (?,?)", (orphaned_session, stale_session), ) db.commit() stale_path = Path(stale_dir) / "identity.json" identity = json.loads(stale_path.read_text()) identity["leader"] = "not this process incarnation" stale_path.write_text(json.dumps(identity)) app.env.update(ALBEDO_TEST_TIME_SCALE="40", ALBEDO_IDLE_SECONDS="400")
self.app.restart(prepare=prepare) with closing(sqlite3.connect(self.app.home / "albedo.sqlite")) as db: recovered = db.execute("SELECT pid FROM kernel_orphans").fetchall() self.assertIn((orphaned,), recovered) self.assertNotIn((stale,), recovered, "a mismatched leader token was accepted") self.assertNotIn((linked,), recovered, "a linked kernel became an orphan") self.assertTrue(alive(orphaned), "recovery skipped the orphan idle limit") wait_until( lambda: not alive(orphaned), 20, "the recovered kernel was not reaped" ) wait_until( lambda: not alive(orphaned_job), 5, "the recovered kernel's job survived" ) wait_until( lambda: not Path(orphaned_dir).exists(), 5, "the recovered run dir survived" ) self.assertTrue(alive(stale), "a stale identity killed a live process") self.assertEqual(self.value("(os.getpid(), sentinel)"), (linked, 41))
# exclusive: boots with the zygote off and restarts @exclusive def test_killed_daemon_stops_unadopted_spares(self): self.boot(zygote="0") ready = self.ready() self.app.daemon.kill() wait_until( lambda: all(not alive(pid) for pid in ready), 5, "an unused spare survived SIGKILL of its daemon", ) self.app.restart()
def test_shutdown_stops_spares_but_detaches_an_adopted_kernel(self): self.boot() self.ready() # A live job keeps the adopted kernel from being released before the # restart, as a user's running work does. adopted = self.value( "import os\nsentinel = 41\nhold = run('sleep', '120')\nos.getpid()" ) ready = self.ready(excluding={adopted}) self.app.restart() wait_until( lambda: all(not alive(pid) for pid in ready), 15, "a fresh spare survived daemon shutdown", ) self.assertTrue(alive(adopted), "shutdown ended an adopted detached kernel") self.assertEqual( self.value("(os.getpid(), sentinel)"), (adopted, 41), "the restarted daemon did not reattach the adopted kernel", )