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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310"""Admission and consumption failures leave recoverable, atomic identified decisions."""
import base64from contextlib import closing, contextmanagerimport jsonimport randomimport structimport zlibimport sqlite3import urllib.error
from harness import exclusive, operation_idfrom image_limits_test import pngfrom operations_test import InputScenario
class OperationLifecycleTests(InputScenario): @contextmanager def database(self): with ( closing(sqlite3.connect(self.app.home / "albedo.sqlite", timeout=10)) as db, db, ): yield db
def create(self, session_id, intent): with self.app.api( f"/sessions/{session_id}", intent, method="PUT", headers={"If-None-Match": "*"}, ) as response: return json.load(response)
@exclusive def test_creation_failure_rolls_back_session_and_receipt(self): session_id = operation_id() intent = { "kind": "new", "workspace": str(self.app.workspace), "provider_profile": self.app.profile, } with self.database() as database: before = database.execute("SELECT count(*) FROM sessions").fetchone()[0] database.execute( "CREATE TRIGGER fail_creation BEFORE INSERT ON sessions BEGIN SELECT RAISE(ABORT,'fixture'); END" ) with self.assertRaises(urllib.error.HTTPError): self.create(session_id, intent) with self.database() as database: self.assertEqual( database.execute("SELECT count(*) FROM sessions").fetchone()[0], before ) self.assertEqual( database.execute( "SELECT count(*) FROM operations WHERE id=?", (session_id,) ).fetchone()[0], 0, ) database.execute("DROP TRIGGER fail_creation") created = self.create(session_id, intent) self.assertEqual(created["id"], session_id) with self.app.api(f"/sessions/{session_id}") as response: self.assertEqual(json.load(response)["creation"], created["creation"])
@exclusive def test_admission_failure_can_retry_original_id(self): session = self.app.session() input_id = operation_id() intent = {"kind": "message", "text": "durable input"} with self.database() as database: database.execute( "CREATE TRIGGER fail_admission BEFORE INSERT ON pending_inputs BEGIN SELECT RAISE(ABORT,'fixture'); END" ) with self.assertRaises(urllib.error.HTTPError): self.put_input(session, input_id, intent) with self.database() as database: self.assertEqual( database.execute( "SELECT count(*) FROM operations WHERE id=?", (input_id,) ).fetchone()[0], 0, ) self.assertEqual( database.execute( "SELECT count(*) FROM transcript WHERE session=?", (session,) ).fetchone()[0], 0, ) database.execute("DROP TRIGGER fail_admission") accepted = self.put_input(session, input_id, intent) self.committed(session, input_id) self.app.idle(session) self.assertEqual( self.put_input(session, input_id, intent)["acceptance_order"], accepted["acceptance_order"], ) self.assertEqual( [entry["input_id"] for entry in self.users(session)], [input_id] )
@exclusive def test_consumption_failure_keeps_pending_input_until_storage_recovers(self): session = self.app.session() input_id = operation_id() intent = { "kind": "message", "text": "recover after storage repair", "images": [ { "mime_type": "image/png", "data": base64.b64encode(png(2, 2)).decode(), } ], } with self.database() as database: database.execute( "CREATE TRIGGER fail_consumption BEFORE INSERT ON submission_events BEGIN SELECT RAISE(ABORT,'fixture'); END" ) accepted = self.put_input(session, input_id, intent) self.assertEqual(accepted["admission"], "accepted") self.assertEqual(self.receipt(session, input_id)["delivery"], "pending") with self.database() as database: self.assertEqual( database.execute( "SELECT count(*) FROM pending_inputs WHERE operation_id=?", (input_id,), ).fetchone()[0], 1, ) self.assertEqual( database.execute( "SELECT count(*) FROM transcript WHERE session=?", (session,) ).fetchone()[0], 0, ) self.assertEqual( database.execute("SELECT count(*) FROM images").fetchone()[0], 0 ) self.assertEqual( database.execute("SELECT count(*) FROM submission_events").fetchone()[ 0 ], 0, ) self.app.restart(crash=True) self.assertEqual( self.put_input(session, input_id, intent)["acceptance_order"], accepted["acceptance_order"], ) self.assertEqual(self.receipt(session, input_id)["delivery"], "pending") with self.database() as database: database.execute("DROP TRIGGER fail_consumption") self.committed(session, input_id) self.app.idle(session) self.app.restart(crash=True) self.assertEqual(self.receipt(session, input_id)["delivery"], "committed") self.conflict(session, input_id, {**intent, "text": "changed after restart"}) self.assertEqual( [entry["input_id"] for entry in self.users(session)], [input_id] )
@exclusive def test_repeated_image_commits_grow_only_compact_metadata(self): session = self.app.session() width, height = 256, 256
def chunk(kind, data): body = kind + data return ( struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body)) )
pixels = random.Random(1).randbytes(width * height * 3) rows = b"".join( b"\x00" + pixels[start : start + width * 3] for start in range(0, len(pixels), width * 3) ) image = ( b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)) + chunk(b"IDAT", zlib.compress(rows)) + chunk(b"IEND", b"") ) attachment = { "mime_type": "image/png", "data": base64.b64encode(image).decode(), }
def durable_bytes(): with self.database() as database: return sum( database.execute( f"SELECT coalesce(sum(length({column})),0) FROM {table}" ).fetchone()[0] for table, column in ( ("images", "data"), ("transcript", "payload"), ("submission_events", "payload"), ("continuation_markers", "payload"), ("pending_inputs", "payload"), ("operations", "response"), ) )
inputs, sizes = [], [] for index in range(3): input_id = operation_id() inputs.append(input_id) self.put_input( session, input_id, { "kind": "message", "text": "same image", "client_id": f"image-echo-{index}", "images": [attachment], }, ) self.committed(session, input_id) self.app.idle(session) sizes.append(durable_bytes()) self.assertLess(sizes[2] - sizes[0], len(image)) self.app.restart(crash=True) users = self.users(session) self.assertEqual([entry["input_id"] for entry in users], inputs) for index, input_id in enumerate(inputs): self.assertEqual( self.receipt(session, input_id)["client_id"], f"image-echo-{index}" ) metadata = [ part["image"] for entry in users for part in entry["content"] if part["kind"] == "image" ] self.assertEqual( [ (item["width"], item["height"], item["original_bytes"]) for item in metadata ], [(width, height, len(image))] * 3, )
@exclusive def test_rejected_creation_replays_after_workspace_is_repaired(self): session_id = operation_id() workspace = self.app.workspace / "missing" intent = { "kind": "new", "workspace": str(workspace), "provider_profile": self.app.profile, } with self.assertRaises(urllib.error.HTTPError) as original: self.create(session_id, intent) problem = json.load(original.exception) with self.assertRaises(urllib.error.HTTPError) as lookup: self.app.api(f"/sessions/{session_id}") self.assertEqual( json.load(lookup.exception)["decision"]["admission"], "rejected" ) workspace.mkdir() self.app.restart(crash=True) with self.assertRaises(urllib.error.HTTPError) as repeated: self.create(session_id, intent) self.assertEqual(repeated.exception.code, original.exception.code) self.assertEqual(json.load(repeated.exception), problem) with self.database() as database: self.assertEqual( database.execute("SELECT count(*) FROM sessions").fetchone()[0], 0 )
@exclusive def test_failed_cancellation_does_not_discard_pending_input(self): session = self.app.session() input_id = operation_id() with self.database() as database: database.execute( "CREATE TRIGGER hold_consumption BEFORE INSERT ON transcript WHEN NEW.row_class='user' BEGIN SELECT RAISE(ABORT,'fixture'); END" ) self.put_input( session, input_id, {"kind": "message", "text": "keep until cancellation commits"}, ) with self.database() as database: database.execute( "CREATE TRIGGER fail_cancellation BEFORE UPDATE ON operations WHEN NEW.delivery='cancelled' BEGIN SELECT RAISE(ABORT,'fixture'); END" ) resource = f"/sessions/{session}/inputs/{input_id}/cancel" with self.assertRaises(urllib.error.HTTPError): self.app.api(resource, {}) self.assertEqual(self.receipt(session, input_id)["delivery"], "pending") with self.database() as database: database.execute("DROP TRIGGER fail_cancellation") with self.app.api(resource, {}) as response: self.assertEqual(json.load(response)["result"], "cancelled") self.app.restart(crash=True) self.assertEqual(self.receipt(session, input_id)["delivery"], "cancelled") with self.database() as database: self.assertEqual( database.execute( "SELECT count(*) FROM pending_inputs WHERE operation_id=?", (input_id,), ).fetchone()[0], 0, )