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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014"""Direct HTTP checks for ACP authorization, admission, restart, and MCP state.
Adapter-visible behavior belongs in acp_test.py."""
import base64import jsonfrom pathlib import Pathimport secretsimport sqlite3import sysimport threadingimport timeimport unittestimport urllib.error
from harness import ( Albedo, Provider, Reply, exclusive, latest_user, operation_id, python, text, wait_until,)
from inspect_support import Inspection
SERVER = Path(__file__).with_name("fake_mcp_server.py")SECRET = "editor-literal-secret-4f1c"
def fake_server(name="editor", secret=SECRET): return { "name": name, "command": sys.executable, "args": [str(SERVER)], "env": [ {"name": "FAKE_SECRET", "value": secret}, {"name": "FAKE_REPORT_CWD", "value": "1"}, ], }
def token(): return base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b"=").decode()
class AcpClient: """One ACP connection's view of the daemon: its token and its sessions."""
def __init__(self, app): self.app = app self.token = token()
def headers(self, extra=None): return {"Albedo-Attachment-Token": self.token, **(extra or {})}
def create(self, session_id=None, workspace=None, servers=()): session_id = session_id or operation_id() with self.app.api( f"/sessions/{session_id}", { "kind": "acp", "workspace": str(workspace or self.app.workspace), "mcp_servers": list(servers), }, method="PUT", headers=self.headers({"If-None-Match": "*"}), ) as response: created = json.load(response) return session_id, created
def attach(self, session_id, workspace=None, servers=()): with self.app.api( f"/sessions/{session_id}/acp-attachment", { "workspace": str(workspace or self.app.workspace), "mcp_servers": list(servers), }, method="PUT", headers=self.headers(), ) as response: return json.load(response)
def renew(self, session_id): with self.app.api( f"/sessions/{session_id}/acp-attachment/renew", {}, method="POST", headers=self.headers(), ) as response: return response.status
def release(self, session_id): with self.app.api( f"/sessions/{session_id}/acp-attachment", method="DELETE", headers=self.headers(), ) as response: return response.status
def prompt(self, session_id, blocks, input_id=None): input_id = input_id or operation_id() with self.app.api( f"/sessions/{session_id}/inputs/{input_id}", {"kind": "acp_prompt", "blocks": blocks}, method="PUT", headers=self.headers(), ) as response: return input_id, json.load(response)
def session_read(app, session_id): with app.api(f"/sessions/{session_id}?tail=0") as response: return json.load(response)
def stored_with(home, value): """Files under the daemon's home that hold `value`. Unreadable files (credential backups are 0000) cannot leak to another reader either.""" found = [] for path in home.rglob("*"): try: if path.is_file() and value.encode() in path.read_bytes(): found.append(str(path)) except PermissionError: continue return found
def refused(test, call, status, code): with test.assertRaises(urllib.error.HTTPError) as failure: call() test.assertEqual(failure.exception.code, status) test.assertEqual(json.load(failure.exception)["code"], code)
class AcpAttachmentTest(unittest.TestCase): def setUp(self): self.app = Albedo() self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def test_creation_records_marker_and_attaches_its_token(self): editor = AcpClient(self.app) session_id, created = editor.create() self.assertEqual(created["acp"]["attachment"], "attached") self.assertEqual( created["creation"]["submitted"], {"kind": "acp", "workspace": str(self.app.workspace), "mcp_servers": []}, ) listed = session_read(self.app, session_id) self.assertEqual(listed["acp"]["attachment"], "attached")
def test_creation_resend_recovers_and_changed_intent_conflicts(self): editor = AcpClient(self.app) session_id, first = editor.create() _, again = editor.create(session_id) self.assertEqual(again["id"], first["id"]) other = self.app.workspace / "other" other.mkdir() refused( self, lambda: editor.create(session_id, workspace=other), 409, "creation_conflict", )
def test_creation_requires_a_token(self): refused( self, lambda: self.app.api( f"/sessions/{operation_id()}", { "kind": "acp", "workspace": str(self.app.workspace), "mcp_servers": [], }, method="PUT", headers={"If-None-Match": "*"}, ), 400, "attachment_token_required", )
def test_competing_attachment_is_refused_until_released(self): first, second = AcpClient(self.app), AcpClient(self.app) session_id, _ = first.create() refused(self, lambda: second.attach(session_id), 409, "attachment_held") refused(self, lambda: second.renew(session_id), 409, "attachment_lost") first.attach(session_id) self.assertEqual(first.renew(session_id), 204) self.assertEqual(first.release(session_id), 204) self.assertEqual(first.release(session_id), 204) self.assertEqual( session_read(self.app, session_id)["acp"]["attachment"], "detached" ) second.attach(session_id) self.assertEqual( session_read(self.app, session_id)["acp"]["attachment"], "attached" )
def test_attach_checks_marker_and_exact_workspace(self): editor = AcpClient(self.app) plain = self.app.session() refused(self, lambda: editor.attach(plain), 409, "not_acp_session") session_id, _ = editor.create() other = self.app.workspace / "nested" other.mkdir() refused( self, lambda: editor.attach(session_id, workspace=other), 409, "workspace_mismatch", ) refused(self, lambda: editor.attach(operation_id()), 404, "session_not_found")
def test_other_clients_read_but_cannot_drive_an_acp_session(self): editor = AcpClient(self.app) session_id, _ = editor.create() refused( self, lambda: self.app.prompt(session_id, "from the terminal"), 409, "acp_owned", ) refused( self, lambda: self.app.api( f"/sessions/{session_id}/inputs/{operation_id()}", {"kind": "acp_prompt", "blocks": [{"type": "text", "text": "hi"}]}, method="PUT", ), 409, "acp_owned", ) stranger = AcpClient(self.app) refused( self, lambda: stranger.prompt(session_id, [{"type": "text", "text": "hi"}]), 409, "attachment_lost", ) refused( self, lambda: self.app.api( f"/sessions/{session_id}?view=configuration", {"model": "fixture-model"}, method="PATCH", ), 409, "acp_owned", ) refused( self, lambda: self.app.api(f"/sessions/{session_id}/reload", {}, method="POST"), 409, "acp_owned", ) with self.app.api(f"/sessions/{session_id}?view=configuration") as response: etag = response.headers["ETag"] refused( self, lambda: self.app.api( f"/sessions/{session_id}?view=configuration", { "model": "fixture-model", "selection": {"extensions": {"files": False}}, }, method="PATCH", headers=editor.headers({"If-Match": etag}), ), 409, "acp_owned", ) refused( self, lambda: self.app.api( f"/sessions/{session_id}?view=configuration", {"model": "fixture-model"}, method="PATCH", headers=stranger.headers({"If-Match": etag}), ), 409, "attachment_lost", ) with self.app.api( f"/sessions/{session_id}?view=configuration", {"name": "renamed by the operator"}, method="PATCH", headers={"If-Match": etag}, ): pass self.assertEqual( session_read(self.app, session_id)["name"], "renamed by the operator" )
def test_acp_prompt_is_refused_outside_acp_sessions(self): editor = AcpClient(self.app) plain = self.app.session() refused( self, lambda: editor.prompt(plain, [{"type": "text", "text": "hi"}]), 409, "not_acp_session", )
def test_resource_links_reject_invalid_uris_and_unknown_path(self): editor = AcpClient(self.app) session_id, _ = editor.create() bad = [ { "type": "resource_link", "uri": "file:///tmp/a", "name": "a", "path": "/tmp/b", }, { "type": "resource_link", "uri": "file://remote/tmp/a", "name": "a", }, { "type": "resource_link", "uri": "file:///tmp/%zz", "name": "a", }, { "type": "resource_link", "uri": "file:///tmp/a", "name": "a", "size": -1, }, {"type": "image", "data": "", "mime_type": "image/png"}, ] for block in bad: with self.subTest(block=block): refused( self, lambda: editor.prompt(session_id, [block]), 400, "invalid_request", )
def test_token_never_reaches_responses_or_disk(self): editor = AcpClient(self.app) session_id, created = editor.create() attached = editor.attach(session_id) self.assertNotIn(editor.token, json.dumps(created)) self.assertNotIn(editor.token, json.dumps(attached)) self.assertNotIn(editor.token, json.dumps(session_read(self.app, session_id))) self.assertEqual(stored_with(self.app.home, editor.token), [])
def fast_leases(app): app.env["ALBEDO_TEST_TIME_SCALE"] = "100"
@exclusiveclass AcpLeaseExpiryTest(unittest.TestCase): def test_expired_lease_drops_the_writer_only(self): with Albedo(prepare=fast_leases) as app: first, second = AcpClient(app), AcpClient(app) session_id, _ = first.create() lease = first.attach(session_id)["lease"] self.assertLessEqual(lease["lease_ms"], 1000) time.sleep(lease["lease_ms"] / 1000 + 0.3) refused(self, lambda: first.renew(session_id), 409, "attachment_lost") self.assertEqual( session_read(app, session_id)["acp"]["attachment"], "detached" ) second.attach(session_id)
class Gate: """A provider that answers each request only once the test opens it."""
def __init__(self, reply="done"): self.opened = threading.Event() self.requests = [] self.reply = reply self.provider = Provider(self.answer)
def answer(self, request): self.requests.append(request) self.opened.wait(30) return text(self.reply)
@exclusiveclass AcpTurnTest(unittest.TestCase): def setUp(self): self.gate = Gate() self.addCleanup(self.gate.provider.close) self.addCleanup(self.gate.opened.set) self.app = Albedo(self.gate.provider, protocol="responses") self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.editor = AcpClient(self.app) self.session_id, _ = self.editor.create()
def receipt(self, input_id): with self.app.api(f"/sessions/{self.session_id}/inputs/{input_id}") as response: return json.load(response)
def test_attach_captures_the_running_turn_and_later_prompts_wait(self): first, _ = self.editor.prompt( self.session_id, [{"type": "text", "text": "one"}] ) wait_until( lambda: self.gate.requests, 30, "the first turn never reached the model" ) second, _ = self.editor.prompt( self.session_id, [{"type": "text", "text": "two"}] ) attached = self.editor.attach(self.session_id) self.assertEqual(attached["active_turn"]["acp_input_id"], first) self.assertEqual(attached["pending_input_ids"], [second]) captured_history = self.app.history(self.session_id)["items"] self.assertEqual( attached["history_cutoff"], max((row["position"] for row in captured_history), default=0), ) self.gate.opened.set() self.app.idle(self.session_id) later = self.app.stream_page( self.session_id, { "generation": attached["cursor"]["generation"], "cursor": attached["cursor"]["sequence"], }, )["events"] later_positions = { event["data"]["entry"]["position"] for event in later if event["type"] == "message" } final_history = self.app.history(self.session_id)["items"] self.assertTrue(later_positions) self.assertTrue( all( any(row["position"] == position for row in final_history) for position in later_positions ) ) self.assertTrue( all(position > attached["history_cutoff"] for position in later_positions) ) self.assertNotEqual( self.receipt(first)["turn"]["id"], self.receipt(second)["turn"]["id"] )
def test_scoped_cancellation_requires_the_current_attachment(self): running, _ = self.editor.prompt( self.session_id, [{"type": "text", "text": "running"}] ) wait_until(lambda: self.gate.requests, 30, "turn never reached the provider") queued, _ = self.editor.prompt( self.session_id, [{"type": "text", "text": "queued"}] ) replacement = AcpClient(self.app)
def cancel(input_id, headers=None): with self.app.api( f"/sessions/{self.session_id}/inputs/{input_id}/cancel", {}, method="POST", headers=headers, ) as response: return json.load(response)["result"]
for headers in (None, replacement.headers()): refused(self, lambda: cancel(running, headers), 409, "attachment_lost") refused(self, lambda: cancel(queued, headers), 409, "attachment_lost") self.editor.release(self.session_id) replacement.attach(self.session_id) refused( self, lambda: cancel(running, self.editor.headers()), 409, "attachment_lost" ) self.assertEqual(cancel(queued, replacement.headers()), "cancelled") # Even a terminal receipt requires the current writer's token. refused( self, lambda: cancel(queued, self.editor.headers()), 409, "attachment_lost" ) self.assertEqual(cancel(queued, replacement.headers()), "not_pending") self.assertEqual(cancel(running, replacement.headers()), "interrupt_requested") self.app.idle(self.session_id) self.assertEqual(self.receipt(queued)["delivery"], "cancelled") self.assertEqual(len(self.gate.requests), 1)
def test_restart_cancels_waiting_prompts_and_does_not_continue_the_turn(self): first, _ = self.editor.prompt( self.session_id, [{"type": "text", "text": "one"}] ) wait_until( lambda: self.gate.requests, 30, "the first turn never reached the model" ) second, _ = self.editor.prompt( self.session_id, [{"type": "text", "text": "two"}] ) asked = len(self.gate.requests) self.app.restart(crash=True) self.gate.opened.set() self.assertEqual(self.receipt(second)["delivery"], "cancelled") self.assertEqual(self.receipt(first)["turn"]["state"], "abandoned") listed = session_read(self.app, self.session_id) self.assertEqual(listed["acp"]["attachment"], "detached") time.sleep(1) self.assertEqual(len(self.gate.requests), asked) refused( self, lambda: self.editor.prompt( self.session_id, [{"type": "text", "text": "three"}] ), 409, "attachment_lost", ) self.editor.attach(self.session_id) self.editor.prompt(self.session_id, [{"type": "text", "text": "three"}]) self.app.idle(self.session_id)
class AcpOverlayValidationTest(unittest.TestCase): def setUp(self): self.app = Albedo() self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def test_overlay_names_must_be_unique(self): editor = AcpClient(self.app) refused( self, lambda: editor.create(servers=[fake_server("twin"), fake_server("twin")]), 409, "mcp_overlay_conflict", )
def test_a_server_that_cannot_start_rejects_the_creation(self): editor = AcpClient(self.app) session_id = operation_id() broken = {**fake_server(), "command": "/nonexistent/editor-mcp"} refused( self, lambda: editor.create(session_id, servers=[broken]), 409, "mcp_overlay_failed", ) refused( self, lambda: session_read(self.app, session_id), 404, "session_not_found", )
def test_command_must_be_absolute(self): editor = AcpClient(self.app) relative = {**fake_server(), "command": "python3"} refused(self, lambda: editor.create(servers=[relative]), 400, "invalid_request")
class OverlayTool: """A model that calls the first editor MCP tool once, then answers."""
def __init__(self): self.opened = threading.Event() self.opened.set() self.results = [] self.provider = Provider(self.answer)
def answer(self, request): self.opened.wait(30) inputs = request["input"] last_user = max( (index for index, item in enumerate(inputs) if item.get("role") == "user"), default=-1, ) outputs = [ item for item in inputs[last_user + 1 :] if item.get("type") == "function_call_output" ] if outputs: self.results.append(outputs[-1]["output"]) return text("done") tools = sorted(tool["name"] for tool in request.get("tools", [])) tool = next((name for name in tools if name.startswith("mcp_")), None) if tool is None: return text("no tools") return Reply("python", tool_name=tool, tool_arguments={"message": "hi"})
@exclusiveclass AcpOverlayTest(unittest.TestCase): def setUp(self): self.model = OverlayTool() self.addCleanup(self.model.provider.close) self.addCleanup(self.model.opened.set) self.app = Albedo( self.model.provider, protocol="responses", prepare=lambda app: app.env.update(ALBEDO_INSPECT="1"), ) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.editor = AcpClient(self.app)
def test_a_running_turn_keeps_its_overlay_unless_resupplied_identically(self): session_id, _ = self.editor.create(servers=[fake_server()]) self.model.opened.clear() self.editor.prompt(session_id, [{"type": "text", "text": "wait"}]) wait_until( lambda: session_read(self.app, session_id)["status"]["phase"] != "idle", 30, "the turn never started", ) self.editor.attach(session_id, servers=[fake_server()]) refused( self, lambda: self.editor.attach( session_id, servers=[fake_server(secret="changed")] ), 409, "session_busy", ) self.model.opened.set() self.app.idle(session_id) self.editor.attach(session_id, servers=[])
def test_restart_loses_the_overlay_until_the_editor_resupplies_it(self): session_id, _ = self.editor.create(servers=[fake_server()]) self.app.restart() self.editor.attach(session_id, servers=[fake_server()]) self.editor.prompt(session_id, [{"type": "text", "text": "use the tool"}]) self.app.idle(session_id) self.assertIn(SECRET, self.model.results[-1])
def test_overlay_preparation_failure_keeps_the_previous_tools(self): session, _ = self.editor.create(servers=[fake_server()]) broken = {**fake_server(), "command": "/nonexistent/editor-mcp"} refused( self, lambda: self.editor.attach(session, servers=[broken]), 409, "mcp_overlay_failed", ) self.editor.prompt(session, [{"type": "text", "text": "use original tool"}]) self.app.idle(session) self.assertIn(SECRET, self.model.results[-1])
def test_published_overlay_survives_a_runtime_reload_failure(self): session, _ = self.editor.create(servers=[fake_server()]) self.editor.prompt(session, [{"type": "text", "text": "use original tool"}]) self.app.idle(session) self.assertIn(SECRET, self.model.results[-1]) with sqlite3.connect(self.app.home / "albedo.sqlite") as db: original = db.execute( "SELECT acp_overlay_identities FROM sessions WHERE id = ?", (session,) ).fetchone()[0] broken_prompt = self.app.workspace / "SYSTEM.md" broken_prompt.write_bytes(b"\xff") replacement = fake_server(secret="replacement-editor-secret") refused( self, lambda: self.editor.attach(session, servers=[replacement]), 503, "overlay_installed_reload_failed", ) refused( self, lambda: self.editor.attach(session, servers=[replacement]), 503, "overlay_installed_reload_failed", ) with sqlite3.connect(self.app.home / "albedo.sqlite") as db: published = db.execute( "SELECT acp_overlay_identities FROM sessions WHERE id = ?", (session,) ).fetchone()[0] self.assertNotEqual(published, original) requested = len(self.model.provider.requests) refused( self, lambda: self.editor.prompt( session, [{"type": "text", "text": "must not use stale tools"}] ), 409, "overlay_installed_reload_failed", ) # The runtime survives an abnormal session-actor exit. Its old cached # tools must remain fenced until an identical reattach repairs them. Inspection(self.app).evaluate( "Call(erlang, apply, [fun() -> " "{some, {subject, Pid, _}} = 'albedo@daemon@session':live(<<" + json.dumps(session) + ">>), Ref = monitor(process, Pid), exit(Pid, kill), " "receive {'DOWN', Ref, process, Pid, _} -> ok after 5000 -> error(actor_alive) end end, []])" ) refused( self, lambda: self.editor.prompt( session, [{"type": "text", "text": "still must not use stale tools"}] ), 409, "overlay_installed_reload_failed", ) self.assertEqual(len(self.model.provider.requests), requested) self.assertEqual(len(self.model.results), 1) broken_prompt.unlink() # Retry the published replacement: execution must use its tool credentials. self.editor.attach(session, servers=[replacement]) self.editor.prompt(session, [{"type": "text", "text": "use replacement tool"}]) self.app.idle(session) self.assertIn("replacement-editor-secret", self.model.results[-1]) self.assertNotIn(SECRET, self.model.results[-1])
def test_empty_overlay_reloads_after_the_attachment_owner_restarts(self): session, _ = self.editor.create(servers=[fake_server()]) self.editor.prompt(session, [{"type": "text", "text": "use original tool"}]) self.app.idle(session) self.assertIn(SECRET, self.model.results[-1]) broken_prompt = self.app.workspace / "SYSTEM.md" broken_prompt.write_bytes(b"\xff") refused( self, lambda: self.editor.attach(session, servers=[]), 503, "overlay_installed_reload_failed", ) Inspection(self.app).evaluate( "Call(erlang, apply, [fun() -> Pid = whereis(albedo_acp_attachments), " "Ref = monitor(process, Pid), exit(Pid, kill), " "receive {'DOWN', Ref, process, Pid, _} -> ok after 5000 -> error(actor_alive) end end, []])" ) # No required server identities remain, but the runtime still has the # original tools. Missing attachment state must force a reload anyway. refused( self, lambda: self.editor.attach(session, servers=[]), 503, "overlay_installed_reload_failed", ) broken_prompt.unlink() self.editor.attach(session, servers=[]) self.editor.prompt( session, [{"type": "text", "text": "no editor tools remain"}] ) self.app.idle(session) self.assertEqual(len(self.model.results), 1) self.assertFalse( any( tool["name"].startswith("mcp_") for tool in self.model.provider.requests[-1]["request"].get("tools", []) ) )
def send_fixture_mail(self, session): def script(request): inputs = request.get("input", []) if any(item.get("type") == "function_call_output" for item in inputs): return text("sent") if "send fixture mail" in latest_user(request): return python(f"await mail.submit({session!r}, 'editor detached')") return text("mail delivered")
self.model.provider.script = script sender = self.app.session() self.app.prompt(sender, "send fixture mail").close() self.app.idle(sender) return sender
def test_missing_overlay_blocks_mail_until_the_editor_reattaches(self): session, _ = self.editor.create(servers=[fake_server()]) self.app.restart() self.send_fixture_mail(session) wait_until( lambda: ( "overlay_reattach_required" in json.dumps(session_read(self.app, session)["activity"]) ), 15, "missing overlay to require editor reattachment", ) self.assertEqual(self.app.history(session)["items"], []) self.editor.attach(session, servers=[fake_server()]) wait_until( lambda: any( row["kind"] == "assistant" for row in self.app.history(session)["items"] ), 15, "pending mail to run after editor reattachment", )
def test_empty_overlay_accepts_mail_after_restart_without_reattachment(self): session, _ = self.editor.create() self.app.restart() self.assertEqual(attachment_state(self.app, session), "detached") refused( self, lambda: self.editor.prompt( session, [{"type": "text", "text": "stale editor prompt"}] ), 409, "attachment_lost", )
sender = self.send_fixture_mail(session) try: wait_until( lambda: any( row["kind"] == "assistant" for row in self.app.history(session)["items"] ), 15, "mail turn to finish with an empty editor overlay", ) except AssertionError: self.fail( f"recipient activity: {session_read(self.app, session)['activity']}; " f"sender history: {self.app.history(sender)}" )
def test_disabled_editor_mcp_does_not_block_mail_after_restart(self): session, _ = self.editor.create(servers=[fake_server()]) path = self.app.home / "extensions.json" settings = json.loads(path.read_text()) if path.exists() else {} settings.setdefault("enabled", {})["editor-mcp"] = False path.write_text(json.dumps(settings)) self.app.restart()
self.send_fixture_mail(session) wait_until( lambda: any( row["kind"] == "assistant" for row in self.app.history(session)["items"] ), 15, "mail turn to finish without the disabled overlay", )
def inject_failure(app, column): """Make the daemon's next write of `column` on sessions fail, as a full disk or a locked database would.""" with sqlite3.connect(app.home / "albedo.sqlite", timeout=10) as db: db.execute( f"CREATE TRIGGER refuse_{column} BEFORE UPDATE OF {column} ON sessions " "BEGIN SELECT RAISE(ABORT, 'injected failure'); END" )
def clear_failure(app, column): with sqlite3.connect(app.home / "albedo.sqlite", timeout=10) as db: db.execute(f"DROP TRIGGER refuse_{column}")
def attachment_state(app, session_id): return session_read(app, session_id)["acp"]["attachment"]
@exclusiveclass AcpOverlayFailureTest(unittest.TestCase): """An attach whose overlay cannot be recorded leaves the session as it was, live and after a restart: no writer and no overlay."""
def test_an_overlay_that_cannot_be_recorded_is_not_installed(self): with Albedo() as app: editor = AcpClient(app) session_id, _ = editor.create() editor.release(session_id) inject_failure(app, "acp_overlay_identities") with self.assertRaises(urllib.error.HTTPError): editor.attach(session_id, servers=[fake_server()]) unchanged = "detached" self.assertEqual(attachment_state(app, session_id), unchanged) app.restart() self.assertEqual(attachment_state(app, session_id), unchanged) clear_failure(app, "acp_overlay_identities") editor.attach(session_id, servers=[fake_server()])
@exclusiveclass AcpTraceSnapshotsTest(unittest.TestCase): def test_large_file_edit_keeps_snapshots_inline(self): def script(request): if request["messages"][-1].get("role") == "user": return python("files.edit('source.txt', 'line 0650:', 'changed 0650:')") return text("done")
provider = Provider(script) self.addCleanup(provider.close) with Albedo(provider) as app: original = "".join( f"line {index:04d}: existing source text\n" for index in range(1300) ) (app.workspace / "source.txt").write_text(original) editor = AcpClient(app) session, _ = editor.create() before = app.stream_page(session) editor.prompt(session, [{"type": "text", "text": "edit source"}]) app.idle(session) live = next( event["data"] for event in app.stream_page(session, before)["events"] if event["type"] == "tool" ) self.assertTrue(live["content_complete"], live) change = live["trace"]["changes"][0] self.assertEqual(change["old_text"], original) self.assertEqual( change["new_text"], original.replace("line 0650:", "changed 0650:") )
def test_history_preserves_snapshots_and_large_trace_reference(self): def script(request): if request["messages"][-1].get("role") == "user": return python( "from pathlib import Path\nPath('existing.txt').write_text('changed text')\nPath('new.txt').write_text('new text')\nPath('binary.txt').write_bytes(b'\\x00binary')\nPath('oversized.txt').write_text('x' * 140000)\nPath('large.txt').write_text('x' * 100000)\nPath('large2.txt').write_text('y' * 100000)\nPath('large3.txt').write_text('z' * 100000)" ) return text("done")
provider = Provider(script) self.addCleanup(provider.close) with Albedo(provider) as app: (app.workspace / "existing.txt").write_text("original text") editor = AcpClient(app) session, _ = editor.create() before = app.stream_page(session) editor.prompt(session, [{"type": "text", "text": "write files"}]) app.idle(session) live = next( event["data"] for event in app.stream_page(session, before)["events"] if event["type"] == "tool" ) self.assertTrue(live["content_complete"]) self.assertIsNotNone(live["trace"]) self.assertIsNone(live["reference"]) tool = next( e for e in app.history(session)["items"] if e["kind"] == "tool_result" ) self.assertEqual(tool["tool"]["status"], "completed") self.assertEqual(tool["tool"]["progress_call_id"], live["progress_call_id"]) reference = next( p["reference"] for p in tool["content"] if p["kind"] == "reference" and p["reference"]["field"] == "trace" ) pieces = [] url = reference["url"] while url: with app.api(url) as response: page = json.load(response) pieces.extend( part["text"] for part in page["parts"] if part["field"] == "trace" ) url = ( reference["url"] + "?next=" + page["next"] if page["next"] else None ) raw = "".join(pieces) self.assertEqual(len(raw.encode()), reference["bytes"]) trace = json.loads(raw) change = next(c for c in trace["changes"] if c["path"].endswith("new.txt")) self.assertIsNone(change["old_text"]) self.assertEqual(change["new_text"], "new text") changes = {Path(c["path"]).name: c for c in trace["changes"]} self.assertEqual(changes["existing.txt"]["old_text"], "original text") self.assertEqual(changes["existing.txt"]["new_text"], "changed text") for name in ("binary.txt", "oversized.txt"): self.assertEqual(changes[name]["kind"], "unavailable") self.assertTrue(changes[name]["reason"]) # Snapshots stop once the 256 KiB budget is spent. total = sum( len((c.get("old_text") or "").encode()) + len(c.get("new_text", "").encode()) for c in changes.values() ) self.assertLessEqual(total, 256 * 1024)
# Retained transcript traces take precedence over the live journal. # Seed the pre-snapshot wire shape as it appeared in older sessions. result = next( p["value"] for p in tool["content"] if p.get("field") == "result" ) cell_id = json.loads(result)["cell_id"] legacy = { "activities": [], "changes": [ { "path": "old.txt", "kind": "diff", "diff": "-old\n+new\n", "added": 1, "removed": 1, } ], "truncated": False, } with sqlite3.connect(app.home / "albedo.sqlite", timeout=10) as db: db.execute( "INSERT INTO transcript_traces(session, cell_id, payload) VALUES (?, ?, ?)", (session, cell_id, json.dumps(legacy)), ) row = next( e for e in app.history(session)["items"] if e["kind"] == "tool_result" ) self.assertEqual( next(p["trace"] for p in row["content"] if p["kind"] == "trace"), legacy )
if __name__ == "__main__": unittest.main()