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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912"""Daemon integration: provider binding, history, tools, workspaces, and recovery."""
import jsonfrom pathlib import Pathimport sqlite3import timeimport unittestimport urllib.error
from harness import Albedo, Provider, exclusive, python, text
def contents(item): content = item.get("content", "") if isinstance(content, str): return content return " ".join(part.get("text", "") for part in content if isinstance(part, dict))
def script(request): inputs = request.get("messages", request.get("input", [])) user = next( (contents(item) for item in reversed(inputs) if item.get("role") == "user"), "" ) tools = sum( item.get("role") == "tool" or item.get("type") == "function_call_output" for item in inputs ) if "continue from branch" in user: return text("finished") if "no first token" in user: return text("never streamed", hang=True) if "over 100 turns" in user: return text("finished") if tools >= 105 else python("1") if "workspace probe" in user: code = ( "import os\nfrom pathlib import Path\n" "assert 'workspace_marker' not in globals()\n" "workspace_marker = os.getcwd()\nPath('cwd-probe').write_text(os.getcwd())" ) done = ( inputs[-1].get("role") == "tool" or inputs[-1].get("type") == "function_call_output" ) return text("finished") if done else python(code) code = ( "import asyncio, os\nfrom pathlib import Path\n" "Path('example.txt').write_text('hello\\n')\n" "saved = Path('example.txt').read_text()\n" "assert 'ALBEDO_API_KEY' not in os.environ\n" "assert 'ALBEDO_TOKEN' not in os.environ\n" f"await asyncio.sleep({30 if 'hang' in user else 2 if 'first task' in user else 0.4})\nlen(saved)" ) return ( text("finished") if tools else python(code, reasoning="reasoning about the task") )
class IntegrationTest(unittest.TestCase): def setUp(self): self.provider = Provider(script) self.addCleanup(self.provider.close)
def app_for(self, protocol, *, prepare=None, providers=None): app = Albedo( self.provider, protocol=protocol, prepare=prepare, providers=providers ) app.__enter__() self.addCleanup(app.__exit__, None, None, None) return app
def read(self, app, path): with app.api(path) as response: return json.load(response)
def send(self, app, session, message): app.prompt(session, message).close() app.idle(session, timeout=90)
@exclusive def test_session_start_refuses_an_unsaved_default_effort(self): app = self.app_for("chat_completions") session = app.session()
def prepare(app): # A newly available catalog gives the persisted session a default. (app.home / "models.json").write_text("{}") with sqlite3.connect(app.home / "albedo.sqlite") as db: db.execute( "UPDATE sessions SET model='o3-unlisted', effort=NULL WHERE id=?", (session,), ) db.execute( "CREATE TRIGGER reject_effort BEFORE UPDATE OF effort ON sessions " "BEGIN SELECT RAISE(FAIL, 'fixture effort write failed'); END" )
app.restart(prepare=prepare) with self.assertRaises(urllib.error.HTTPError) as caught: self.read(app, f"/sessions/{session}/status") self.assertIn("session could not start", json.load(caught.exception)["error"]) with sqlite3.connect(app.home / "albedo.sqlite") as db: self.assertIsNone( db.execute( "SELECT effort FROM sessions WHERE id=?", (session,) ).fetchone()[0] ) db.execute("DROP TRIGGER reject_effort") self.assertFalse(self.provider.requests) self.assertIn( "fixture effort write failed", (app.home / "daemon.log").read_text() ) self.read(app, f"/sessions/{session}/status") with sqlite3.connect(app.home / "albedo.sqlite") as db: self.assertEqual( db.execute( "SELECT effort FROM sessions WHERE id=?", (session,) ).fetchone()[0], "medium", )
@exclusive def test_scheduler_reports_a_failed_occurrence_write_and_retries(self): def prepare(app): app.env["ALBEDO_SCHEDULE_TICK_MS"] = "50"
app = self.app_for("chat_completions", prepare=prepare) session = app.session() with sqlite3.connect(app.home / "albedo.sqlite") as db: db.execute( "CREATE TRIGGER reject_advance BEFORE DELETE ON schedules " "BEGIN SELECT RAISE(FAIL, 'fixture advance failed'); END" ) db.execute( "INSERT INTO schedules(session,kind,prompt,next_at) VALUES(?,'once',?,0)", (session, "scheduled fixture"), ) deadline = time.monotonic() + 15 while "fixture advance failed" not in (app.home / "daemon.log").read_text(): self.assertLess(time.monotonic(), deadline) time.sleep(0.05) self.assertIn( "schedule occurrence could not be advanced", (app.home / "daemon.log").read_text(), ) with sqlite3.connect(app.home / "albedo.sqlite") as db: self.assertEqual( db.execute("SELECT COUNT(*) FROM schedules").fetchone()[0], 1 ) db.execute("DROP TRIGGER reject_advance") while True: with sqlite3.connect(app.home / "albedo.sqlite") as db: if db.execute("SELECT COUNT(*) FROM schedules").fetchone()[0] == 0: break self.assertLess(time.monotonic(), deadline) time.sleep(0.05) app.idle(session) self.assertTrue(self.provider.requests)
@exclusive def test_deletion_refuses_a_failed_child_lookup(self): app = self.app_for("chat_completions") session = app.session() with sqlite3.connect(app.home / "albedo.sqlite") as db: db.execute("ALTER TABLE session_family RENAME TO unavailable_family") for suffix in ("", "?tree=1"): with self.subTest(tree=bool(suffix)): with self.assertRaises(urllib.error.HTTPError) as caught: app.api(f"/sessions/{session}{suffix}", method="DELETE") self.assertEqual(caught.exception.code, 409) self.assertIn("session_family", json.load(caught.exception)["error"]) with sqlite3.connect(app.home / "albedo.sqlite") as db: self.assertIsNotNone( db.execute( "SELECT id FROM sessions WHERE id=?", (session,) ).fetchone() ) with sqlite3.connect(app.home / "albedo.sqlite") as db: db.execute("ALTER TABLE unavailable_family RENAME TO session_family") with app.api(f"/sessions/{session}?tree=1", method="DELETE") as response: self.assertEqual(json.load(response)["deleted"], 1)
def settings(self, app, protocol, active="alpha", **models): other = "responses" if protocol == "chat_completions" else "chat_completions" configured = { "alpha": ("alpha-2", models.get("alpha", "fixture-alpha"), protocol), "beta": ("beta-1", models.get("beta", "fixture-beta"), protocol), "gamma": ("gamma-1", models.get("gamma", "fixture-gamma"), other), } providers = { name: { "baseUrl": self.provider.url + f"/{name}/v1", "apiKey": key, "model": model, "protocol": binding, } for name, (key, model, binding) in configured.items() } (app.home / "config.json").write_text( json.dumps({"active": active, "providers": providers}) )
def records(self, start): return self.provider.requests[start:]
def wait_for_request(self, start): deadline = time.monotonic() + 15 while len(self.provider.requests) <= start and time.monotonic() < deadline: time.sleep(0.05) self.assertGreater(len(self.provider.requests), start)
def assert_projected(self, record, protocol, users): items = record["request"][ "messages" if protocol == "chat_completions" else "input" ] serialized = json.dumps(items) for user in users: self.assertIn(user, serialized) self.assertIn("finished", serialized) if protocol == "chat_completions": calls = [ call["id"] for item in items if item.get("role") == "assistant" for call in item.get("tool_calls", []) ] outputs = [ item["tool_call_id"] for item in items if item.get("role") == "tool" ] else: calls = [ item["call_id"] for item in items if item.get("type") == "function_call" ] outputs = [ item["call_id"] for item in items if item.get("type") == "function_call_output" ] self.assertTrue(any(call.startswith("fixture-call-") for call in calls)) self.assertTrue(set(calls) & set(outputs))
@exclusive def test_unconfigured_startup_and_legacy_provider_migration(self): def prepare(app): workspace = app.root / "legacy-workspace" workspace.mkdir(exist_ok=True) with sqlite3.connect(app.home / "albedo.sqlite") as db: db.execute( "CREATE TABLE sessions(id TEXT PRIMARY KEY,cwd TEXT NOT NULL,model TEXT NOT NULL,protocol TEXT NOT NULL,stage TEXT NOT NULL DEFAULT 'idle')" ) for protocol in ("responses", "chat_completions"): db.execute( "INSERT INTO sessions(id,cwd,model,protocol) VALUES(?,?,?,?)", ( "legacy-" + protocol, str(workspace), "legacy-model", protocol, ), )
app = self.app_for("responses", prepare=prepare, providers={}) health = self.read(app, "/health") self.assertTrue(health["ok"]) self.assertEqual(health["version"], 2) self.assertTrue( {"session_provider", "session_workspace"} <= set(health["capabilities"]) ) with self.assertRaises(urllib.error.HTTPError) as caught: app.api("/sessions", {"workspace": str(app.workspace)}).close() self.assertEqual(caught.exception.code, 400) self.assertIn("/login", json.load(caught.exception)["error"]) for protocol in ("responses", "chat_completions"): with self.subTest(protocol=protocol): legacy = "legacy-" + protocol self.settings(app, protocol) config = json.loads((app.home / "config.json").read_text()) config["providers"]["default"] = { "baseUrl": self.provider.url + "/legacy/v1", "apiKey": "legacy-key", "model": "legacy-model", "protocol": protocol, } (app.home / "config.json").write_text(json.dumps(config)) start = len(self.provider.requests) self.send(app, legacy, "inspect the legacy workspace") requests = self.records(start) self.assertTrue(requests) self.assertTrue(all("/legacy/v1/" in r["path"] for r in requests)) self.assertTrue( all(r["authorization"] == "Bearer legacy-key" for r in requests) ) session = app.session() info = next( s for s in self.read(app, "/sessions") if s["id"] == session ) self.assertEqual( (info["provider"], info["title"]), ("alpha", "new session") ) pid = app.connection["pid"] app.cli("sessions") self.assertEqual( json.loads((app.home / "daemon.json").read_text())["pid"], pid ) self.restart(app) info = next(s for s in self.read(app, "/sessions") if s["id"] == legacy) self.assertEqual( (info["provider"], info["title"]), ("default", "inspect the legacy workspace"), )
def restart(self, app): app.restart(crash=True)
def test_image_metadata_is_verified_and_only_metadata_is_exposed(self): for protocol in ("responses", "chat_completions"): with self.subTest(protocol=protocol): app = self.app_for(protocol) session = app.session() png = "iVBORw0KGgoAAAANSUhEUgAAAAIAAAAD" attachment = { "mimeType": "image/png", "data": png, "width": 2, "height": 3, "bytes": 24, } with self.assertRaises(urllib.error.HTTPError) as caught: app.api( f"/sessions/{session}/events", { "content": "reject this", "image": dict(attachment, width=4), }, ).close() self.assertEqual(caught.exception.code, 409) start = len(self.provider.requests) app.api( f"/sessions/{session}/events", { "content": "describe this image", "image": attachment, }, ).close() app.idle(session) self.send(app, session, "confirm image history") requests = self.records(start) self.assertEqual(len(requests), 3) self.assertTrue( all( "data:image/png;base64," + png in json.dumps(r["request"]) for r in requests ) ) image = next( e["image"] for e in app.events(session) if e.get("type") == "user" and e.get("image") ) self.assertEqual( image, {"mimeType": "image/png", "width": 2, "height": 3, "bytes": 24}, ) self.assertNotIn(png, json.dumps(image))
def test_busy_messages_join_next_model_request_in_order(self): app = self.app_for("responses") session = app.session() start = len(self.provider.requests) with app.api( f"/sessions/{session}/events", {"content": "first task"} ) as response: self.assertFalse(json.load(response)["queued"]) self.wait_for_request(start) with app.api( f"/sessions/{session}/events", {"content": "queued direction"} ) as response: self.assertTrue(json.load(response)["queued"]) app.prompt(session, "another direction").close() app.idle(session) requests = self.records(start) self.assertEqual(len(requests), 2) first, second = (json.dumps(r["request"]) for r in requests) self.assertNotIn("queued direction", first) self.assertIn("queued direction", second) self.assertIn("another direction", second) self.assertLess( second.index("queued direction"), second.index("another direction") ) self.assertTrue( any( e.get("type") == "user" and e.get("text") == "queued direction" for e in app.events(session) ) )
@exclusive def test_workspace_repair_validates_path_and_preserves_history(self): app = self.app_for("responses") old = app.root / "original-workspace" old.mkdir() moved = app.root / "moved-workspace" session = app.session(old) self.send(app, session, "workspace probe before move") self.assertEqual(Path((old / "cwd-probe").read_text()).resolve(), old.resolve()) old.rename(moved) pid = app.connection["pid"] before = app.events(session) with self.assertRaises(urllib.error.HTTPError) as caught: app.prompt(session, "workspace probe after move").close() self.assertEqual(caught.exception.code, 409) failure = json.load(caught.exception) self.assertEqual(failure["code"], "workspace_missing") self.assertEqual(failure["workspace"], str(old)) for invalid in ("relative", str(old), str(moved / "cwd-probe")): with self.assertRaises(urllib.error.HTTPError) as caught: app.api( f"/sessions/{session}/workspace", {"workspace": invalid} ).close() self.assertEqual(caught.exception.code, 409) with app.api( f"/sessions/{session}/workspace", {"workspace": str(moved)} ) as response: self.assertEqual(json.load(response)["workspace"], str(moved)) self.assertEqual( next(s for s in self.read(app, "/sessions") if s["id"] == session)[ "workspace" ], str(moved), ) self.assertEqual(json.loads((app.home / "daemon.json").read_text())["pid"], pid) self.send(app, session, "workspace probe after move") self.assertEqual( Path((moved / "cwd-probe").read_text()).resolve(), moved.resolve() ) self.assertFalse(old.exists()) after = app.events(session) self.assertEqual( [e for e in before if e.get("type") == "user"], [e for e in after if e.get("type") == "user"][:-1], ) with sqlite3.connect(app.home / "albedo.sqlite") as db: self.assertEqual( db.execute( "SELECT cwd FROM sessions WHERE id=?", (session,) ).fetchone()[0], str(moved), ) self.restart(app) info = next(s for s in self.read(app, "/sessions") if s["id"] == session) self.assertEqual(info["workspace"], str(moved))
@exclusive def test_tools_trace_detach_and_fork_checkpoint(self): for protocol in ("responses", "chat_completions"): with self.subTest(protocol=protocol): app = self.app_for(protocol) session = app.session() app.prompt(session, "write and inspect a file").close() app.events(session) app.idle(session) events = app.events(session) self.assertEqual((app.workspace / "example.txt").read_text(), "hello\n") self.assertTrue( any( e.get("type") == "message" and e.get("text") == "finished" for e in events ) ) self.assertTrue( any( e.get("type") == "thinking" and e.get("text") == "reasoning about the task" for e in events ) ) tool = next(e for e in events if e.get("type") == "tool") self.assertEqual(json.loads(tool["result"])["status"], "ok") self.assertTrue( any( a["kind"] == "read" and a["target"].endswith("example.txt") for a in tool["trace"]["activities"] ) ) self.assertTrue( any( c["path"].endswith("example.txt") for c in tool["trace"]["changes"] ) ) source = self.read(app, f"/sessions/{session}/tree?after=0&limit=100") items = source["items"] self.assertFalse(source["hasMore"]) self.assertEqual( {i["type"] for i in items}, {"user", "assistant", "tool"} ) if protocol == "responses": self.assertTrue( any(i["preview"].startswith("[reasoning]") for i in items) ) checkpoint = next(i for i in items if i["preview"] == "call python") with app.api( f"/sessions/{session}/fork", {"checkpoint": checkpoint["id"]} ) as response: branch = json.load(response) self.assertEqual(branch["title"], "write and inspect a file") self.assertEqual( (branch["provider"], branch["model"], branch["protocol"]), ("fixture", "fixture-model", protocol), ) branch_items = self.read( app, f"/sessions/{branch['id']}/tree?after=0&limit=100" )["items"] self.assertEqual( branch_items[-1]["preview"], "not executed after branch checkpoint" ) self.assertEqual( [i["preview"] for i in branch_items[:-1]], [i["preview"] for i in items if i["id"] <= checkpoint["id"]], ) start = len(self.provider.requests) self.send(app, branch["id"], "continue from branch") requests = self.records(start) self.assertEqual(len(requests), 1) self.assertTrue( all( "/chat/completions" in r["path"] if protocol == "chat_completions" else "/responses" in r["path"] for r in requests ) ) inputs = requests[0]["request"][ "messages" if protocol == "chat_completions" else "input" ] outputs = [ i for i in inputs if i.get("role") == "tool" or i.get("type") == "function_call_output" ] self.assertEqual( [i.get("content", i.get("output")) for i in outputs], ["not executed after branch checkpoint"], ) self.assertTrue( any( e.get("type") == "message" and e.get("text") == "finished" for e in app.events(branch["id"]) ) ) self.assertEqual( self.read(app, f"/sessions/{session}/tree?after=0&limit=100")[ "items" ], items, ) self.assertEqual( next(s for s in self.read(app, "/sessions") if s["id"] == session)[ "title" ], "write and inspect a file", ) self.send(app, session, "review the result") self.assertEqual( next(s for s in self.read(app, "/sessions") if s["id"] == session)[ "title" ], "review the result", ) timestamps = [ (e["type"], e["text"], e.get("timestamp")) for e in app.events(session) if e.get("type") in ("user", "message") ] self.assertTrue(timestamps) self.assertTrue( all(isinstance(stamp, int) for _, _, stamp in timestamps) ) self.restart(app) self.assertEqual( [ (e["type"], e["text"], e.get("timestamp")) for e in app.events(session) if e.get("type") in ("user", "message") ], timestamps, )
@exclusive def test_provider_switch_projects_history_across_protocols_and_restart(self): for protocol in ("responses", "chat_completions"): with self.subTest(protocol=protocol): app = self.app_for(protocol) self.settings(app, protocol) session = app.session() route = f"/sessions/{session}/commands" app.api( route, { "name": "/model", "args": { "provider": "alpha", "model": "initial-alpha", }, }, ).close() start = len(self.provider.requests) self.send(app, session, "build switch history") self.assertGreaterEqual(len(self.records(start)), 2) self.assertTrue( all("/alpha/v1/" in r["path"] for r in self.records(start)) ) other = ( "responses" if protocol == "chat_completions" else "chat_completions" ) for provider, model, expected in ( ("gamma", "chosen-gamma", other), (None, "renamed-gamma", other), ): args = {"model": model} if provider: args["provider"] = provider with app.api(route, {"name": "/model", "args": args}) as response: result = json.load(response) self.assertEqual( result, { "result": { "provider": "gamma", "model": model, "protocol": expected, "effort": None, } }, ) defaults = json.loads((app.home / "config.json").read_text()) self.assertEqual(defaults["active"], "gamma") self.assertEqual(defaults["providers"]["gamma"]["model"], model) with app.api( "/sessions", {"workspace": str(app.workspace)} ) as response: default = json.load(response) self.assertEqual( (default["provider"], default["model"]), ("gamma", "renamed-gamma") ) with self.assertRaises(urllib.error.HTTPError) as caught: app.api( route, { "name": "/model", "args": { "provider": "unknown", "model": "wrong", }, }, ).close() self.assertEqual(caught.exception.code, 409) start = len(self.provider.requests) self.send(app, session, "after first switch") gamma = self.records(start) self.assertTrue(gamma) self.assertTrue( all( "/gamma/v1/" in r["path"] and r["model"] == "renamed-gamma" and r["authorization"] == "Bearer gamma-1" for r in gamma ) ) self.assert_projected( gamma[0], other, ["build switch history", "after first switch"] ) with app.api( route, { "name": "/model", "args": { "provider": "alpha", "model": "returned-alpha", }, }, ) as response: result = json.load(response) self.assertEqual( result, { "result": { "provider": "alpha", "model": "returned-alpha", "protocol": protocol, "effort": None, } }, ) defaults = json.loads((app.home / "config.json").read_text()) self.assertEqual(defaults["active"], "alpha") self.assertEqual( defaults["providers"]["alpha"]["model"], "returned-alpha" ) start = len(self.provider.requests) self.send(app, session, "after second switch") returned = self.records(start) self.assertTrue(returned) self.assertTrue( all( "/alpha/v1/" in r["path"] and r["model"] == "returned-alpha" for r in returned ) ) users = [ "build switch history", "after first switch", "after second switch", ] self.assert_projected(returned[0], protocol, users) info = next( s for s in self.read(app, "/sessions") if s["id"] == session ) self.assertEqual( (info["provider"], info["model"], info["protocol"]), ("alpha", "returned-alpha", protocol), ) self.restart(app) start = len(self.provider.requests) self.send(app, session, "after persisted restart") restarted = self.records(start) self.assertTrue(restarted) self.assertTrue( all( "/alpha/v1/" in r["path"] and r["model"] == "returned-alpha" for r in restarted ) ) self.assert_projected( restarted[0], protocol, users + ["after persisted restart"] )
@exclusive def test_provider_config_reload_and_interrupt_recovery(self): for protocol in ("responses", "chat_completions"): with self.subTest(protocol=protocol): app = self.app_for(protocol) self.settings(app, protocol) initial = json.loads((app.home / "config.json").read_text()) initial["providers"]["alpha"]["apiKey"] = "alpha-1" (app.home / "config.json").write_text(json.dumps(initial)) session = app.session() self.settings(app, protocol, active="beta") start = len(self.provider.requests) self.send(app, session, "write and inspect a file") alpha = self.records(start) self.assertTrue(alpha) self.assertTrue( all( "/alpha/v1/" in r["path"] and r["authorization"] == "Bearer alpha-2" and r["model"] == "fixture-alpha" for r in alpha ) ) second = app.session() info = next(s for s in self.read(app, "/sessions") if s["id"] == second) self.assertEqual( (info["provider"], info["model"]), ("beta", "fixture-beta") ) start = len(self.provider.requests) app.prompt(second, "hang").close() self.wait_for_request(start) for operation, body in ( ("workspace", {"workspace": str(app.root)}), ( "commands", { "name": "/model", "args": { "provider": "gamma", "model": "busy-rejected", }, }, ), ): with self.assertRaises(urllib.error.HTTPError) as caught: app.api(f"/sessions/{second}/{operation}", body).close() self.assertEqual(caught.exception.code, 409) info = next(s for s in self.read(app, "/sessions") if s["id"] == second) self.assertEqual( (info["workspace"], info["provider"]), (str(app.workspace), "beta") ) app.api(f"/sessions/{second}/interrupt", {}).close() app.idle(second) beta = self.records(start) self.assertTrue(beta) self.assertTrue( all( "/beta/v1/" in r["path"] and r["authorization"] == "Bearer beta-1" for r in beta ) ) self.assertTrue( any(e.get("type") == "user" for e in app.events(second)) ) third = app.session() app.prompt(third, "hang then recover").close() time.sleep(0.5) stamp = (app.workspace / "example.txt").stat().st_mtime_ns self.settings( app, protocol, active="alpha", beta="changed-beta-default" ) config = json.loads((app.home / "config.json").read_text()) config["providers"]["beta"]["apiKey"] = "beta-2" (app.home / "config.json").write_text(json.dumps(config)) start = len(self.provider.requests) self.restart(app) app.idle(third) restored = app.events(third) self.assertTrue( any( e.get("type") == "message" and e.get("text") == "finished" for e in restored ) ) resumed = self.records(start) self.assertTrue(resumed) self.assertTrue( all( "/beta/v1/" in r["path"] and r["authorization"] == "Bearer beta-2" and r["model"] == "fixture-beta" for r in resumed ) ) inputs = resumed[0]["request"][ "messages" if protocol == "chat_completions" else "input" ] note = [contents(i) for i in inputs if i.get("role") == "user"][-1] self.assertTrue( note.startswith( '<system-note origin="daemon restart">albedo restarted' ) ) self.assertIn("<system-note>The python kernel", note) markers = [ e for e in restored if e.get("type") == "user" and e.get("source") == "daemon restart" ] self.assertEqual(len(markers), 1) self.assertTrue(markers[0]["text"].startswith("albedo restarted")) sessions = {s["id"]: s for s in self.read(app, "/sessions")} self.assertEqual( (sessions[session]["provider"], sessions[session]["title"]), ("alpha", "write and inspect a file"), ) self.assertEqual( (sessions[third]["provider"], sessions[third]["title"]), ("beta", "hang then recover"), ) self.assertEqual( (app.workspace / "example.txt").stat().st_mtime_ns, stamp ) self.assertTrue( any(e.get("type") == "message" for e in app.events(session)) )
def test_interrupt_before_first_token_settles_quickly(self): for protocol in ("responses", "chat_completions"): with self.subTest(protocol=protocol): app = self.app_for(protocol) session = app.session() start = len(self.provider.requests) app.prompt(session, "no first token").close() self.wait_for_request(start) before = time.monotonic() app.api(f"/sessions/{session}/interrupt", {}).close() app.idle(session) # The provider holds the stream open without ever sending a # token, so only the worker noticing the stop latch while it # waits can settle the interrupt: the Abort backstop 2.5s in # is the old path, and the provider answers after 60s. self.assertLess(time.monotonic() - before, 1.0) # The interrupted session takes a new turn and commits it. self.send(app, session, "continue from branch") self.assertTrue( any( e.get("type") == "message" and e.get("text") == "finished" for e in app.events(session) ) )
def test_more_than_hundred_tool_turns_complete(self): for protocol in ("responses", "chat_completions"): with self.subTest(protocol=protocol): app = self.app_for(protocol) session = app.session() start = len(self.provider.requests) self.send(app, session, "continue for over 100 turns") self.assertEqual(len(self.records(start)), 106) events = app.events(session) self.assertEqual(sum(e.get("type") == "tool" for e in events), 105) self.assertTrue( any( e.get("type") == "message" and e.get("text") == "finished" for e in events ) )