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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315"""The stream pushes the live facts a watching client used to poll for."""
import http.clientimport jsonimport queueimport sqlite3import threadingimport timeimport unittestimport urllib.errorimport urllib.parseimport uuid
from harness import ( ROOT, Albedo, Provider, exclusive, latest_user, python, text,)from inspect_support import Inspection
# Long enough for the daemon's sidebar pace plus a loaded test host.PUSH_SECONDS = 25
class Stream: """One session subscription, read in the background so a test can wait."""
def __init__(self, app, session): self.connection = http.client.HTTPConnection( "127.0.0.1", app.connection["port"] ) self.connection.request( "GET", f"/sessions/{session}?tail=0", headers={ "Authorization": "Bearer " + app.connection["token"], "Accept": "text/event-stream", }, ) self.response = self.connection.getresponse() self.batches = queue.Queue() self.reader = threading.Thread(target=self._read, daemon=True) self.reader.start()
def _read(self): try: for line in self.response: if line.startswith(b"data: "): self.batches.put(json.loads(line[6:])) except (OSError, ValueError, AttributeError): # The stream was closed under the reader at cleanup; http.client # reports that as a missing buffer rather than a socket error. return
def batch(self, seconds=PUSH_SECONDS): try: return self.batches.get(timeout=seconds) except queue.Empty: raise AssertionError(f"no stream batch within {seconds}s") from None
def pushed(self, matches, seconds=PUSH_SECONDS): deadline = time.monotonic() + seconds while time.monotonic() < deadline: for event in self.batch(deadline - time.monotonic())["events"]: if matches(event): return event raise AssertionError(f"no pushed event within {seconds}s")
def close(self): self.connection.close()
class SessionStatusPushTests(unittest.TestCase): def setUp(self): def script(request): if request["messages"][-1].get("role") == "tool": return text("job started") if "start job" in latest_user(request): # A job that outlives the cell: the kernel reports it live, and # its end is the change a watching client must hear about. return python("handle = run('sleep', '4')\nprint(handle.id)") return text("done")
self.provider = Provider(script) self.addCleanup(self.provider.close)
def subscribe(self, app, session): stream = Stream(app, session) self.addCleanup(stream.close) return stream
def activate_work_item(self, app, title): """An item the work sidebar lists: it shows only what is active.""" workspace = urllib.parse.quote(str(app.workspace), safe="") with app.api( f"/extensions/work/items?workspace={workspace}", {"title": title} ) as created: resource = json.load(created)["resource"] self.assertEqual(resource["value"]["title"], title) with app.api( f"/extensions/work/items/{resource['value']['id']}?workspace={workspace}", {"status": "active"}, method="PATCH", headers={"If-Match": resource["etag"]}, ) as changed: self.assertEqual( json.load(changed)["resource"]["value"]["status"], "active" )
@exclusive def test_quiet_session_hibernates_and_handles_the_next_turn(self): def prepare(app): app.daemon.env["ALBEDO_INSPECT"] = "1"
with Albedo(self.provider, prepare=prepare) as app: session = app.session() app.prompt(session, "first turn").close() app.idle(session) probe = Inspection(app) expression = ( "io:put_chars(Call(erlang, apply, [fun() -> " "{some, {subject, Pid, _}} = 'albedo@daemon@session':live(" + "<<" + json.dumps(session) + ">>" + "), {current_function, {_, F, _}} = process_info(Pid, current_function), " "json:encode(atom_to_binary(F)) end, []]))" ) deadline = time.monotonic() + 10 while True: current = json.loads(probe.evaluate(expression)) # hibernate/0 keeps the caller's frame, so the actor waits in quiet. if current == "quiet": break if time.monotonic() >= deadline: self.fail(f"quiet session did not hibernate: {current}") time.sleep(0.05) app.prompt(session, "second turn").close() app.idle(session) self.assertEqual(len(self.provider.requests), 2) self.assertIn( "second turn", latest_user(self.provider.requests[-1]["request"]) )
@exclusive def test_kernel_notifications_do_not_retain_session_state(self): def prepare(app): app.daemon.env["ALBEDO_INSPECT"] = "1"
with Albedo(self.provider, prepare=prepare) as app: session = app.session() # A job's cell opens a kernel, which the daemon then watches. app.prompt(session, "start job").close() app.idle(session) sizes = json.loads( Inspection(app).evaluate( "io:put_chars(Call(erlang, apply, [fun() -> " "json:encode([erts_debug:flat_size(F) || {_, F} <- " "ets:tab2list(albedo_kernel_watch)]) end, []]))" ) ) self.assertTrue(sizes) self.assertLess(max(sizes), 256)
def test_stream_pushes_job_lifecycle_and_sidebar_changes(self): with Albedo(self.provider) as app: session = app.session() stream = self.subscribe(app, session) # The initial batch is still a complete snapshot: a fresh client # never needs a read to fill its footer. first = stream.batch() self.assertEqual(first["snapshot"]["id"], session) self.assertIn("kernel", first["snapshot"]) self.assertIn("glances", first["snapshot"])
app.prompt(session, "start job").close() started = stream.pushed( lambda event: ( event["type"] == "kernel" and (event["data"]["kernel"]["live_job_count"] or 0) >= 1 ) ) self.assertEqual(started["data"]["kernel"]["state"], "attached") self.assertTrue(started["data"]["kernel"]["running_jobs"])
app.idle(session) ended = stream.pushed( lambda event: ( event["type"] == "kernel" and event["data"]["kernel"]["live_job_count"] == 0 ) ) self.assertEqual(ended["data"]["kernel"]["running_jobs"], [])
self.activate_work_item(app, "pushed glance")
def shows_work(event): return event["type"] == "glances" and any( row["text"] == "pushed glance" for glance in event["data"]["glances"] for row in glance["rows"] )
event = stream.pushed(shows_work) glance = next( glance for glance in event["data"]["glances"] if glance["extension"] == "work" ) self.assertEqual(glance["title"], "active work")
# exclusive: holds the daemon's one runtime actor still@exclusiveclass RuntimeStallTests(unittest.TestCase): def test_status_reads_wait_out_a_busy_runtime(self): # A runtime whose mailbox is stalled can take seconds to answer. # A session without a kernel asks it what is loaded; # asking from the session's own actor failed every read of that session # ("session capture unavailable") once the runtime took 5 s. def prepare(app): app.daemon.env["ALBEDO_INSPECT"] = "1"
with Albedo(prepare=prepare) as app: probe = Inspection(app, ROOT / "test/e2e/albedo_runtime_stall_probe.erl") session = app.session() reads = {}
def read(view): try: with app.api(f"/sessions/{session}{view}") as response: reads[view] = (response.status, json.load(response)) except urllib.error.HTTPError as error: reads[view] = (error.code, json.load(error))
self.assertEqual(probe.call_json("suspend_json"), {"suspended": True}) try: readers = [ threading.Thread(target=read, args=(view,)) for view in ("", "?view=status") ] for reader in readers: reader.start() time.sleep(7) finally: self.assertEqual(probe.call_json("resume_json"), {"resumed": True}) for reader in readers: reader.join(30) self.assertEqual(reads[""][0], 200, reads[""]) self.assertEqual(reads["?view=status"][0], 200, reads["?view=status"]) self.assertEqual(reads["?view=status"][1]["kernel"]["state"], "none")
if __name__ == "__main__": unittest.main()
# exclusive: fills the store with thousands of sessions@exclusiveclass CollectionStreamTests(unittest.TestCase): def test_the_sessions_stream_starts_however_many_sessions_it_lists(self): # Following the collection reads every session it covers. That once ran # where mist gives a chunked response one second to start, and a big # store (or a busy host) answered every watcher with an empty stream. with Albedo() as app: seed = app.session()
def fill(app): with sqlite3.connect(app.home / "albedo.sqlite") as db: columns = [ row[1] for row in db.execute("PRAGMA table_info(sessions)") ] row = db.execute( f"SELECT {','.join(columns)} FROM sessions WHERE id=?", (seed,) ).fetchone() at = columns.index("id") db.executemany( f"INSERT INTO sessions({','.join(columns)}) " f"VALUES({','.join('?' * len(columns))})", ( [*row[:at], str(uuid.uuid4()), *row[at + 1 :]] for _ in range(8000) ), )
app.restart(prepare=fill) connection = http.client.HTTPConnection( "127.0.0.1", app.connection["port"], timeout=60 ) self.addCleanup(connection.close) connection.request( "GET", "/sessions?scope=all", headers={ "Authorization": "Bearer " + app.connection["token"], "Accept": "text/event-stream", }, ) first = next( ( line for line in connection.getresponse() if line.startswith(b"data: ") ), None, ) if first is None: self.fail("the sessions stream ended without a batch") events = [event["type"] for event in json.loads(first[6:])["events"]] self.assertEqual(events[:2], ["ready", "reset"])