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.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677"""A python tree that fails at startup never leaves a session without a kernel.
In a checkout the daemon boots new kernels straight from the working tree, so ahalf-saved edit there used to break every fresh kernel in every session, withonly "python exited at startup" to go on. Each tree a fresh kernel startscleanly from is saved under the home; a fresh boot that fails starts from thatcopy instead, and the model is told why, with the traceback's last lines. Oncethe tree is fixed, the next fresh kernel runs it again."""
import jsonfrom pathlib import Pathimport unittest
import harnessfrom harness import Albedo, Provider, exclusive, python, textfrom kernel_support import ModelCells, editable_daemon
BROKEN = '\nraise TypeError("broken on purpose")\n'
class Fallback(ModelCells, unittest.TestCase): bundle: Path
def setUp(self): self.cells = []
def script(request): if request["input"][-1].get("role") == "user" and self.cells: return python(self.cells.pop(0)) return text("ok")
def prepare(app): launcher, self.bundle = editable_daemon(app.root) app.env["ALBEDO_DAEMON"] = str(launcher)
self.provider = Provider(script) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses", prepare=prepare) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def fresh(self, code): """A cell in a new session, so its kernel is a fresh boot.""" self.session = self.app.session() return self.cell(code)
@exclusive def test_a_tree_that_fails_at_startup_boots_the_last_one_that_started(self): api = self.bundle / "albedo_api.py" good = self.app.home / "kernel-bundles" / "good" first, prompt = self.fresh("import albedo_api\nhasattr(albedo_api, 'BROKEN')") self.assertEqual(first.get("status"), "ok", first) self.assertNotIn("started cleanly", prompt) harness.wait_until(good.exists, 10, "the tree that booted was never saved")
original = api.read_text() api.write_text(original + BROKEN) fallen, prompt = self.fresh("import albedo_api\nalbedo_api.__file__") self.assertEqual(fallen.get("status"), "ok", fallen) # It runs the saved copy, not the broken tree. self.assertIn("kernel-bundles", fallen["value"]) self.assertIn("runs the last albedo python code that started cleanly", prompt) self.assertIn("TypeError: broken on purpose", prompt) with self.app.api(f"/sessions/{self.session}?tail=0") as response: self.assertFalse(json.load(response)["kernel"]["stale"])
api.write_text(original) fixed, prompt = self.fresh("import albedo_api\nalbedo_api.__file__") self.assertEqual(fixed.get("status"), "ok", fixed) self.assertNotIn("kernel-bundles", fixed["value"]) self.assertNotIn("started cleanly", prompt)
if __name__ == "__main__": unittest.main()