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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533"""Identified inputs retain their admission and deliver once through failure and restart."""
from concurrent.futures import ThreadPoolExecutorfrom contextlib import closingimport base64import jsonimport sqlite3import threadingimport timeimport unittestimport urllib.error
from harness import Albedo, Provider, exclusive, operation_id, python, textfrom image_limits_test import png
class InputScenario(unittest.TestCase): def setUp(self): self.provider = Provider(lambda _: text("done")) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses") self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def put_input(self, session, input_id, intent): with self.app.api( f"/sessions/{session}/inputs/{input_id}", intent, method="PUT" ) as response: return json.load(response)
def receipt(self, session, input_id): with self.app.api(f"/sessions/{session}/inputs/{input_id}") as response: return json.load(response)
def committed(self, session, input_id): deadline = time.monotonic() + 30 while time.monotonic() < deadline: receipt = self.receipt(session, input_id) if receipt["delivery"] == "committed": return receipt time.sleep(0.03) self.fail(f"input remained pending: {receipt}")
def duplicates(self, session, input_id, intent): with ThreadPoolExecutor(max_workers=6) as executor: results = list( executor.map( lambda _: self.put_input(session, input_id, intent), range(6) ) ) self.assertTrue( all( result["id"] == input_id and result["admission"] == "accepted" for result in results ) ) self.assertEqual(len({result["acceptance_order"] for result in results}), 1) return results[0]
def conflict(self, session, input_id, intent): with self.assertRaises(urllib.error.HTTPError) as failure: self.put_input(session, input_id, intent) self.assertEqual(failure.exception.code, 409) self.assertEqual(json.load(failure.exception)["code"], "input_conflict")
def users(self, session): return [ entry for entry in self.app.history(session)["items"] if entry["kind"] == "user" ]
def skill_intent(self, session, arguments): with self.app.api(f"/sessions/{session}/catalog?kind=skills") as response: discovery = json.load(response)["discovery"] candidate = next( row for row in discovery["candidates"] if row["title"] == "retry-skill" ) return { "kind": "skill", "candidate_id": candidate["id"], "catalog_revision": discovery["revision"], "arguments": arguments, }
def write_skill(self): skill = self.app.workspace / ".agents/skills/retry-skill/SKILL.md" skill.parent.mkdir(parents=True) skill.write_text( "---\nname: retry-skill\ndescription: test activation\n---\nOriginal saved instructions.\n" ) return skill
def assert_user_order(self, session, inputs): users = self.users(session) self.assertEqual( [entry["input_id"] for entry in users], [input_id for input_id, _ in inputs] ) self.assertEqual( [ part["text"] for entry in users for part in entry["content"] if part["kind"] == "text" ], [intent["text"] for _, intent in inputs], )
class OperationsTests(InputScenario): def test_targeted_cancel_preserves_shared_turn_and_cannot_interrupt_a_later_turn( self, ): first_entered, second_entered, later_entered = [ threading.Event() for _ in range(3) ] first_release, second_release, later_release = [ threading.Event() for _ in range(3) ] for gate in (first_release, second_release, later_release): self.addCleanup(gate.set) calls = 0
def reply(_request): nonlocal calls calls += 1 entered, release = [ (first_entered, first_release), (second_entered, second_release), (later_entered, later_release), ][calls - 1] entered.set() if not release.wait(20): raise AssertionError( "targeted cancellation scenario did not release provider" ) return python("1") if calls == 1 else text("completed")
self.provider.script = reply session = self.app.session() identities = [operation_id() for _ in range(3)] self.put_input( session, identities[0], {"kind": "message", "text": "start shared turn"} ) self.assertTrue(first_entered.wait(10)) for index, identity in enumerate(identities[1:]): self.put_input( session, identity, {"kind": "message", "text": f"join shared turn {index}"}, ) first_release.set() self.assertTrue(second_entered.wait(15)) receipts = [self.committed(session, identity) for identity in identities] turn_ids = {receipt["turn"]["id"] for receipt in receipts} self.assertEqual(len(turn_ids), 1) with self.app.api( f"/sessions/{session}/inputs/{identities[1]}/cancel", {} ) as response: cancellation = json.load(response) self.assertEqual(cancellation["result"], "shared_running") with self.app.api(f"/sessions/{session}?tail=0") as response: self.assertFalse(json.load(response)["status"]["interrupt_requested"]) second_release.set() self.app.idle(session) self.assertTrue( all( self.receipt(session, identity)["turn"]["state"] == "completed" for identity in identities ) ) later = operation_id() self.put_input( session, later, {"kind": "message", "text": "a later independent turn"} ) self.assertTrue(later_entered.wait(10)) with self.app.api( f"/sessions/{session}/inputs/{identities[0]}/cancel", {} ) as response: self.assertEqual(json.load(response)["result"], "not_pending") with self.app.api(f"/sessions/{session}?tail=0") as response: snapshot = json.load(response) self.assertFalse(snapshot["status"]["interrupt_requested"]) self.assertNotIn(snapshot["status"]["run_id"], turn_ids) later_release.set() self.app.idle(session) self.assertEqual(self.receipt(session, later)["turn"]["state"], "completed")
def test_concurrent_creation_has_one_resource_and_immutable_provenance(self): session_id = operation_id() resource = f"/sessions/{session_id}" intent = { "kind": "new", "workspace": str(self.app.workspace), "provider_profile": self.app.profile, }
def create(_): try: with self.app.api( resource, intent, method="PUT", headers={"If-None-Match": "*"} ) as response: return response.status, json.load(response) except urllib.error.HTTPError as failure: self.assertEqual(failure.code, 412) return failure.code, None
with ThreadPoolExecutor(max_workers=6) as executor: results = list(executor.map(create, range(6))) self.assertEqual([status for status, _ in results].count(201), 1) with self.app.api(resource) as response: created = json.load(response) self.assertEqual( created["creation"]["submitted"]["workspace"], intent["workspace"] ) self.assertEqual( created["creation"]["resolved"], { key: created[key] for key in ("workspace", "provider_profile", "model", "effort") }, ) configuration = created["configuration_resource"] with self.app.api( configuration["url"], {"name": "edited after creation"}, method="PATCH", headers={"If-Match": configuration["etag"]}, ) as response: response.read() with self.app.api(resource) as response: edited = json.load(response) self.assertEqual(edited["creation"], created["creation"]) self.assertEqual(edited["name"], "edited after creation") with self.app.api("/sessions?scope=all") as response: matches = [ row for row in json.load(response)["items"] if row["id"] == session_id ] self.assertEqual(len(matches), 1)
def test_submission_identity_and_durable_echo(self): session = self.app.session() first = operation_id() intent = {"kind": "message", "text": "same message", "client_id": "first"} self.duplicates(session, first, intent) duplicate = self.put_input(session, first, {**intent, "client_id": "retry"}) self.assertEqual(duplicate["client_id"], "first") self.committed(session, first) self.app.idle(session) self.conflict(session, first, {**intent, "text": "changed message"}) self.conflict(session, first, {"kind": "continue"}) second = operation_id() self.put_input(session, second, {**intent, "client_id": "second"}) self.committed(session, second) self.app.idle(session) self.assert_user_order(session, [(first, intent), (second, intent)])
def test_image_duplicates_and_changed_image_conflict(self): session = self.app.session() input_id = operation_id() intent = { "kind": "message", "text": "inspect image", "images": [ { "mime_type": "image/png", "data": base64.b64encode(png(2, 2)).decode(), } ], } self.duplicates(session, input_id, intent) self.committed(session, input_id) self.app.idle(session) self.conflict( session, input_id, { **intent, "images": [ { **intent["images"][0], "data": base64.b64encode(png(3, 2)).decode(), } ], }, ) users = self.users(session) self.assertEqual([entry["input_id"] for entry in users], [input_id]) image = next( part["image"] for part in users[0]["content"] if part["kind"] == "image" ) self.assertEqual(image["width"], 2)
def test_continue_is_one_admission_without_an_extra_user_row(self): session = self.app.session() self.app.prompt(session, "start").close() self.app.idle(session) input_id = operation_id() self.duplicates(session, input_id, {"kind": "continue"}) self.committed(session, input_id) self.app.idle(session) self.assertEqual(len(self.users(session)), 1) self.assertEqual(len(self.provider.requests), 2) self.assertIsNotNone(self.receipt(session, input_id)["transcript_position"])
def test_the_turn_after_a_continue_repeats_what_the_continue_sent(self): # The continue prompt is history the model read: dropping it from the # next request changed that request's prefix, and the provider read # the whole conversation again uncached. session = self.app.session() self.app.prompt(session, "start").close() self.app.idle(session) input_id = operation_id() self.put_input(session, input_id, {"kind": "continue"}) self.committed(session, input_id) self.app.idle(session) self.app.prompt(session, "after").close() self.app.idle(session) continued, after = ( record["request"]["input"] for record in self.provider.requests[-2:] ) self.assertIn("continue your unfinished task", json.dumps(continued[-1])) self.assertEqual(after[: len(continued)], continued) self.assertEqual( [entry["content"][0]["text"] for entry in self.users(session)], ["start", "after"], )
def test_skill_admission_uses_resolved_content_once(self): skill = self.write_skill() session = self.app.session() input_id = operation_id() intent = self.skill_intent(session, "do this") accepted = self.duplicates(session, input_id, intent) self.committed(session, input_id) self.app.idle(session) skill.unlink() duplicate = self.put_input(session, input_id, intent) self.assertEqual(duplicate["acceptance_order"], accepted["acceptance_order"]) self.conflict(session, input_id, {**intent, "arguments": "other"}) self.assertIn( "Original saved instructions.", json.dumps(self.provider.requests) ) self.assertEqual( [entry["input_id"] for entry in self.users(session)], [input_id] )
@exclusiveclass WaitingOperationsTests(InputScenario): def setUp(self): self.gate = threading.Event() self.started = threading.Event() self.first = True
def reply(_): if self.first: self.first = False self.started.set() self.gate.wait(40) return text("done")
self.provider = Provider(reply) self.addCleanup(self.provider.close) self.addCleanup(self.gate.set) self.app = Albedo(self.provider, protocol="responses") self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def test_full_queue_duplicate_bypasses_limit_and_interrupt_cancels(self): session = self.app.session() self.app.prompt(session, "hold active generation").close() self.assertTrue(self.started.wait(10)) inputs = [ (operation_id(), {"kind": "message", "text": f"waiting {index}"}) for index in range(32) ] results = [ self.put_input(session, input_id, intent) for input_id, intent in inputs ] self.assertTrue(all(result["delivery"] == "pending" for result in results)) self.assertEqual( self.put_input(session, *inputs[0])["acceptance_order"], results[0]["acceptance_order"], ) with self.assertRaises(urllib.error.HTTPError) as full: self.put_input( session, operation_id(), {"kind": "message", "text": "queue overflow"} ) self.assertEqual(full.exception.code, 429) with self.app.api(f"/sessions/{session}?tail=0") as response: snapshot = json.load(response) self.app.api( f"/sessions/{session}/interrupt", { "run_id": snapshot["status"]["run_id"], "through_input_order": snapshot["input_order"], }, ).close() self.gate.set() self.app.idle(session) self.app.restart(crash=True) for (input_id, intent), accepted in zip(inputs, results): self.assertEqual(self.receipt(session, input_id)["delivery"], "cancelled") self.assertEqual( self.put_input(session, input_id, intent)["acceptance_order"], accepted["acceptance_order"], )
def test_crash_recovers_order_images_and_resolved_skill_content(self): skill = self.write_skill() session = self.app.session() with self.app.prompt(session, "hold active generation") as response: active_input = json.load(response)["id"] self.assertTrue(self.started.wait(10)) inputs = [ (operation_id(), {"kind": "message", "text": "first waiting"}), ( operation_id(), { "kind": "message", "text": "second image", "images": [ { "mime_type": "image/png", "data": base64.b64encode(png(2, 2)).decode(), } ], }, ), (operation_id(), self.skill_intent(session, "saved arguments")), ] accepted = [ self.put_input(session, input_id, intent) for input_id, intent in inputs ] self.assertTrue( all( self.receipt(session, input_id)["delivery"] == "pending" for input_id, _ in inputs ) ) skill.unlink() self.app.restart(crash=True) self.assertEqual( self.receipt(session, active_input)["turn"]["state"], "abandoned" ) self.gate.set() for input_id, _ in inputs: self.committed(session, input_id) self.app.idle(session) self.app.restart(crash=True) for (input_id, intent), result in zip(inputs, accepted): self.assertEqual( self.put_input(session, input_id, intent)["acceptance_order"], result["acceptance_order"], ) users = [ entry for entry in self.users(session) if entry["input_id"] in {input_id for input_id, _ in inputs} ] self.assertEqual( [entry["input_id"] for entry in users], [input_id for input_id, _ in inputs] ) image = next( part["image"] for part in users[1]["content"] if part["kind"] == "image" ) self.assertEqual(image["width"], 2) self.assertIn( "Original saved instructions.", json.dumps(self.provider.requests) )
@exclusiveclass BlockedOperationsTests(InputScenario): def test_full_blocked_queue_survives_restart_and_replays_rejection(self): session = self.app.session() inputs = [ (operation_id(), {"kind": "message", "text": f"blocked {index}"}) for index in range(32) ] with ( closing(sqlite3.connect(self.app.home / "albedo.sqlite")) as database, database, ): database.execute( "CREATE TRIGGER hold_inputs BEFORE INSERT ON transcript WHEN NEW.row_class='user' BEGIN SELECT RAISE(ABORT,'fixture'); END" ) accepted = [ self.put_input(session, input_id, intent) for input_id, intent in inputs ] deadline = time.monotonic() + 10 while time.monotonic() < deadline: if self.receipt(session, inputs[0][0])["blocking_reason"]: break time.sleep(0.03) else: self.fail("waiting inputs were never blocked") with self.app.api(f"/sessions/{session}?tail=0") as response: snapshot = json.load(response) self.assertEqual( snapshot["status"]["blocking_reason"]["detail"], self.receipt(session, inputs[0][0])["blocking_reason"]["detail"], ) rejected_id = operation_id() rejected = {"kind": "message", "text": "blocked overflow"} with self.assertRaises(urllib.error.HTTPError) as failure: self.put_input(session, rejected_id, rejected) self.assertEqual(failure.exception.code, 429) rejection = json.load(failure.exception) self.assertEqual(self.receipt(session, rejected_id)["admission"], "rejected") self.app.restart(crash=True) self.assertEqual( self.put_input(session, *inputs[-1])["acceptance_order"], accepted[-1]["acceptance_order"], ) with self.assertRaises(urllib.error.HTTPError) as restored_full: self.put_input(session, operation_id(), {"kind": "continue"}) self.assertEqual(restored_full.exception.code, 429) with ( closing(sqlite3.connect(self.app.home / "albedo.sqlite")) as database, database, ): database.execute("DROP TRIGGER hold_inputs") for input_id, _ in inputs: self.committed(session, input_id) self.app.idle(session) self.assert_user_order(session, inputs) with self.assertRaises(urllib.error.HTTPError) as replay: self.put_input(session, rejected_id, rejected) self.assertEqual(replay.exception.code, 429) self.assertEqual(json.load(replay.exception), rejection)