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.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798"""A Python crash injects exactly one reset notice without corrupting history; adaemon crash leaves the detached kernel alive, so it injects only the line thatsays nothing was lost."""
import jsonimport urllib.errorimport unittest
from harness import resident_kernelfrom harness import latest_userfrom harness import Albedo, Provider, exclusive, python, text
NOTICE = ( "<system-note>The python kernel got reset and all variables are lost</system-note>")REATTACHED = ( "<system-note>The python kernel reattached; its variables and jobs carried on." " Nothing was running.</system-note>")
# exclusive: crashes and restarts the daemon@exclusiveclass KernelResetTests(unittest.TestCase): def setUp(self): def script(request): if request["input"][-1].get("role") == "user": prompt = latest_user(request) if prompt == "lose kernel": return python("import os; os._exit(1)") if prompt == "recover kernel" + NOTICE: return python("print('kernel-ready')") return text("ok")
self.provider = Provider(script) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses") self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def turn(self, session, prompt): self.app.prompt(session, prompt).close() self.app.idle(session) return latest_user(self.provider.requests[-1]["request"])
@resident_kernel def test_kernel_crash_emits_one_shot_reset_notice_and_daemon_crash_none(self): app = self.app session = app.session() self.assertEqual(self.turn(session, "first prompt"), "first prompt") unused = app.session() app.restart(crash=True) self.assertEqual( self.turn(unused, "unused session's first prompt"), "unused session's first prompt", ) with self.assertRaises(urllib.error.HTTPError) as rejected: app.prompt(session, " ").close() self.assertEqual(rejected.exception.code, 400) self.assertEqual( self.turn(session, "after restart"), "after restart" + REATTACHED ) self.assertEqual(self.turn(session, "ordinary turn"), "ordinary turn") self.assertEqual(self.turn(session, "lose kernel"), "lose kernel") self.assertEqual( self.turn(session, "recover kernel"), "recover kernel" + NOTICE ) self.assertTrue( any( "kernel-ready" in item.get("output", "") for item in self.provider.requests[-1]["request"]["input"] ) ) self.assertEqual(self.turn(session, "still alive"), "still alive") users = [ part["text"] for entry in app.history(session)["items"] if entry["kind"] == "user" for part in entry["content"] if part["kind"] == "text" ] self.assertEqual( users, [ "first prompt", "after restart", "ordinary turn", "lose kernel", "recover kernel", "still alive", ], ) with app.api("/sessions") as response: info = next( item for item in json.load(response)["items"] if item["id"] == session ) self.assertEqual(info["name"], "still alive")