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.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553"""Agent families: spawn, forwarding, addressing, snapshots, and routing."""
import http.clientimport jsonimport socketimport sqlite3import tempfileimport threadingimport timeimport unittestimport urllib.errorimport urllib.requestimport urllib.parsefrom pathlib import Path
from harness import ( Albedo, Provider, Reply, exclusive, latest_user, prompt_text, operation_id, python, text,)from stream_support import StreamProbe
def wait_for(predicate, timeout=40): deadline = time.monotonic() + timeout while time.monotonic() < deadline: value = predicate() if value: return value time.sleep(0.05) raise AssertionError("agent event never arrived")
REMINDER = "You have not answered your parent yet."
class AgentsTests(unittest.TestCase): def setUp(self): def answer(request): # A child reminded to answer its parent repeats what it found. users = [ item["content"] for item in request.get("messages", []) if item.get("role") == "user" and isinstance(item.get("content"), str) ] user = next( (item for item in reversed(users) if REMINDER not in item), latest_user(request), ) line = user.splitlines()[-2 if "</mail>" in user else -1] return text("answered: " + line)
self.provider = Provider(answer) self.alpha = f"alpha-{self.provider.route}" self.beta = f"beta-{self.provider.route}" providers = { profile: { "extension": "openai", "baseUrl": self.provider.url + f"/{name}/v1", "apiKey": "key", "model": f"fixture-{name}-{self.provider.route}", "protocol": "chat_completions", } for profile, name in ((self.alpha, "alpha"), (self.beta, "beta")) } if ( self._testMethodName == "test_switch_late_projection_failure_keeps_the_old_model_and_mails_parent" ): providers[self.alpha]["protocol"] = "responses" if self._testMethodName == "test_a_stalled_endpoint_never_stalls_the_daemon": # Accepts connections and never answers, like a gateway whose host # is up and whose server is wedged. stalled = socket.create_server(("127.0.0.1", 0)) held = []
def hold(): while True: try: held.append(stalled.accept()[0]) except OSError: return
threading.Thread(target=hold, daemon=True).start() self.stalled_connections = held self.addCleanup(lambda: [c.close() for c in held]) self.addCleanup(stalled.close) providers[f"stalled-{self.provider.route}"] = { "extension": "openai", "baseUrl": f"http://127.0.0.1:{stalled.getsockname()[1]}/v1", "apiKey": "key", "model": f"fixture-stalled-{self.provider.route}", "protocol": "chat_completions", }
# The cached catalog lists a model for the fixture endpoint that no profile names. # models.json is daemon-wide, so only the exclusive catalog test writes it. def prepare(app): if ( self._testMethodName == "test_captured_subtree_deletion_blocks_new_child_admission" ): app.env["ALBEDO_INSPECT"] = "1" return (app.home / "models.json").write_text( json.dumps( { "fixture-gateway": { "api": "http://127.0.0.1/v1", "env": [], "models": {"fixture-listed": {"id": "fixture-listed"}}, } } ) )
catalog = ( self._testMethodName == "test_generic_profile_lists_its_endpoints_catalog_models" ) self.app = Albedo( self.provider, providers=providers, prepare=prepare if catalog or self._testMethodName == "test_captured_subtree_deletion_blocks_new_child_admission" else None, ) self.app.__enter__() self.addCleanup(self.provider.close) self.addCleanup(self.app.__exit__, None, None, None)
def api(self, path, body=None, **options): with self.app.api(path, body, **options) as response: return json.load(response)
def snapshot(self, session): return self.api(f"/sessions/{session}?tail=0")
def change(self, session, patch): with self.app.api(f"/sessions/{session}?view=configuration") as response: json.load(response) revision = response.headers["ETag"] return self.api( f"/sessions/{session}?view=configuration", patch, method="PATCH", headers={"If-Match": revision}, )["session"]
def move(self, session, workspace): snapshot = self.snapshot(session) return self.change( session, { "workspace": str(workspace), "family_revision": snapshot["family_revision"], }, )
def family(self, session): root = self.snapshot(session)["root_id"] return self.api(f"/sessions?scope=all&family_id={root}")["items"]
def asked(self, fragment): return [ content for record in self.provider.requests for item in record["request"].get( "messages", record["request"].get("input", []) ) if item.get("role") == "user" for content in [item.get("content", "")] if isinstance(content, str) and fragment in content ]
def spawn(self, parent, name="coder", task="map every wake path", model=None): session = operation_id() body = { "kind": "child", "parent_id": parent, "address": name, "name": name, "initial_input_id": operation_id(), "task": task, } if model is not None: body["model"] = model return self.api( f"/sessions/{session}", body, method="PUT", headers={"If-None-Match": "*"} )
def send_mail(self, sender, target, body, *, rejected=False): marker = "dispatch fixture mail " + operation_id() answer = self.provider.script
def dispatch(request): if ( request["messages"][-1].get("role") == "user" and prompt_text(request) == marker ): return python( "import json\n" f"receipt = await mail.submit({target!r}, {body!r})\n" "print(json.dumps(receipt))" ) return answer(request)
self.provider.script = dispatch try: self.send(sender, marker) self.app.idle(sender) finally: self.provider.script = answer [result] = [ entry for entry in self.app.history(sender)["items"] if entry["kind"] == "tool_result" ][-1:] execution = next( json.loads(part["value"]) for part in result["content"] if part["kind"] == "json" and part["field"] == "result" ) self.assertEqual(execution["status"], "error" if rejected else "ok", execution) return execution if rejected else json.loads(execution["output"])
def execute(self, session, source): marker = "execute fixture python " + operation_id() answer = self.provider.script
def dispatch(request): messages = request.get("messages", request.get("input", [])) if messages[-1].get("role") == "user" and prompt_text(request) == marker: return python(source) return answer(request)
self.provider.script = dispatch try: self.send(session, marker) self.app.idle(session) finally: self.provider.script = answer result = [ entry for entry in self.app.history(session)["items"] if entry["kind"] == "tool_result" ][-1] execution = next( json.loads(part["value"]) for part in result["content"] if part["kind"] == "json" and part["field"] == "result" ) self.assertEqual(execution["status"], "ok", execution) return execution["output"]
def send(self, session, message): return self.api( f"/sessions/{session}/inputs/{operation_id()}", {"kind": "message", "text": message}, method="PUT", )
def listen(self): heard = [] ready, stopped = threading.Event(), threading.Event() address = urllib.parse.urlsplit(self.app.base) connection = http.client.HTTPConnection( address.hostname, address.port, timeout=30 )
def consume(): try: connection.request( "GET", "/sessions?scope=all", headers={ "Authorization": "Bearer " + self.app.connection["token"], "Accept": "text/event-stream", }, ) with connection.getresponse() as response: for raw in response: if raw.startswith(b"data: "): heard.extend(json.loads(raw[6:])["events"]) ready.set() except (OSError, ValueError): if not stopped.is_set(): raise
thread = threading.Thread(target=consume, daemon=True) thread.start() self.assertTrue(ready.wait(timeout=5), "collection stream did not send ready")
def close(): stopped.set() if connection.sock is not None: connection.sock.shutdown(socket.SHUT_RDWR) thread.join(timeout=2) connection.close() self.assertFalse(thread.is_alive(), "collection reader did not stop")
self.addCleanup(close) return heard
def switch(self, parent, child, model): output = self.execute( parent, "import json\n" f"target = await agents.get({child!r})\n" f"print(json.dumps(await target.switch({model!r})))", ) return json.loads(output)
def test_switch_idle_child_preserves_work_and_session_defaults(self): parent = self.app.session() old_model = f"fixture-alpha-{self.provider.route}" new_model = f"fixture-beta-{self.provider.route}" child = self.spawn(parent, model=f"{self.alpha}/{old_model}")["id"] self.app.idle(child) self.app.idle(parent) self.execute(child, "kept_value = 41") self.app.idle(parent) result = self.switch(parent, child, new_model) self.assertEqual( result, {"model": f"{self.beta}/{new_model}", "pending": False} ) self.assertEqual(self.snapshot(child)["provider_profile"], self.beta) self.assertEqual(self.execute(child, "print(kept_value + 1)").strip(), "42") routed = [r for r in self.provider.requests if r["model"] == new_model] self.assertTrue(routed) self.assertTrue(all("/beta/v1/" in r["path"] for r in routed)) self.app.session() # The fixture still creates on its original default. history = self.app.history(child)["items"] self.assertIn("model switched", json.dumps(history))
def test_switch_running_request_finishes_tools_before_the_switch_note(self): parent = self.app.session() old_model = f"fixture-alpha-{self.provider.route}" new_model = f"fixture-beta-{self.provider.route}" child = self.spawn(parent, model=f"{self.alpha}/{old_model}")["id"] self.app.idle(child) self.app.idle(parent) entered, release = threading.Event(), threading.Event() self.addCleanup(release.set) answer = self.provider.script marker = "blocked request before switch"
def dispatch(request): if ( prompt_text(request) == marker and request["messages"][-1]["role"] == "user" ): entered.set() if not release.wait(timeout=30): raise AssertionError("test never released the old request") return python("preserved_step = 42\nprint('step complete')") if ( request["model"] == new_model and "model switched" in latest_user(request) and request["messages"][-1]["role"] == "user" ): return python( "assert preserved_step == 42\nawait mail.submit('parent', 'same turn complete')" ) return answer(request)
self.provider.script = dispatch self.send(child, marker) self.assertTrue(entered.wait(timeout=10)) self.assertEqual( self.switch(parent, child, f"{self.beta}/{new_model}"), {"model": f"{self.beta}/{new_model}", "pending": True}, ) self.assertEqual(self.snapshot(child)["model"], old_model) release.set() self.app.idle(child) self.app.idle(parent) routed = [ r["request"] for r in self.provider.requests if r["model"] == new_model ] self.assertTrue(routed) messages = routed[0]["messages"] self.assertEqual([m["role"] for m in messages[-2:]], ["tool", "user"]) self.assertIn("step complete", messages[-2]["content"]) self.assertIn('origin="model switched"', messages[-1]["content"]) self.assertIn(f"{self.alpha}/{old_model}", messages[-1]["content"]) self.assertIn(f"{self.beta}/{new_model}", messages[-1]["content"]) with sqlite3.connect(self.app.home / "albedo.sqlite") as db: turns = db.execute( "SELECT DISTINCT turn_id FROM transcript WHERE session=? AND turn_id IS NOT NULL", (child,), ).fetchall() # The initial task was one turn, the gated request and its beta continuation another. self.assertEqual(len(turns), 2) self.assertIn("same turn complete", json.dumps(self.app.history(parent)))
def test_switch_does_not_wait_for_a_running_cell_and_replaces_pending_choice(self): parent = self.app.session() old_model = f"fixture-alpha-{self.provider.route}" new_model = f"fixture-beta-{self.provider.route}" child = self.spawn(parent, model=f"{self.alpha}/{old_model}")["id"] self.app.idle(child) self.app.idle(parent) entered = self.app.workspace / "cell-entered" release = self.app.workspace / "cell-release" self.addCleanup(release.touch) answer = self.provider.script marker = "blocked cell before switch"
def dispatch(request): if ( prompt_text(request) == marker and request["messages"][-1]["role"] == "user" ): return python( f"Path({str(entered)!r}).touch()\n" f"while not Path({str(release)!r}).exists():\n await asyncio.sleep(0.05)\n" "print('cell complete')" ) return answer(request)
self.provider.script = dispatch self.send(child, marker) wait_for(entered.exists) self.assertTrue( self.switch(parent, child, f"{self.beta}/{new_model}")["pending"] ) self.assertTrue(self.switch(parent, child, old_model)["pending"]) self.assertTrue(self.switch(parent, child, new_model)["pending"]) release.touch() self.app.idle(child) self.assertEqual(self.snapshot(child)["model"], new_model) notes = [ entry for entry in self.app.history(child)["items"] if entry["kind"] == "note" and "Your parent switched this session" in json.dumps(entry) ] self.assertEqual(len(notes), 1)
def test_switch_late_projection_failure_keeps_the_old_model_and_mails_parent(self): parent = self.app.session() old_model = f"fixture-alpha-{self.provider.route}" new_model = f"fixture-beta-{self.provider.route}" child = self.spawn(parent, model=f"{self.alpha}/{old_model}")["id"] self.app.idle(child) self.app.idle(parent) entered, release = threading.Event(), threading.Event() self.addCleanup(release.set) answer = self.provider.script marker = "receive an opaque response before the switch"
def dispatch(request): if request["model"] == old_model and prompt_text(request) == marker: entered.set() if not release.wait(timeout=30): raise AssertionError("test never released opaque response") return Reply( "text", events=[ {"type": "response.created", "response": {"id": "opaque"}}, { "type": "response.output_text.delta", "output_index": 1, "content_index": 0, "delta": "kept opaque work", }, { "type": "response.completed", "response": { "id": "opaque", "output": [ { "type": "web_search_call", "id": "opaque-search", "status": "completed", }, { "type": "message", "role": "assistant", "content": [ { "type": "output_text", "text": "kept opaque work", } ], }, ], }, }, ], ) return answer(request)
self.provider.script = dispatch self.send(child, marker) self.assertTrue(entered.wait(timeout=10)) self.assertTrue(self.switch(parent, child, new_model)["pending"]) release.set() self.app.idle(child) wait_for(lambda: self.asked("was not applied")) self.app.idle(parent) self.assertEqual(self.snapshot(child)["model"], old_model) self.assertNotIn("model switched", json.dumps(self.app.history(child))) self.assertIn("-provider", json.dumps(self.app.history(child))) output = self.execute( parent, f"""try: await (await agents.get({child!r})).switch({new_model!r})except AgentsError as error: print(str(error))else: raise AssertionError('opaque history should refuse a switch immediately')""", ) self.assertIn("not portable", output) self.assertFalse([r for r in self.provider.requests if r["model"] == new_model])
def test_switch_refuses_unknown_models_and_non_children_without_changes(self): parent = self.app.session() child = self.spawn(parent)["id"] unrelated = self.app.session() self.app.idle(child) self.app.idle(parent) before = self.snapshot(child) output = self.execute( parent, f"""for target, model in [({child!r}, 'no-such-switch-model'), ({child!r}, {self.beta!r} + '/no-such-switch-model'), ({unrelated!r}, {before["model"]!r})]: try: await (await agents.get(target)).switch(model) except AgentsError as error: print(str(error)) else: raise AssertionError('switch should have been refused')""", ) self.assertIn("not available", output) self.assertIn("your own children", output) after = self.snapshot(child) self.assertEqual( (after["provider_profile"], after["model"]), (before["provider_profile"], before["model"]), ) self.assertNotIn("model switched", json.dumps(self.app.history(child)))
def test_spawn_task_and_unreviewed_forward(self): parent = self.app.session() child = self.spawn(parent) self.assertEqual( (child["parent_id"], child["address"], child["depth"], child["closed"]), (parent, "coder", 1, False), ) self.assertEqual(child["name"], "coder") resource = self.api(f"/sessions/{child['id']}") self.assertEqual( resource["creation"]["resolved"], { key: resource[key] for key in ("workspace", "provider_profile", "model", "effort") }, ) self.assertNotIn( child["id"], [item["id"] for item in self.api("/sessions")["items"]] ) self.app.idle(child["id"]) task = self.asked("map every wake path")[0] self.assertIn('kind="task"', task) self.assertIn(f'session="{parent}"', task) self.assertIn("map every wake path", task) forwarded = wait_for(lambda: self.asked('kind="unreviewed"'))[0] self.assertIn('from="coder"', forwarded) self.assertIn("answered: map every wake path", forwarded) # Asked once to answer before it stopped; the forward is the fallback. self.assertEqual(len(self.asked(REMINDER)), 1) self.app.idle(parent) self.app.idle(child["id"]) self.assertEqual( [ item["address"] for item in self.api(f"/sessions?scope=all&parent_id={parent}")["items"] ], ["coder"], )
# exclusive: saves extensions.json @exclusive def test_saved_subagent_limits_hold_for_the_next_child(self): settings = self.app.home / "extensions.json" settings.write_text( json.dumps({"subagents": {"maxOpenChildren": 1, "maxDepth": 1}}) ) parent = self.app.session() first = self.spawn(parent, "first")["id"] for under, name in ((parent, "second"), (first, "deeper")): with self.subTest(name=name): with self.assertRaises(urllib.error.HTTPError): self.spawn(under, name)
# -1 takes both caps away for the next child, past the defaults of 12 # open children and 3 deep. settings.write_text( json.dumps({"subagents": {"maxOpenChildren": -1, "maxDepth": -1}}) ) self.spawn(parent, "second") lineage = [first] for depth in range(2, 5): child = self.spawn(lineage[-1], f"depth{depth}") self.assertEqual(child["depth"], depth) lineage.append(child["id"]) self.assertEqual(self.snapshot(lineage[-1])["root_id"], parent)
@exclusive def test_mail_attribution_and_agent_card_survive_reconnect_and_restart(self): parent = self.app.session() self.change(parent, {"name": "lead desk"}) child = self.spawn(parent, task="map wake paths")["id"] self.app.idle(child) self.app.idle(parent) task = next( entry for entry in self.app.history(child)["items"] if (entry.get("mail") or {}).get("kind") == "task" ) self.assertEqual(task["turn_type"], "agent") self.assertEqual(task["mail"]["sender_label"], "lead desk") self.assertEqual(task["mail"]["sender_session_id"], parent) self.assertEqual(task["input_id"], task["mail"]["mail_id"]) self.assertEqual(task["content"][0]["text"], "map wake paths") # Retrying creation must not admit the initial task twice. with self.assertRaises(urllib.error.HTTPError) as repeated: self.api( f"/sessions/{child}", { "kind": "child", "parent_id": parent, "address": "coder", "name": "coder", "initial_input_id": task["input_id"], "task": "map wake paths", }, method="PUT", headers={"If-None-Match": "*"}, ) self.assertEqual(repeated.exception.code, 412) self.assertEqual( sum( entry["id"] == task["id"] for entry in self.app.history(child)["items"] ), 1, ) self.assertEqual( self.snapshot(child)["activity"]["current_request"]["text"], "map wake paths", ) before = self.app.stream_page(child) answer = self.provider.script marker = "report bounded progress"
def progress(request): if ( request["messages"][-1].get("role") == "user" and prompt_text(request) == marker ): return python('await agents.progress("tracing delivery")') return answer(request)
self.provider.script = progress self.send(child, marker) self.app.idle(child) self.app.idle(parent) self.provider.script = answer self.assertEqual( self.snapshot(child)["activity"]["latest_progress"], "tracing delivery" ) # A new subscriber sees progress in its authoritative reset snapshot. reset = self.app.stream_page(child) self.assertEqual( reset["snapshot"]["activity"]["latest_progress"], "tracing delivery" ) receipt = self.send_mail(parent, child, "also cover replay") self.app.idle(child) self.app.idle(parent) activity = self.snapshot(child)["activity"] self.assertEqual( activity["current_request"], {"input_id": receipt["id"], "text": "also cover replay"}, ) self.assertIsNone(activity["latest_progress"]) history = self.app.history(child)["items"] followup = next( entry for entry in history if entry["input_id"] == receipt["id"] ) live = self.app.stream_page(child, before) self.assertEqual( [ event["data"]["entry"] for event in live["events"] if event["type"] == "message" and event["data"]["entry"]["id"] == followup["id"] ], [followup], ) # Peer messages remain visible but cannot replace the parent's request. peer = self.app.session() self.send_mail(peer, child, "peer observation") self.app.idle(child) self.app.idle(parent) self.assertEqual( self.snapshot(child)["activity"]["current_request"], activity["current_request"], ) forwarded = next( entry for entry in self.app.history(parent)["items"] if (entry.get("mail") or {}).get("kind") == "unreviewed" ) self.assertEqual(forwarded["mail"]["sender_label"], "coder") self.app.restart() recovered = self.snapshot(child)["activity"] self.assertEqual(recovered["current_request"], activity["current_request"]) self.assertIsNone(recovered["latest_progress"]) replayed = self.app.stream_page(child)["snapshot"]["history"]["items"] self.assertEqual( next(entry for entry in replayed if entry["id"] == followup["id"]), followup )
def test_spawning_or_deleting_a_child_keeps_the_parent_validator(self): """A family's membership is not its configuration: a client holding the configuration it displayed may still change the model after agents come and go.""" parent = self.app.session() with self.app.api(f"/sessions/{parent}?view=configuration") as response: observed = response.headers["ETag"] response.read() before = self.app.family_revision(parent) child = self.spawn(parent)["id"] self.app.idle(child) self.app.idle(parent) with self.app.api(f"/sessions/{parent}?view=configuration") as response: self.assertEqual(response.headers["ETag"], observed) response.read() self.assertNotEqual(self.app.family_revision(parent), before) with self.app.api( f"/sessions/{parent}?view=configuration", {"model": "changed-model"}, method="PATCH", headers={"If-Match": observed}, ) as response: self.assertEqual(json.load(response)["session"]["model"], "changed-model") edited = response.headers["ETag"] with self.app.api(f"/sessions/{child}?view=configuration") as response: child_etag = response.headers["ETag"] self.api( f"/sessions/{child}?view=configuration", method="DELETE", headers={"If-Match": child_etag}, ) with self.app.api(f"/sessions/{parent}?view=configuration") as response: self.assertEqual(response.headers["ETag"], edited) self.assertEqual(json.load(response)["model"], "changed-model")
def test_tree_snapshot_and_event_stream(self): parent = self.app.session() heard = self.listen() child = self.spawn(parent)["id"] wait_for(lambda: self.asked('kind="unreviewed"')) for asker in (parent, child): nodes = {item["id"]: item for item in self.family(asker)} self.assertEqual(set(nodes), {parent, child}) self.assertEqual( (nodes[parent]["parent_id"], nodes[parent]["depth"]), (None, 0) ) self.assertEqual( ( nodes[child]["parent_id"], nodes[child]["depth"], nodes[child]["address"], ), (parent, 1, "coder"), ) wait_for( lambda: any( event["type"] == "invalidate" and child in event["data"]["session_ids"] for event in list(heard) ) ) wait_for( lambda: any( event["type"] == "activity" and event["data"]["session_id"] == child for event in list(heard) ) ) wait_for( lambda: any( event["type"] == "mail" and event["data"]["receiver_session_id"] == parent for event in list(heard) ) )
def test_rename_keeps_family_address_and_reports_stream_event(self): parent = self.app.session() child = self.spawn(parent)["id"] self.app.idle(child) wait_for(lambda: self.asked('kind="unreviewed"')) self.app.idle(parent) heard = self.listen() self.assertEqual( self.change(child, {"name": " wake path\naudit "})["name"], "wake path audit", ) self.assertEqual( self.change(parent, {"name": "orchestration"})["name"], "orchestration" ) nodes = {item["id"]: item for item in self.family(parent)} self.assertEqual( (nodes[parent]["name"], nodes[parent]["address"]), ("orchestration", None) ) self.assertEqual( (nodes[child]["name"], nodes[child]["address"]), ("wake path audit", "coder"), ) wait_for( lambda: any( event["type"] == "invalidate" and child in event["data"]["session_ids"] for event in list(heard) ) ) self.assertEqual(self.snapshot(child)["name"], "wake path audit") with self.assertRaises(urllib.error.HTTPError) as caught: self.send("no-such-session", "cannot deliver") self.assertEqual(caught.exception.code, 404) receipt = self.send_mail(parent, "coder", "also cover schedules") self.assertEqual((receipt["to"], receipt["name"]), (child, "coder")) wait_for(lambda: self.asked("also cover schedules")) wait_for( lambda: any( "also cover schedules" in message for message in self.asked('kind="unreviewed"') ) ) self.app.idle(parent)
def test_reply_and_external_id_mail(self): parent, radio = self.app.session(), self.app.session() child = self.spawn(parent)["id"] wait_for(lambda: self.asked('kind="unreviewed"')) self.app.idle(parent) self.app.idle(child) receipt = self.send_mail(child, "parent", "schedules covered") self.assertEqual(receipt["to"], parent) wait_for(lambda: self.asked("schedules covered")) self.app.idle(parent) failed = self.send_mail(child, "radio", "by unrelated name", rejected=True) self.assertIn("radio", failed["output"]) self.assertFalse(self.asked("by unrelated name")) receipt = self.send_mail(child, radio, "deploy when green") self.assertEqual(receipt["to"], radio) wait_for(lambda: self.asked("deploy when green")) self.app.idle(radio) for recipient, body in ( (parent, "schedules covered"), (radio, "deploy when green"), ): delivered = [ entry for entry in self.app.history(recipient)["items"] if entry["kind"] == "user" and any( part["kind"] == "text" and body in part["text"] for part in entry["content"] ) ] self.assertEqual(len(delivered), 1)
def test_tree_deletion_and_renamed_workspace(self): parent = self.app.session() child = self.spawn(parent)["id"] self.app.idle(child) wait_for(lambda: self.asked('kind="unreviewed"')) self.app.idle(parent) with self.app.api(f"/sessions/{parent}?view=configuration") as response: response.read() revision = response.headers["ETag"] family = self.app.family_revision(parent) with self.assertRaises(urllib.error.HTTPError) as caught: self.api( f"/sessions/{parent}?view=configuration", method="DELETE", headers={"If-Match": revision}, ) self.assertEqual(caught.exception.code, 409) self.assertIn("child", json.load(caught.exception)["detail"]) grandchild = self.spawn(child, "deep")["id"] self.app.idle(grandchild) wait_for(lambda: len(self.asked('kind="unreviewed"')) >= 2) self.app.idle(child) self.app.idle(parent) with self.assertRaises(urllib.error.HTTPError) as caught: self.api( f"/sessions/{parent}?view=configuration&scope=subtree&family_revision={family}", method="DELETE", headers={"If-Match": revision}, ) self.assertEqual(caught.exception.code, 412) entered, release = threading.Event(), threading.Event() self.addCleanup(release.set) answer = self.provider.script
def held(request): if "hold subtree deletion" in latest_user(request): entered.set() release.wait(30) return python( "from pathlib import Path\nPath('deleted-run-executed').write_text('wrong')" ) return answer(request)
self.provider.script = held receipt = self.send(child, "hold subtree deletion") self.assertTrue(entered.wait(30)) with self.app.api(f"/sessions/{parent}?view=configuration") as response: response.read() revision = response.headers["ETag"] family = self.app.family_revision(parent) outcome = self.api( f"/sessions/{parent}?view=configuration&scope=subtree&family_revision={family}", method="DELETE", headers={"If-Match": revision}, ) self.assertEqual(outcome["state"], "complete") self.assertEqual(outcome["deleted_ids"], [grandchild, child, parent]) terminal = self.api(f"/sessions/{child}/inputs/{receipt['id']}") self.assertEqual(terminal["delivery"], "committed") self.assertEqual(terminal["turn"]["state"], "interrupted") self.assertIsNotNone(terminal["turn"]["ended_at"]) release.set() self.assertFalse((self.app.workspace / "deleted-run-executed").exists()) listed = [item["id"] for item in self.api("/sessions?scope=all")["items"]] self.assertNotIn(parent, listed) self.assertNotIn(child, listed) radio = self.app.session() self.change(radio, {"name": "radio desk"}) moved = self.app.root / "moved" moved.mkdir() self.assertEqual(self.move(radio, moved)["name"], "radio desk") self.assertEqual(self.snapshot(radio)["name"], "radio desk")
@exclusive def test_captured_subtree_deletion_blocks_new_child_admission(self): parent = self.app.session() child = self.spawn(parent, "captured")["id"] self.app.idle(child) wait_for(lambda: self.asked('kind="unreviewed"')) self.app.idle(parent) with self.app.api(f"/sessions/{parent}?view=configuration") as response: response.read() revision = response.headers["ETag"] family = self.app.family_revision(parent) probe = StreamProbe(self.app) self.assertEqual( probe.call("suspend_session_json", f"[<<{json.dumps(child)}>>]"), {"suspended": True}, ) outcomes, failures = [], []
def delete(): try: outcomes.append( self.api( f"/sessions/{parent}?view=configuration&scope=subtree&family_revision={family}", method="DELETE", headers={"If-Match": revision}, ) ) except Exception as error: failures.append(error)
worker = threading.Thread(target=delete) worker.start() try:
def captured(): with sqlite3.connect(self.app.home / "albedo.sqlite") as db: rows = db.execute( "SELECT id,deletion_id FROM sessions WHERE id IN (?,?)", (parent, child), ).fetchall() return len(rows) == 2 and all(row[1] for row in rows)
wait_for(captured, timeout=15) self.assertTrue(worker.is_alive()) with self.assertRaises(urllib.error.HTTPError) as caught: self.spawn(parent, "too-late", "must not execute") self.assertEqual(caught.exception.code, 409) with sqlite3.connect(self.app.home / "albedo.sqlite") as db: admitted = { row[0] for row in db.execute( "SELECT id FROM sessions WHERE id=? UNION ALL " "SELECT session FROM session_family WHERE parent=?", (parent, parent), ) } self.assertEqual(admitted, {parent, child}) finally: self.assertEqual( probe.call("resume_session_json", f"[<<{json.dumps(child)}>>]"), {"resumed": True}, ) worker.join(35) self.assertFalse( worker.is_alive(), "deletion did not finish after owner resumed" ) self.assertEqual(failures, []) self.assertEqual(len(outcomes), 1) self.assertEqual(outcomes[0]["state"], "complete") self.assertEqual(outcomes[0]["deleted_ids"], [child, parent]) self.assertFalse(self.asked("must not execute"))
@exclusive def test_subtree_deletion_reports_the_rows_a_failed_commit_leaves(self): parent = self.app.session() failed = self.spawn(parent, "failed")["id"] removed = self.spawn(parent, "removed")["id"] for session in (failed, removed): self.app.idle(session) wait_for(lambda: self.asked('from="failed"') and self.asked('from="removed"')) self.app.idle(parent) with self.app.api(f"/sessions/{parent}?view=configuration") as response: response.read() revision = response.headers["ETag"] family = self.app.family_revision(parent) with sqlite3.connect(self.app.home / "albedo.sqlite") as db: db.execute( "CREATE TRIGGER fail_observed_deletion BEFORE DELETE ON sessions " f"WHEN OLD.id='{failed}' BEGIN SELECT RAISE(ABORT,'injected deletion failure'); END" ) self.addCleanup(self._remove_deletion_failure) outcome = self.api( f"/sessions/{parent}?view=configuration&scope=subtree&family_revision={family}", method="DELETE", headers={"If-Match": revision}, ) self.assertEqual(outcome["state"], "partial") self.assertEqual(outcome["deleted_count"], 1) self.assertEqual(outcome["remaining_count"], 2) self.assertEqual(outcome["deleted_ids"], [removed]) remaining = {row["id"]: row["reason"] for row in outcome["remaining"]} self.assertEqual(set(remaining), {parent, failed}) self.assertTrue(all(remaining.values())) self.assertIn("captured child", remaining[parent]["detail"]) self._remove_deletion_failure() # Reading authoritative state never resumes a partially failed deletion. for _ in range(2): self.assertEqual( {row["id"] for row in self.family(parent)}, {parent, failed} ) with self.assertRaises(urllib.error.HTTPError) as caught: self.api( f"/sessions/{parent}?view=configuration&scope=subtree&family_revision={family}", method="DELETE", headers={"If-Match": revision}, ) self.assertEqual(caught.exception.code, 412) self.assertEqual({row["id"] for row in self.family(parent)}, {parent, failed})
def _remove_deletion_failure(self): with sqlite3.connect(self.app.home / "albedo.sqlite") as db: db.execute("DROP TRIGGER IF EXISTS fail_observed_deletion")
def test_children_follow_a_workspace_move(self): release = threading.Event() self.addCleanup(release.set) answer = self.provider.script
def held(request): user = latest_user(request) if "hold this turn" in user or "probe moved child" in user: if request["messages"][-1].get("role") == "tool": return text("workspace probe finished") if "hold this turn" in user: release.wait(30) marker = ( "old-child-cwd" if "hold this turn" in user else "new-child-cwd" ) return python( "import os\nfrom pathlib import Path\n" f"Path({marker!r}).write_text(os.getcwd())" ) return answer(request)
self.provider.script = held parent = self.app.session() still = self.spawn(parent, "still")["id"] away = self.spawn(parent, "away")["id"]
def heard(): return { entry["mail"]["sender_session_id"] for entry in self.app.history(parent)["items"] if entry.get("mail") }
wait_for(lambda: {still, away} <= heard()) for session in (still, away, parent): self.app.idle(session) elsewhere = Path(tempfile.mkdtemp(prefix="elsewhere-", dir=self.app.root)) self.move(away, elsewhere) busy = self.spawn(parent, "busy", "hold this turn")["id"] wait_for(lambda: self.asked("hold this turn")) moved = Path(tempfile.mkdtemp(prefix="moved-", dir=self.app.root)) self.assertEqual(self.move(parent, moved)["workspace"], str(moved))
def workspaces(): return {item["id"]: item["workspace"] for item in self.family(parent)}
self.assertEqual( workspaces(), { parent: str(moved), still: str(moved), away: str(elsewhere), busy: str(self.app.workspace), }, ) latest = Path(tempfile.mkdtemp(prefix="latest-", dir=self.app.root)) self.move(parent, latest) before_move_applied = self.app.stream_page(busy) release.set() self.app.idle(busy) self.assertEqual(workspaces()[busy], str(latest)) self.assertEqual( (self.app.workspace / "old-child-cwd").read_text(), str(self.app.workspace) ) self.assertFalse((moved / "old-child-cwd").exists()) self.send(busy, "probe moved child") self.app.idle(busy) self.assertEqual((latest / "new-child-cwd").read_text(), str(latest)) self.assertFalse((moved / "new-child-cwd").exists()) events = self.app.stream_page(busy, before_move_applied)["events"] notes = [ event["data"]["text"] for event in events if event["type"] == "note" and event["data"]["origin"] == "workspace" ] self.assertTrue( any( "python variables were cleared" in note and "transcript is intact" in note for note in notes ), events, )
@exclusive def test_failed_deferred_workspace_application_survives_restart(self): entered, release = threading.Event(), threading.Event() self.addCleanup(release.set) answer = self.provider.script
def held(request): user = latest_user(request) if "deferred workspace hold" in user: entered.set() release.wait(30) return text("old turn completed") if "restored workspace probe" in user: if request["messages"][-1].get("role") == "tool": return text("restored workspace completed") return python( "import os\nfrom pathlib import Path\n" "Path('restored-child-cwd').write_text(os.getcwd())" ) return answer(request)
self.provider.script = held parent = self.app.session() child = self.spawn(parent, "deferred", "deferred workspace hold")["id"] self.assertTrue(entered.wait(30), "child never reached provider gate") destination = Path(tempfile.mkdtemp(prefix="deferred-", dir=self.app.root)) self.move(parent, destination) destination.rmdir() release.set() self.app.idle(child)
def assert_pending(): snapshot = self.snapshot(child) self.assertEqual(snapshot["workspace"], str(self.app.workspace)) change = snapshot["workspace_change"] self.assertEqual(change["active"], str(self.app.workspace)) self.assertEqual(change["desired"], str(destination)) return change["revision"]
revision = assert_pending() self.app.restart(crash=True) self.assertEqual(assert_pending(), revision) destination.mkdir() self.send(child, "restored workspace probe") self.app.idle(child) self.assertEqual( (destination / "restored-child-cwd").read_text(), str(destination) ) snapshot = self.snapshot(child) self.assertEqual(snapshot["workspace"], str(destination)) self.assertIsNone(snapshot["workspace_change"]) self.assertFalse((self.app.workspace / "restored-child-cwd").exists())
def test_cancel_all_stops_every_running_turn_beneath_the_caller(self): entered = {"mid": threading.Event(), "leaf": threading.Event()} release = threading.Event() self.addCleanup(release.set) answer = self.provider.script
def held(request): user = latest_user(request) for name, gate in entered.items(): if f"hold {name}" in user: gate.set() release.wait(30) return text("held turn ended") if request["messages"][-1].get("role") == "user" and user == "stop all": return python("print(sorted(await agents.cancel_all()))") return answer(request)
self.provider.script = held parent = self.app.session() mid = self.spawn(parent, "mid", "hold mid")["id"] leaf = self.spawn(mid, "leaf", "hold leaf")["id"] for gate in entered.values(): self.assertTrue(gate.wait(30), "a held turn never reached the provider") self.send(parent, "stop all") for session in (parent, mid, leaf): self.app.idle(session) self.assertFalse(release.is_set()) result = next( json.loads(part["value"]) for entry in reversed(self.app.history(parent)["items"]) if entry["kind"] == "tool_result" for part in entry["content"] if part["kind"] == "json" and part["field"] == "result" ) self.assertEqual(result["status"], "ok", result) self.assertEqual(result["output"].strip(), str(sorted([mid, leaf])))
def test_qualified_and_inferred_cross_provider_models(self): parent = self.app.session() self.change(parent, {"name": "radio desk"}) with self.assertRaises(urllib.error.HTTPError) as caught: self.spawn( parent, "missing", "should not run", f"{self.beta}/no-such-model" ) self.assertEqual(caught.exception.code, 409) for name, model in ( ("explicit", f"{self.beta}/fixture-beta-{self.provider.route}"), ("inferred", f"fixture-beta-{self.provider.route}"), ): made = self.spawn(parent, name, "cross-provider task", model) self.assertEqual(made["provider_profile"], self.beta) self.assertEqual(made["model"], f"fixture-beta-{self.provider.route}") self.assertEqual(made["parent_id"], parent) self.app.idle(made["id"]) # Each child's requests repeat its task, so count distinct letters. tasks = wait_for( lambda: ( set(self.asked('kind="task">\ncross-provider')) if len(set(self.asked('kind="task">\ncross-provider'))) == 2 else None ) ) self.assertTrue(all('from="radio desk"' in task for task in tasks)) self.assertGreaterEqual( sum("/beta/" in record["path"] for record in self.provider.requests), 2 )
# exclusive: writes the daemon-wide models.json catalog @exclusive def test_generic_profile_lists_its_endpoints_catalog_models(self): parent = self.app.session() made = self.spawn( parent, "listed", "catalog task", f"{self.beta}/fixture-listed" ) self.assertEqual( (made["provider_profile"], made["model"]), (self.beta, "fixture-listed") ) self.app.idle(made["id"])
def test_a_stalled_endpoint_never_stalls_the_daemon(self): # Spawning a child lists provider models, and a generic profile asks its # endpoint. That ran inside the registry, which every request waits on: # an endpoint that never answered held it for seconds per spawn, and # every request in the daemon failed with 503 meanwhile. parent = self.app.session() latencies, stop = [], threading.Event()
def probe(): while not stop.is_set(): began = time.monotonic() try: self.app.api("/server").close() status = 200 except urllib.error.HTTPError as error: status = error.code latencies.append((time.monotonic() - began, status)) time.sleep(0.05)
prober = threading.Thread(target=probe) prober.start() try: # A bare model the parent's provider does not list: every profile # is asked, the stalled one included. made = self.spawn( parent, "scout", "scan", f"fixture-beta-{self.provider.route}" ) finally: stop.set() prober.join() self.assertEqual(made["provider_profile"], self.beta) self.assertEqual([s for _, s in latencies if s != 200], []) self.assertLess(max(t for t, _ in latencies), 2, latencies) # The endpoint's failure is remembered: the next spawn does not ask it # again. Counted, not timed: a spawn on a loaded machine takes seconds. asked = len(self.stalled_connections) self.assertGreater(asked, 0) self.spawn(parent, "scout-2", "scan", f"fixture-beta-{self.provider.route}") self.assertEqual(len(self.stalled_connections), asked) self.app.idle(made["id"])
def test_request_to_missing_session_is_rejected(self): session = self.app.session() marker = "missing request " + operation_id() answer = self.provider.script
def dispatch(request): if ( request["messages"][-1].get("role") == "user" and prompt_text(request) == marker ): return python( "try:\n" " await mail.request('does-not-exist', 'body')\n" "except Exception as error:\n" " print(type(error).__name__)\n" ) return answer(request)
self.provider.script = dispatch try: self.send(session, marker) self.app.idle(session) finally: self.provider.script = answer result = [ entry for entry in self.app.history(session)["items"] if entry["kind"] == "tool_result" ][-1] execution = next( json.loads(part["value"]) for part in result["content"] if part["kind"] == "json" and part["field"] == "result" ) self.assertEqual(execution["status"], "ok", execution) self.assertIn("WorkError", execution["output"])
def test_start_peer_and_role_are_visible_in_sessions(self): parent = self.app.session() marker = "start peer " + operation_id() answer = self.provider.script
def dispatch(request): if ( request["messages"][-1].get("role") == "user" and prompt_text(request) == marker ): return python( "peer = await agents.start(" + repr("peer task") + ", name=" + repr("peer") + ", model=" + repr(f"{self.alpha}/fixture-alpha-{self.provider.route}") + ", role=" + repr("research") + ")\n" ) return answer(request)
self.provider.script = dispatch try: self.send(parent, marker) self.app.idle(parent) finally: self.provider.script = answer peers = [ item for item in self.api("/sessions?scope=all")["items"] if item["name"] == "peer" ] self.assertEqual(len(peers), 1) peer = peers[0] self.assertEqual(peer["started_by"], parent) self.assertEqual(peer["role"], "research") self.app.idle(peer["id"]) task = next( entry for entry in self.app.history(peer["id"])["items"] if (entry.get("mail") or {}).get("kind") == "request" ) self.assertEqual(task["mail"]["sender_session_id"], parent) self.assertEqual(task["content"][0]["text"], "peer task") self.execute( parent, f"await agents.set_role('verification', id={peer['id']!r})", ) self.assertEqual(self.snapshot(peer["id"])["role"], "verification")
def test_started_peer_unreviewed_forward_and_open_requests(self): starter = self.app.session() task = "peer obligation task " + operation_id() answer = self.provider.script peer_open_sent = threading.Event() reviewed_sent = threading.Event()
def dispatch(request): latest = latest_user(request) if peer_open_sent.is_set() and task in latest: return text("peer last words") if reviewed_sent.is_set() and "review request" in latest: return text("reviewed reply sent") if "You owe a reply to" in latest: return text("peer final words") if task in latest: peer_open_sent.set() return python( "await asyncio.sleep(1)\n" "opened = await mail.open()\n" "assert len(opened['owed']) == 1, opened\n" f"assert opened['owed'][0]['session'] == {starter!r}, opened\n" "print('peer-open-ok')" ) if "review request" in latest: reviewed_sent.set() return python( f"await mail.submit({starter!r}, 'reviewed response')\n" "print('reviewed-ok')" ) return answer(request)
self.provider.script = dispatch try: output = self.execute( starter, "peer = await agents.start(" + repr(task) + ", name='worker', model=" + repr(f"{self.alpha}/fixture-alpha-{self.provider.route}") + ")\n" # The request counts as awaiting once the peer's session has # taken it, which a loaded machine can take seconds to do. "for _ in range(1500):\n" " opened = await mail.open()\n" " if opened['awaiting']:\n" " break\n" " await asyncio.sleep(0.01)\n" "assert len(opened['awaiting']) == 1, opened\n" "assert opened['awaiting'][0]['session'] == peer.id, opened\n" "print(peer.id)", ) peer = output.strip() self.app.idle(peer) forwarded = wait_for(lambda: self.asked('kind="unreviewed"') or None)[0] self.assertIn("peer final words", forwarded) self.assertEqual(len(self.asked("You owe a reply to")), 1)
self.execute(starter, f"await mail.request({peer!r}, 'review request')") self.app.idle(peer) self.app.idle(starter) for session in (starter, peer): self.execute( session, "opened = await mail.open()\n" "assert opened == {'owed': [], 'awaiting': []}, opened\n" "print('open-clear')", ) finally: self.provider.script = answer
def test_only_starter_can_cancel_and_close_peer(self): starter = self.app.session() peer = self.execute( starter, "peer = await agents.start('authorization task', name='owned-peer', model=" + repr(f"{self.alpha}/fixture-alpha-{self.provider.route}") + ")\n" "print(peer.id)", ).strip() self.app.idle(peer) stranger = self.app.session() output = self.execute( stranger, "target = await agents.get(" + repr(peer) + ")\n" "errors = []\n" "for action in (target.cancel, target.close):\n" " try:\n" " await action()\n" " except Exception as error:\n" " errors.append(str(error))\n" "assert len(errors) == 2, errors\n" "print('refused')", ) self.assertIn("refused", output) self.execute( starter, "target = await agents.get(" + repr(peer) + ")\n" "await target.cancel()\n" "await target.close()", ) self.assertTrue(self.snapshot(peer)["closed"])
def test_root_request_reminds_and_forwards_unreviewed(self): sender = self.app.session() recipient = self.app.session() body = "plain root request " + operation_id() answer = self.provider.script
def dispatch(request): if "You owe a reply to" in latest_user(request): return text("root final words: " + body) return answer(request)
self.provider.script = dispatch try: self.execute(sender, f"await mail.request({recipient!r}, {body!r})") self.app.idle(recipient) finally: self.provider.script = answer forwarded = wait_for(lambda: self.asked('kind="unreviewed"') or None)[0] self.assertIn("root final words: " + body, forwarded) self.assertEqual(len(self.asked("You owe a reply to")), 1)