Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260"""Slow reader admission and kernel teardown must not stall the session actor."""
import jsonimport timeimport threadingimport unittest
from harness import ( ROOT, Albedo, Provider, exclusive, exclusive_class, latest_user, operation_id, python, text,)from inspect_support import Inspection
@exclusive_classclass 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,"))
@exclusive 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", )