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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106"""Shared editable daemon and model-driven cells for kernel E2E scenarios."""
import jsonfrom pathlib import Pathimport shutilimport subprocessimport tempfile
import harnessfrom harness import Albedo, Provider, prompt_text, turn_notes
MARKER = '\nMARKER = "new bundle"\n'
def process_name(pid): """The OS name shown by process viewers, without the executable directory.""" return Path( subprocess.check_output( ["ps", "-p", str(pid), "-o", "comm="], text=True ).strip() ).name
def editable_daemon(root): """A copy of the test daemon whose python bundle is ours to edit: every package links back to the snapshot except albedo's priv/python.""" assert harness._executable is not None, "the runner snapshots the daemon first" source = Path(harness._executable).parent out = Path(tempfile.mkdtemp(prefix="daemon-", dir=root)) for package in source.iterdir(): if package.is_dir() and package.name != "albedo": (out / package.name).symlink_to(package) albedo = out / "albedo" (albedo / "priv").mkdir(parents=True) (albedo / "ebin").symlink_to(source / "albedo" / "ebin") priv = (source / "albedo" / "priv").resolve() for entry in priv.iterdir(): if entry.name == "python": shutil.copytree( entry, albedo / "priv" / "python", ignore=shutil.ignore_patterns("__pycache__"), ) else: (albedo / "priv" / entry.name).symlink_to(entry) launcher = out / "albedo-daemon" launcher.write_text( "#!/bin/sh\n" f"exec erl -pa {out}/*/ebin -eval 'albedo@@main:run(albedo)' -noshell -extra\n" ) launcher.chmod(0o755) return launcher, albedo / "priv" / "python"
def linked_daemon(root): """A copy of the test daemon whose albedo/priv is a link to the snapshot's priv, as a checkout's build tree links it to the source priv. Returns the launcher and that link.""" assert harness._executable is not None, "the runner snapshots the daemon first" source = Path(harness._executable).parent out = Path(tempfile.mkdtemp(prefix="daemon-", dir=root)) for package in source.iterdir(): if package.is_dir() and package.name != "albedo": (out / package.name).symlink_to(package) albedo = out / "albedo" albedo.mkdir() (albedo / "ebin").symlink_to(source / "albedo" / "ebin") (albedo / "priv").symlink_to((source / "albedo" / "priv").resolve()) launcher = out / "albedo-daemon" launcher.write_text( "#!/bin/sh\n" f"exec erl -pa {out}/*/ebin -eval 'albedo@@main:run(albedo)' -noshell -extra\n" ) launcher.chmod(0o755) return launcher, albedo / "priv"
class ModelCells: """Run queued Python cells through the fixture's provider and session."""
cells: list[str] app: Albedo provider: Provider session: str
def cell(self, code): """Run one cell through a model turn; its result, and the turn's prompt followed by the notes the model saw with it or after the cell.""" self.cells.append(code) self.app.prompt(self.session, "run it").close() self.app.idle(self.session) results = [ json.loads(part["value"]) for entry in self.app.history(self.session)["items"] if entry["kind"] == "tool_result" and entry["tool"]["name"] == "python" for part in entry["content"] if part["kind"] == "json" and part["field"] == "result" ] request = next( r["request"] for r in reversed(self.provider.requests) if any(i.get("role") == "user" for i in r["request"]["input"]) ) prompt = prompt_text(request) + "".join(turn_notes(request)) return results[-1], prompt