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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394"""CLI attachment and startup must preserve an independently owned real daemon."""
import jsonimport osimport ptyimport selectimport signalimport socketimport statimport subprocessimport sysimport termiosimport timeimport unittest
from inspect_support import Inspectionfrom harness import Albedo, Provider, exclusive, operation_id, python, text
def direct_start(daemon): daemon.start_launcher() daemon.await_ready()
def runtime_limits(app): expression = ( "Limits = [Call(erlang, system_info, [Key]) " "|| Key <- [process_limit, port_limit, schedulers_online]], " "io:put_chars(json:encode(Limits))" ) return json.loads(Inspection(app).evaluate(expression, timeout=20))
def gated_launcher(app): """A pipe barrier signals the real launch attempt before letting it exec.""" original = app.env["ALBEDO_DAEMON"] started, release = app.root / "launcher-started", app.root / "launcher-release" os.mkfifo(started) os.mkfifo(release) attempts = app.root / "launcher-attempts" wrapper = app.root / "gated-launcher" wrapper.write_text( f"#!{sys.executable}\n" "import os\n" f"with open({str(attempts)!r}, 'a') as attempts:\n" " attempts.write(str(os.getpid()) + '\\n')\n" " attempts.flush()\n" f"with open({str(started)!r}, 'w') as ready:\n" " ready.write(str(os.getpid()) + '\\n')\n" f"with open({str(release)!r}, 'rb', buffering=0) as gate:\n" " gate.read(1)\n" f"os.execv({original!r}, [{original!r}])\n" ) wrapper.chmod(0o700) app.env["ALBEDO_DAEMON"] = str(wrapper) return started, release, attempts
class TerminalCommand: def __init__(self, app, *arguments): self.master, slave = pty.openpty() termios.tcsetwinsize(slave, (24, 100)) self.process = app.daemon.start_cli( *arguments, stdin=slave, stdout=slave, stderr=slave ) os.close(slave) self.output = b""
def prompt(self, timeout=10): deadline = time.monotonic() + timeout while time.monotonic() < deadline: readable, _, _ = select.select( [self.master], [], [], max(0, deadline - time.monotonic()) ) if readable: try: chunk = os.read(self.master, 65536) except OSError as error: raise AssertionError( "CLI exited before restart prompt: " + self.output.decode(errors="replace") ) from error self.output += chunk if b"\x1b[6n" in chunk: os.write(self.master, b"\x1b[1;1R") if b"\x1b[c" in chunk: os.write(self.master, b"\x1b[?1;2c") if b"[y/N]" in self.output or b"[Y/n]" in self.output: return raise AssertionError( "CLI did not offer a restart: " + self.output.decode(errors="replace") )
def answer(self, value): os.write(self.master, (value + "\r").encode()) result = self.process.wait(timeout=20) os.close(self.master) return result
# exclusive: owns daemon startup and restart behavior@exclusiveclass LifecycleTest(unittest.TestCase): def test_standalone_default_home_authentication_and_vm_defaults_match_cli(self): def prepare(app): app.home = app.daemon.home = app.root / "user-home" / ".albedo" app.home.mkdir(mode=0o755) app.env.pop("ALBEDO_HOME") app.env.pop("ALBEDO_TOKEN", None) app.env["ALBEDO_INSPECT"] = "1" app.daemon.boot_action = direct_start
with Albedo(prepare=prepare) as app: direct = dict(app.connection) self.assertGreaterEqual(len(direct["token"]), 32) self.assertEqual(stat.S_IMODE(app.home.stat().st_mode), 0o700) self.assertEqual( stat.S_IMODE((app.home / "daemon.json").stat().st_mode), 0o600 ) connection = socket.create_connection( ("127.0.0.1", direct["port"]), timeout=5 ) with connection: connection.sendall( f"GET /server HTTP/1.1\r\nHost: localhost:{direct['port']}\r\nConnection: close\r\n\r\n".encode() ) self.assertIn(b"401", connection.recv(1024).split(b"\r\n", 1)[0]) direct_limits = runtime_limits(app) session = app.session() app.prompt(session, "independent daemon is usable").close() app.idle(session) self.assertIn( "ok", [ part["text"] for entry in app.history(session)["items"] for part in entry["content"] if part["kind"] == "text" ], ) self.assertEqual(app.connection, direct) app.restart() self.assertNotEqual(app.connection["pid"], direct["pid"]) self.assertNotEqual(app.connection["token"], direct["token"]) self.assertEqual(runtime_limits(app), direct_limits) app.restart( prepare=lambda fixture: fixture.env.update( ERL_FLAGS=fixture.env["ERL_FLAGS"] + " +P 131072 +Q 32768", ALBEDO_TOKEN="operator-chosen-authentication-token", ) ) self.assertEqual(runtime_limits(app)[:2], [131072, 32768]) self.assertEqual( app.connection["token"], "operator-chosen-authentication-token" ) self.assertEqual(app.api("/server").status, 200)
def test_cancelling_cli_startup_wait_leaves_the_started_daemon_owned(self): captured = {}
def prepare(app): started, release, attempts = gated_launcher(app)
def start(daemon): command = daemon.start_cli( "daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE ) with started.open() as ready: captured["child"] = int(ready.readline()) adopted = False try: command.send_signal(signal.SIGINT) stdout, stderr = command.communicate(timeout=5) captured["cancelled"] = command.returncode captured["output"] = stdout + stderr with release.open("wb", buffering=0) as gate: gate.write(b"x") daemon.await_ready() adopted = True finally: if not adopted: try: os.kill(captured["child"], signal.SIGKILL) except ProcessLookupError: pass captured["attempts"] = attempts.read_text().splitlines()
app.daemon.boot_action = start
with Albedo(prepare=prepare) as app: self.assertNotEqual(captured["cancelled"], 0, captured["output"]) self.assertEqual(app.connection["pid"], captured["child"]) self.assertEqual(len(captured["attempts"]), 1) session = app.session() app.prompt(session, "daemon survived waiter cancellation").close() app.idle(session) self.assertIn( "ok", [ part["text"] for entry in app.history(session)["items"] for part in entry["content"] if part["kind"] == "text" ], )
def test_simultaneous_cli_starters_converge_on_one_real_execution(self): captured = {}
def prepare(app): started, release, attempts = gated_launcher(app)
def start(daemon): first = daemon.start_cli( "daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE ) second = daemon.start_cli( "daemon", stdout=subprocess.PIPE, stderr=subprocess.PIPE ) with started.open() as ready: captured["owner"] = int(ready.readline()) adopted = False try: with release.open("wb", buffering=0) as gate: gate.write(b"x") captured["outputs"] = [ command.communicate(timeout=20) for command in (first, second) ] captured["codes"] = [first.returncode, second.returncode] daemon.await_ready() adopted = True finally: if not adopted: try: os.kill(captured["owner"], signal.SIGKILL) except ProcessLookupError: pass captured["attempts"] = attempts.read_text().splitlines()
app.daemon.boot_action = start
with Albedo(prepare=prepare) as app: self.assertEqual(captured["codes"], [0, 0], captured["outputs"]) self.assertEqual(captured["attempts"], [str(app.connection["pid"])]) self.assertEqual(app.connection["pid"], captured["owner"]) self.assertEqual(app.api("/sessions").status, 200)
def test_compatible_other_build_is_kept_without_a_terminal_and_explicit_yes_restarts( self, ): def prepare(app): app.env["ALBEDO_BUILD"] = "independent-compatible-build" app.daemon.boot_action = direct_start
with Albedo(prepare=prepare) as app: previous = dict(app.connection) self.assertEqual(previous["build"], "independent-compatible-build") # The daemon records the content digest of the code it runs, so a # client can prove a build difference without a label. self.assertRegex(previous["digest"], r"^[0-9a-f]{64}$") app.env.pop("ALBEDO_BUILD") app.cli("daemon") self.assertEqual( json.loads((app.home / "daemon.json").read_text()), previous ) terminal = TerminalCommand(app, "daemon") terminal.prompt() self.assertEqual(terminal.answer(""), 0) self.assertEqual( json.loads((app.home / "daemon.json").read_text()), previous ) for cancel in ("\x03", "\x04"): terminal = TerminalCommand(app, "daemon") terminal.prompt() self.assertNotEqual(terminal.answer(cancel), 0) self.assertEqual( json.loads((app.home / "daemon.json").read_text()), previous ) terminal = TerminalCommand(app, "daemon") terminal.prompt() self.assertEqual(terminal.answer("y"), 0) app.daemon._refresh() self.assertNotEqual(app.connection["pid"], previous["pid"]) self.assertNotEqual(app.connection["token"], previous["token"])
def test_restart_confirmation_cannot_replace_a_newer_daemon(self): with Albedo() as app: terminal = TerminalCommand(app, "daemon") terminal.prompt() app.restart() replacement = dict(app.connection) self.assertNotEqual(terminal.answer("y"), 0) self.assertEqual( json.loads((app.home / "daemon.json").read_text()), replacement ) self.assertEqual(app.api("/server").status, 200)
def test_explicit_upgrade_recovers_history_receipt_and_uncertain_tool_without_repeating_effect( self, ): fixture = {}
def script(request): messages = request["messages"] if any( "daemon restart" in str(message.get("content", "")) for message in messages ): return text("recovered after upgrade") if messages[-1]["role"] == "tool": return text("tool finished") return python( "from pathlib import Path\nimport threading\n" f"with Path({str(fixture['effect'])!r}).open('a') as effect:\n" " effect.write('effect applied\\n')\n" " effect.flush()\n" "threading.Event().wait(60)" )
provider = Provider(script) self.addCleanup(provider.close) with Albedo(provider) as app: fixture["effect"] = app.workspace / "effect" session = app.session() operation = operation_id() app.api( f"/sessions/{session}/inputs/{operation}", {"kind": "message", "text": "apply an effect then wait"}, method="PUT", ).close() deadline = time.monotonic() + 20 while ( not fixture["effect"].exists() or fixture["effect"].read_text() != "effect applied\n" ): if time.monotonic() > deadline: app._fail("tool never published its external effect") time.sleep(0.02) self.assertTrue( json.load(app.api(f"/sessions/{session}?tail=0"))["status"]["phase"] == "running" ) previous = dict(app.connection) terminal = TerminalCommand(app, "daemon") terminal.prompt() self.assertEqual(terminal.answer("y"), 0) app.daemon._refresh() self.assertNotEqual(app.connection["pid"], previous["pid"]) app.idle(session) receipt = json.load(app.api(f"/sessions/{session}/inputs/{operation}")) self.assertEqual(receipt["delivery"], "committed") self.assertEqual(receipt["turn"]["state"], "abandoned") entries = app.history(session)["items"] self.assertEqual( sum(entry["input_id"] == operation for entry in entries), 1 ) self.assertTrue( any( entry["kind"] == "note" and any( part["kind"] == "text" and "restart" in part["text"] for part in entry["content"] ) for entry in entries ) ) self.assertTrue( any( part["kind"] == "text" and part["text"] == "recovered after upgrade" for entry in entries for part in entry["content"] ) ) self.assertEqual(fixture["effect"].read_text(), "effect applied\n")
def test_failed_discovery_publication_exits_and_releases_home_for_retry(self): captured = {}
def prepare(app): def start(daemon): blocked = app.home / "daemon.json" blocked.mkdir() process = daemon.start_launcher() captured["exit"] = process.wait(timeout=20) captured["pid"] = process.pid blocked.rmdir()
app.daemon.boot_action = start
with Albedo(prepare=prepare) as app: self.assertNotEqual(captured["exit"], 0) self.assertNotEqual(app.connection["pid"], captured["pid"]) self.assertEqual(app.api("/server").status, 200)