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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204"""The nix and cargo shims through the real kernel: in a jj workspace without.git, `nix develop -c` runs a program in a dev shell built from theworkspace's commit (with its packages and shellHook) and cached, and nix readsthe flake from that commit instead of copying the directory; cargo goesthrough mbx, with the environment mbx needs to share builds across checkouts."""
import jsonimport osimport shutilimport subprocessimport sysimport unittestfrom pathlib import Path
ROOT = Path(__file__).resolve().parents[2]sys.path.insert(0, str(ROOT / "priv/python"))import albedo_shims # noqa: E402
from harness import Albedo, Provider, exclusive, python, text # noqa: E402
def script_for(test): def script(request): if request["messages"][-1].get("role") == "user": return python(test.code) return text("done")
return script
def cell_results(app, session): return [ json.loads(part["value"]) for entry in app.history(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" ]
FAKE_MBX = """#!/bin/shecho "mbx $* sdk=${SDKROOT-unset} cc=${CC-unset} mode=$MBX_CARGO_SHIM_MODE flags=$HOST_CFLAGS""""
class CargoShimTests(unittest.TestCase): def setUp(self): self.provider = Provider(script_for(self)) self.addCleanup(self.provider.close) self.app = Albedo(self.provider) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def test_cargo_runs_through_mbx_without_what_keeps_builds_apart(self): bin = self.app.workspace / "bin" bin.mkdir() for name, script in ( ("mbx", FAKE_MBX), ("cc", "#!/bin/sh\n"), ("gcc", "#!/bin/sh\n"), ): (bin / name).write_text(script) (bin / name).chmod(0o755) # the shell's compiler under another name, as clang is cc in a nix shell (bin / "clang").symlink_to("cc") session = self.app.session() self.code = ( f"path = {str(bin)!r} + ':' + os.environ['PATH']\n" "for cc, extra in (('clang', {}), ('clang', {'ALBEDO_NIX_ENV': 'abc'}),\n" " ('gcc', {'ALBEDO_NIX_ENV': 'abc'})):\n" " env = {'PATH': path, 'SDKROOT': '/nix/store/x-sdk', 'CC': cc, **extra}\n" " job = run('cargo', 'build', env=env)\n" " await job\n" " print(job.tail().strip())\n" ) self.app.prompt(session, "build").close() self.app.idle(session) lines = cell_results(self.app, session)[-1]["output"].splitlines() # outside a nix shell a chosen compiler stays self.assertEqual(lines[0], "mbx build sdk=unset cc=clang mode=1 flags=") # inside a nix shell the shell's own cc goes to mbx, and the wrapper is keyed self.assertEqual( lines[1], "mbx build sdk=unset cc=unset mode=1 flags= -DALBEDO_NIX_ENV_abc" ) # a shell that picks another compiler keeps it self.assertEqual( lines[2], "mbx build sdk=unset cc=gcc mode=1 flags= -DALBEDO_NIX_ENV_abc" )
def nixpkgs_lock(): """albedo's own locked nixpkgs: in the store wherever its dev shell is, so the test flake needs no network.""" lock = json.loads((ROOT / "flake.lock").read_text()) return lock["nodes"][lock["nodes"]["root"]["inputs"]["nixpkgs"]]
FLAKE = """{ inputs.nixpkgs.url = "github:%(owner)s/%(repo)s/%(ref)s"; outputs = { self, nixpkgs }: { devShells = nixpkgs.lib.genAttrs [ "aarch64-darwin" "x86_64-darwin" "aarch64-linux" "x86_64-linux" ] (system: let pkgs = nixpkgs.legacyPackages.${system}; in { default = pkgs.mkShell { packages = [ (pkgs.writeShellScriptBin "albedo-nix-probe" "echo probe:$ALBEDO_NIX_MARK") ]; shellHook = "export ALBEDO_NIX_MARK=from-hook"; }; }); marker = "read-from-the-commit"; };}"""
@exclusive@unittest.skipUnless(shutil.which("nix") and shutil.which("jj"), "needs nix and jj")class NixShellTests(unittest.TestCase): def setUp(self): def prepare(app): # Keep two inherited shim generations to cover nested daemon launches. source = albedo_shims.NIX inherited = [] for name in ("outer-a", "outer-b"): directory = app.workspace / name / "shims" / "nix" directory.mkdir(parents=True) shim = directory / "nix" shim.write_text(source) shim.chmod(0o755) inherited.append(str(directory)) app.env["PATH"] = os.pathsep.join([*inherited, app.env["PATH"]]) app.env["NIX_CONFIG"] = "experimental-features = nix-command flakes" # a first dev shell evaluates nixpkgs; let the cell finish in place app.env["ALBEDO_CELL_BACKGROUND_SECONDS"] = "900"
self.provider = Provider(script_for(self)) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, prepare=prepare) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def jj(self, *args, cwd): subprocess.run( ["jj", *args], cwd=cwd, env=self.app.env, check=True, capture_output=True, timeout=120, )
def test_a_jj_workspace_is_read_from_its_commit_and_its_shell_is_cached(self): node = nixpkgs_lock() main = self.app.workspace / "main" main.mkdir() (main / "flake.nix").write_text(FLAKE % node["original"]) lock = { "nodes": {"nixpkgs": node, "root": {"inputs": {"nixpkgs": "nixpkgs"}}}, "root": "root", "version": 7, } (main / "flake.lock").write_text(json.dumps(lock)) (main / ".gitignore").write_text("/build\n") self.jj("git", "init", cwd=main) self.jj("commit", "-m", "flake", cwd=main) workspace = self.app.workspace / "ws" self.jj("workspace", "add", "--name", "ws", str(workspace), cwd=main) self.assertFalse((workspace / ".git").exists()) (workspace / "build").mkdir() (workspace / "build" / "output").write_bytes(b"\0" * 1_000_000) session = self.app.session(workspace) self.code = ( "for name, argv in (('first', ['nix', 'develop', '-c', 'albedo-nix-probe']),\n" " ('again', ['nix', 'develop', '-c', 'albedo-nix-probe']),\n" " ('eval', ['nix', 'eval', '--raw', '.#marker'])):\n" " job = run(*argv, timeout=900)\n" " await job\n" " print(name, job.exit_code, json.dumps(job.tail()))\n" ) self.app.prompt(session, "use the dev shell").close() self.app.idle(session) result = cell_results(self.app, session)[-1] self.assertEqual(result["status"], "ok", result) runs = { name: (int(code), json.loads(output)) for name, code, output in ( line.split(" ", 2) for line in result["output"].splitlines() ) } self.assertEqual(runs["first"][0], 0, runs["first"]) self.assertIn("albedo: building the dev shell", runs["first"][1]) self.assertTrue(runs["first"][1].endswith("probe:from-hook\n")) # the cached shell runs the program and says nothing else self.assertEqual(runs["again"], (0, "probe:from-hook\n")) code, output = runs["eval"] self.assertEqual(code, 0, output) self.assertIn(f"git+file://{main.resolve()}?rev=", output) self.assertTrue(output.endswith("read-from-the-commit")) # nothing nix locked went into the main checkout self.assertEqual(json.loads((main / "flake.lock").read_text()), lock)
if __name__ == "__main__": unittest.main()