Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620#!/usr/bin/env python3"""Exercise the packed bridge at the historical npm entrypoint.
The installer fixture isolates launcher behavior; the legacy-update end-to-endsuite separately exercises the real Rust installer and published TS clients."""import hashlibimport ioimport jsonimport osfrom pathlib import Pathimport shutilimport signalimport subprocessimport sysimport tarfileimport tempfileimport unittestfrom unittest.mock import patch
import npm_bridgefrom npm_bridge import assemble
def rust_socket(tmp, uid=None): """Mirror pa-daemon socket_dir(): reserve the longest worker endpoint.""" uid = os.getuid() if uid is None else uid tmp = str(tmp) worker = f"{tmp}/prime-agent-rust-{uid}/worker-000000000000-000000000000.sock" if len(worker.encode()) > 103: suffix = hashlib.sha256(tmp.encode()).hexdigest()[:12] return f"/tmp/prime-agent-rust-{uid}-{suffix}/daemon.sock" return f"{tmp}/prime-agent-rust-{uid}/daemon.sock"
@unittest.skipUnless(sys.platform in ("darwin", "linux") and shutil.which("node") and shutil.which("npm"), "requires Node and npm on a historical TypeScript platform")class NpmBridgeTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.repo = self.root / "repo" self.repo.mkdir() self.fallback_tarball = self.root / "fallback.tgz" fallback_files = { "package/package.json": json.dumps({"name": "prime-agent", "version": "0.9.8", "dependencies": {}, "scripts": {"postinstall": "exit 91"}}).encode(), "package/dist/bundle/cli-node.js": b'''#!/usr/bin/env nodeif (process.argv[2] === "--version") console.log("0.9.8");else if (process.argv[2] === "entrypoint") console.log(process.argv[1]);else if (process.argv[2] === "prefix") console.log(process.env.npm_config_prefix);else if (process.argv[2] === "descriptor") console.log(require("node:fs").readFileSync(Number(process.env.TEST_DESCRIPTOR), "utf8"));else if (process.argv[2] === "ipc") { process.send({fallback:true}); process.disconnect(); }else if (process.argv.includes("update") && !process.argv.includes("--print")) { if (process.env.PRIME_AGENT_INTERNAL_TS_UPDATE) throw new Error("delegation flag leaked into the TS updater"); const {spawnSync}=require("node:child_process"); let installed={status:0}; if (process.env.FUTURE_BRIDGE_TARBALL) { installed=spawnSync("npm",["install","--global","--ignore-scripts","--no-audit","--no-fund", "--prefix",process.env.npm_config_prefix,process.env.FUTURE_BRIDGE_TARBALL],{encoding:"utf8"}); } if (installed.status !== 0) { console.error(installed.stderr); process.exitCode=installed.status || 1; } else { // Historical TS CLI warns on a failed coordinator but can still exit zero. // The outer bridge must verify activation instead of trusting that status. const result=spawnSync(process.execPath,[process.argv[1],"update","--internal-update-restart-coordinator"], {env:{...process.env,EXPECT_DEFER:"1"},stdio:"ignore"}); if (result.status !== 0) console.error("TypeScript coordinator warning: native migration failed"); process.exitCode=0; }}else console.log(JSON.stringify({fallback:true,args:process.argv.slice(2)}));''', } with tarfile.open(self.fallback_tarball, "w:gz") as archive: for name, data in fallback_files.items(): member = tarfile.TarInfo(name) member.size = len(data) member.mode = 0o755 archive.addfile(member, io.BytesIO(data)) # Unit tests use an offline synthetic payload; release code has no # checksum override and only accepts the immutable published archive. fallback_pin = patch.object(npm_bridge, "FALLBACK_SHA256", hashlib.sha256(self.fallback_tarball.read_bytes()).hexdigest()) fallback_pin.start() self.addCleanup(fallback_pin.stop) self.prefix = self.root / "native prefix" self.npm_prefix = self.root / "npm prefix" self.installs = self.root / "installs" self.fixture = self.root / "native.js" self.fixture.write_text('''#!/usr/bin/env nodeif (process.argv[2] === "wait") { console.log("ready"); setInterval(() => {}, 1000); }else if (process.argv[2] === "--version") console.log(require("node:fs").readFileSync(require("node:path").join(__dirname,"version"),"utf8").trim());else if (process.argv[2] === "socket") console.log(process.env.PRIME_AGENT_DAEMON_SOCKET);else if (process.argv[2] === "signal") process.kill(process.pid, "SIGTERM");else { console.log(JSON.stringify(process.argv.slice(2))); process.exitCode = Number(process.env.NATIVE_EXIT || 0); }''') self.fixture.chmod(0o755) (self.repo / "install-rust.sh").write_text('''#!/bin/shset -eutest "$PRIME_AGENT_VERSION" = "${EXPECT_VERSION:-1.0.0}"test "$PRIME_AGENT_RELEASE_CHANNEL" = "stable"test "$PRIME_AGENT_PRESERVE_NPM_BRIDGE" = "1"test -f "$PRIME_AGENT_NPM_BRIDGE_ENTRYPOINT"echo install >> "$INSTALL_LOG"echo "installer output stays on stderr"if [ "${FAIL_INSTALL:-0}" = 1 ]; then exit 23; fitest "$PRIME_AGENT_DEFER_DAEMON_STOP" = "${EXPECT_DEFER:-0}"test "$PRIME_AGENT_USE_LEGACY_DAEMON_SOCKET" = 1mkdir -p "$PRIME_AGENT_RUST_PREFIX/share/prime-agent"cp "$FIXTURE_NATIVE" "$PRIME_AGENT_RUST_PREFIX/share/prime-agent/prime-agent"printf '%s\\n' "$PRIME_AGENT_VERSION" > "$PRIME_AGENT_RUST_PREFIX/share/prime-agent/version"mkdir -p "$PRIME_AGENT_RUST_PREFIX/share/.prime-agent-npm-bridge"touch "$PRIME_AGENT_RUST_PREFIX/share/.prime-agent-npm-bridge/legacy-daemon-socket"''') self.out = self.root / "out" self.metadata = assemble(self.repo, self.out, "1.0.0", "stable", self.fallback_tarball) self.env = {**os.environ, "PRIME_AGENT_RUST_PREFIX": str(self.prefix), "INSTALL_LOG": str(self.installs), "FIXTURE_NATIVE": str(self.fixture), "npm_config_cache": str(self.root / "npm-cache")} self.env.pop("NODE_OPTIONS", None) installed = subprocess.run(["npm", "install", "--global", "--ignore-scripts", "--no-audit", "--no-fund", "--prefix", str(self.npm_prefix), str(self.out / self.metadata["tarball"])], env=self.env, capture_output=True, text=True) self.assertEqual(installed.returncode, 0, installed.stderr) self.entrypoint = self.npm_prefix / "lib/node_modules/prime-agent/dist/bundle/cli.js"
def launch(self, *args, **environment): return subprocess.run(["node", str(self.entrypoint), *args], env={**self.env, **environment}, capture_output=True, text=True, timeout=10)
def test_npm_install_does_not_migrate_until_old_entrypoint_relaunches(self): self.assertFalse(self.installs.exists()) args = ("--internal-update-restart-coordinator", "a space", "line\nbreak", "--", "-x") result = self.launch(*args, NATIVE_EXIT="37", EXPECT_DEFER="1") self.assertEqual(result.returncode, 37, result.stderr) self.assertEqual(json.loads(result.stdout), list(args)) self.assertIn("installer output stays on stderr", result.stderr) self.assertTrue(self.entrypoint.exists(), "migration must preserve the active npm package")
def test_later_ts_node_fallback_entrypoint_survives_npm_replacement(self): # v0.9.5+ cli.js shims spawn cli-node.js. Their update coordinator # retains that absolute filename after npm replaces the package. self.entrypoint = self.entrypoint.with_name("cli-node.js") args = ("update", "--internal-update-restart-coordinator", "a space", "--", "-x") result = self.launch(*args, NATIVE_EXIT="37", EXPECT_DEFER="1") self.assertEqual(result.returncode, 37, result.stderr) self.assertEqual(json.loads(result.stdout), list(args)) self.assertTrue(self.entrypoint.exists())
def test_earliest_ts_entrypoint_resolves_package_root_after_replacement(self): # Early published packages capture dist/cli.js. Its compatibility shim # must resolve the bundled installer from the actual package root. self.entrypoint = self.entrypoint.parents[1] / "cli.js" args = ("update", "--internal-update-restart-coordinator", "a space", "--", "-x") result = self.launch(*args, NATIVE_EXIT="37", EXPECT_DEFER="1") self.assertEqual(result.returncode, 37, result.stderr) self.assertEqual(json.loads(result.stdout), list(args)) self.assertTrue(self.entrypoint.exists())
def test_native_self_update_does_not_trigger_pinned_downgrade(self): self.assertEqual(self.launch("first").returncode, 0) # A subsequent native update replaces its payload, but the npm shim # must keep forwarding to it without reinstalling the old pinned release. payload = self.prefix / "share/prime-agent" shutil.rmtree(payload) payload.mkdir() shutil.copy2(self.fixture, payload / "prime-agent") result = self.launch("after-native-update") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(self.installs.read_text(), "install\n")
def test_completed_migration_launch_does_not_wait_for_migration_lock(self): self.assertEqual(self.launch("first").returncode, 0) lock = self.prefix / "share/.prime-agent-npm-bridge/migration.lock" lock.write_text(json.dumps({"pid": os.getpid(), "token": "another-install"})) result = self.launch("already-installed") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(result.stdout), ["already-installed"]) self.assertEqual(self.installs.read_text(), "install\n") self.assertTrue(lock.exists(), "ordinary launch must not modify another install's lock")
def test_new_npm_package_version_does_not_reuse_previous_migration_receipt(self): self.assertEqual(self.launch("first").returncode, 0) package_path = self.entrypoint.parents[2] / "package.json" package = json.loads(package_path.read_text()) package["version"] = "1.1.0" package_path.write_text(json.dumps(package)) result = self.launch("upgraded-package", EXPECT_VERSION="1.1.0") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(result.stdout), ["upgraded-package"]) self.assertEqual(self.installs.read_text(), "install\ninstall\n")
def test_failed_install_does_not_activate_or_record_success(self): result = self.launch("update", FAIL_INSTALL="1") self.assertEqual(result.returncode, 23, result.stderr) self.assertEqual(result.stdout, "") self.assertFalse((self.prefix / "share/.prime-agent-npm-bridge/1.0.0").exists()) self.assertEqual(self.launch("update").returncode, 0) self.assertEqual(self.installs.read_text(), "install\ninstall\n")
def test_failed_native_install_keeps_working_typescript_without_repeated_attempts(self): result = self.launch("a prompt", FAIL_INSTALL="1") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(result.stdout), {"fallback": True, "args": ["a prompt"]}) self.assertIn("Rust 1.0.0 migration failed", result.stderr) self.assertIn("Continuing with TypeScript 0.9.8", result.stderr) again = self.launch("--version", FAIL_INSTALL="1") self.assertEqual(again.returncode, 0, again.stderr) self.assertEqual(again.stdout.strip(), "0.9.8") self.assertEqual(self.installs.read_text(), "install\n") retry = self.launch("update", FAIL_INSTALL="1") self.assertNotEqual(retry.returncode, 0, retry.stderr) self.assertEqual(retry.stdout, "", "failed update must not recurse through the TS updater") self.assertEqual(self.installs.read_text(), "install\ninstall\n")
def test_explicit_update_after_fallback_fetches_new_bridge_and_installer(self): self.assertEqual(self.launch("--version", FAIL_INSTALL="1").stdout.strip(), "0.9.8") future = assemble(self.repo, self.root / "future", "1.1.0", "stable", self.fallback_tarball) result = self.launch("--offline", "--daemon-socket", str(self.root / "socket"), "update", FUTURE_BRIDGE_TARBALL=str(self.root / "future" / future["tarball"]), EXPECT_VERSION="1.1.0") self.assertEqual(result.returncode, 0, result.stderr) package = json.loads((self.entrypoint.parents[2] / "package.json").read_text()) self.assertEqual(package["version"], "1.1.0") self.assertTrue((self.prefix / "share/.prime-agent-npm-bridge/1.1.0").exists()) self.assertEqual(self.launch("--version").stdout.strip(), "1.1.0")
def test_new_bridge_that_still_cannot_install_rust_is_not_reported_as_success(self): self.assertEqual(self.launch("--version", FAIL_INSTALL="1").stdout.strip(), "0.9.8") future = assemble(self.repo, self.root / "future", "1.1.0", "stable", self.fallback_tarball) result = self.launch("update", FUTURE_BRIDGE_TARBALL=str(self.root / "future" / future["tarball"]), EXPECT_VERSION="1.1.0", FAIL_INSTALL="1") self.assertNotEqual(result.returncode, 0, result.stderr) self.assertIn("did not activate a working Rust release", result.stderr) self.assertEqual(self.launch("--version").stdout.strip(), "0.9.8")
def test_update_after_global_options_fails_honestly_but_update_prompt_can_fallback(self): result = self.launch("--offline", "--daemon-socket", str(self.root / "socket"), "update", FAIL_INSTALL="1") self.assertEqual(result.returncode, 23, result.stderr) self.assertEqual(result.stdout, "") prompt = self.launch("--print", "update") self.assertEqual(prompt.returncode, 0, prompt.stderr) self.assertEqual(json.loads(prompt.stdout), {"fallback": True, "args": ["--print", "update"]})
def test_coordinator_failure_remains_failed_while_later_commands_use_typescript(self): status = self.root / "agent/update-restarts/failed.json" result = self.launch("update", "--internal-update-restart-coordinator", "--internal-update-restart-status", str(status), FAIL_INSTALL="1", EXPECT_DEFER="1") self.assertEqual(result.returncode, 23, result.stderr) self.assertEqual(json.loads(status.read_text())["phase"], "failed") self.assertEqual(result.stdout, "") self.assertEqual(self.launch("--version").stdout.strip(), "0.9.8")
def test_fallback_self_update_captures_public_bridge_entrypoint(self): result = self.launch("entrypoint", FAIL_INSTALL="1") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.strip(), str(self.entrypoint)) self.assertEqual(self.launch("prefix").stdout.strip(), str(self.npm_prefix))
def test_fallback_preserves_inherited_worker_descriptor(self): channel = self.root / "worker-channel" channel.write_text("worker ready") descriptor = os.open(channel, os.O_RDONLY) try: result = subprocess.run(["node", str(self.entrypoint), "descriptor"], env={**self.env, "FAIL_INSTALL": "1", "TEST_DESCRIPTOR": str(descriptor)}, pass_fds=(descriptor,), capture_output=True, text=True, timeout=10) finally: os.close(descriptor) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.strip(), "worker ready")
def test_fallback_preserves_node_catalog_ipc_channel(self): driver = '''const {fork}=require("node:child_process");const child=fork(process.argv[1],["ipc"],{execArgv:[],stdio:["ignore","pipe","pipe","ipc"]});child.stderr.pipe(process.stderr);child.on("message",value=>console.log(JSON.stringify(value)));child.on("exit",code=>{process.exitCode=code;});''' result = subprocess.run(["node", "-e", driver, str(self.entrypoint)], env={**self.env, "FAIL_INSTALL": "1"}, capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(result.stdout), {"fallback": True})
def test_native_signal_status_is_preserved(self): result = self.launch("signal") self.assertEqual(result.returncode, -signal.SIGTERM, result.stderr)
def test_signal_to_npm_launcher_reaches_native_child(self): with subprocess.Popen(["node", str(self.entrypoint), "wait"], env=self.env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) as child: self.assertEqual(child.stdout.readline(), "ready\n") child.send_signal(signal.SIGTERM) _, stderr = child.communicate(timeout=10) self.assertEqual(child.returncode, -signal.SIGTERM, stderr)
def test_missing_native_install_is_repaired(self): self.assertEqual(self.launch().returncode, 0) (self.prefix / "share/prime-agent/prime-agent").unlink() self.assertEqual(self.launch().returncode, 0) self.assertEqual(self.installs.read_text(), "install\ninstall\n")
def test_npm_and_native_prefix_overlap_does_not_recurse(self): self.prefix = self.npm_prefix self.env["PRIME_AGENT_RUST_PREFIX"] = str(self.prefix) self.assertEqual(self.launch("first").returncode, 0) # npm owns this link and may replace it during npm install at any time. command = self.prefix / "bin/prime-agent" self.assertEqual(command.resolve(), self.entrypoint.resolve()) result = subprocess.run([str(command), "after-npm-reinstall"], env=self.env, capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(result.stdout), ["after-npm-reinstall"]) self.assertEqual(self.installs.read_text(), "install\n")
def test_later_launch_keeps_migrated_daemon_socket_namespace(self): self.env.pop("PRIME_AGENT_DAEMON_SOCKET", None) self.env["TMPDIR"] = str(self.root / "sockets") self.assertEqual(self.launch("--internal-update-restart-coordinator", EXPECT_DEFER="1").returncode, 0) expected = str(self.root / "sockets" / f"prime-agent-{os.getuid()}" / "daemon.sock") result = self.launch("socket") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.strip(), expected) custom = self.launch("socket", PRIME_AGENT_DAEMON_SOCKET=str(self.root / "custom.sock")) self.assertEqual(custom.stdout.strip(), str(self.root / "custom.sock"))
def install_real_launcher(self, **environment): # Reuse the real published launcher section, not the fixture # installer: an old TUI captures the public symlink path in Node's # argv before npm replaces the package, and the launcher this # section writes is the file users run after the takeover. installer = (Path(__file__).resolve().parents[2] / "install-rust.sh").read_text() section = installer.split("# --- the launcher (the takeover lives here)", 1)[1] section = section.split("# The cmd/PowerShell launcher twin", 1)[0] section = "# --- the launcher (the takeover lives here)" + section self.env["PRIME_AGENT_RUST_PREFIX"] = str(self.npm_prefix) payload = self.npm_prefix / "share/prime-agent" payload.mkdir(parents=True, exist_ok=True) shutil.copyfile(self.fixture, payload / "prime-agent") (payload / "prime-agent").chmod(0o755) (payload / "version").write_text("1.0.0\n") return subprocess.run(["sh", "-eu"], input='''PREFIX="$PRIME_AGENT_RUST_PREFIX"bin_dir="$PREFIX/bin"launcher="$bin_dir/prime-agent"WINDOWS=nosay() { :; }; note() { :; }; die() { echo "$*" >&2; exit 1; }trap '[ -z "${launcher_tmp:-}" ] || rm -f "$launcher_tmp"' EXIT''' + section + '\n"$install_probe_launcher" --version\n', text=True, capture_output=True, timeout=10, env={**self.env, **environment})
def test_real_installer_preserves_node_relaunch_only_for_its_npm_bridge(self): receipt_dir = self.npm_prefix / "share/.prime-agent-npm-bridge" receipt_dir.mkdir(parents=True) (receipt_dir / "1.0.0").write_text("installed\n") command = self.npm_prefix / "bin/prime-agent" foreign_entry = self.root / "unrelated.js" foreign_entry.write_text(self.entrypoint.read_text()) for flag, target, preserved in (("1", self.entrypoint, True), ("1", foreign_entry, False), ("0", self.entrypoint, False)): with self.subTest(flag=flag, target=target): command.unlink(missing_ok=True) command.symlink_to(target) result = self.install_real_launcher( PRIME_AGENT_PRESERVE_NPM_BRIDGE=flag, PRIME_AGENT_NPM_BRIDGE_ENTRYPOINT=str(self.entrypoint), PRIME_AGENT_USE_LEGACY_DAEMON_SOCKET="1") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.strip(), "1.0.0") self.assertEqual(command.is_symlink(), preserved) invocation = ["node", str(command)] if preserved else [str(command)] relaunched = subprocess.run([*invocation, "--version"], env=self.env, text=True, capture_output=True, timeout=10) self.assertEqual(relaunched.returncode, 0, relaunched.stderr) self.assertEqual(relaunched.stdout.strip(), "1.0.0") self.assertEqual(list(command.parent.glob(".prime-agent.*")), [])
def test_real_launcher_pin_respects_the_sun_path_limit(self): # A TMPDIR whose longest worker endpoint cannot fit the portable # 103-byte sun_path limit must pin a hashed /tmp directory, never the # product default. The rule no longer varies by platform. command = self.npm_prefix / "bin/prime-agent" result = self.install_real_launcher() self.assertEqual(result.returncode, 0, result.stderr) shim_dir = self.root / "uname-shim" shim_dir.mkdir() long_tmp = self.root / ("t" * 80) expected = rust_socket(long_tmp) self.assertTrue(expected.startswith("/tmp/prime-agent-rust-")) for system in ("Darwin", "Linux"): with self.subTest(system=system): (shim_dir / "uname").write_text(f'#!/bin/sh\necho {system}\n') (shim_dir / "uname").chmod(0o755) probe_env = {**self.env, "TMPDIR": str(long_tmp), "PATH": str(shim_dir) + os.pathsep + self.env["PATH"]} probe_env.pop("PRIME_AGENT_DAEMON_SOCKET", None) probe = subprocess.run([str(command), "socket"], env=probe_env, text=True, capture_output=True, timeout=10) self.assertEqual(probe.returncode, 0, probe.stderr) self.assertEqual(probe.stdout.strip(), expected)
def test_utf8_socket_limit_applies_to_launcher_and_stop_probe(self): result = self.install_real_launcher() self.assertEqual(result.returncode, 0, result.stderr) command = self.npm_prefix / "bin/prime-agent" installer = (Path(__file__).resolve().parents[2] / "install-rust.sh").read_text() stop_probe = installer.split('rust_socket="${TMPDIR:-/tmp}', 1)[1] stop_probe = 'rust_socket="${TMPDIR:-/tmp}' + stop_probe.split('env_socket=', 1)[0] shim_dir = self.root / "utf8-uname-shim" shim_dir.mkdir() tmp = "/tmp/" + "é" * 40 pin = f"{tmp}/prime-agent-rust-{os.getuid()}/daemon.sock" self.assertLessEqual(len(pin), 103) self.assertGreater(len(pin.encode()), 103) for system in ("Darwin", "Linux"): with self.subTest(system=system): (shim_dir / "uname").write_text(f"#!/bin/sh\necho {system}\n") (shim_dir / "uname").chmod(0o755) env = {**self.env, "OS": system, "TMPDIR": tmp, "LC_ALL": "en_US.UTF-8", "PATH": str(shim_dir) + os.pathsep + self.env["PATH"]} env.pop("PRIME_AGENT_DAEMON_SOCKET", None) expected = rust_socket(tmp) for shell in ("sh", "bash"): with self.subTest(shell=shell): probe = subprocess.run([shell, str(command), "socket"], env=env, text=True, capture_output=True, timeout=10) self.assertEqual(probe.returncode, 0, probe.stderr) self.assertEqual(probe.stdout.strip(), expected) stop = subprocess.run([shell, "-eu"], env=env, input=stop_probe + '\nprintf "%s" "$rust_socket"\n', text=True, capture_output=True, timeout=10) self.assertEqual(stop.returncode, 0, stop.stderr) self.assertEqual(stop.stdout, expected)
def test_stale_bridge_does_not_downgrade_newer_native_without_receipt(self): payload = self.prefix / "share/prime-agent" payload.mkdir(parents=True) executable = payload / "prime-agent" executable.write_text('''#!/usr/bin/env nodeif (process.argv[2] === "--version") console.log("prime-agent 1.2.0");else if (process.argv[2] === "socket") console.log(process.env.PRIME_AGENT_DAEMON_SOCKET);else console.log(JSON.stringify(process.argv.slice(2)));''') executable.chmod(0o755) result = self.launch("keep-newer") self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(self.installs.exists()) self.assertEqual(json.loads(result.stdout), ["keep-newer"]) self.env.pop("PRIME_AGENT_DAEMON_SOCKET", None) self.env["TMPDIR"] = str(self.root / "sockets") expected = str(self.root / "sockets" / f"prime-agent-rust-{os.getuid()}" / "daemon.sock") self.assertEqual(self.launch("socket").stdout.strip(), expected) # A TMPDIR deeper than the portable 103-byte worker limit pins a # hashed /tmp directory instead. long_tmp = str(self.root / ("t" * 80)) self.env["TMPDIR"] = long_tmp expected = rust_socket(long_tmp) self.assertEqual(self.launch("socket").stdout.strip(), expected)
def test_concurrent_first_launches_install_once(self): processes = [subprocess.Popen(["node", str(self.entrypoint), str(i)], env=self.env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) for i in range(4)] try: for i, child in enumerate(processes): stdout, stderr = child.communicate(timeout=10) self.assertEqual(child.returncode, 0, stderr) self.assertEqual(json.loads(stdout), [str(i)]) finally: for child in processes: if child.poll() is None: child.kill() child.wait() self.assertEqual(self.installs.read_text(), "install\n")
def test_coordinator_status_exists_before_install_and_defers_daemon_stop(self): status = self.root / "agent/update-restarts/status.json" installer = self.repo / "install-rust.sh" installer.write_text(installer.read_text() + '\ntest -s "$TEST_RESTART_STATUS"\n') assemble(self.repo, self.out, "1.0.0", "stable", self.fallback_tarball) # Replace the installed fixture, preserving the real npm entrypoint. shutil.copy2(installer, self.entrypoint.parents[2] / "install-rust.sh") args = ["update", "--internal-update-restart-coordinator", "--internal-update-restart-status", str(status), "--daemon-socket", str(self.root / "old.sock")] result = self.launch(*args, EXPECT_DEFER="1", TEST_RESTART_STATUS=str(status)) self.assertEqual(result.returncode, 0, result.stderr) value = json.loads(status.read_text()) self.assertEqual(value["phase"], "starting") self.assertEqual(value["socketPath"], str(self.root / "old.sock")) self.assertEqual(json.loads(result.stdout), args)
def test_archive_is_deterministic_and_has_no_lifecycle_scripts(self): other = assemble(self.repo, self.root / "other", "1.0.0", "stable", self.fallback_tarball) self.assertEqual(other, self.metadata) with tarfile.open(self.out / self.metadata["tarball"]) as archive: package = json.load(archive.extractfile("package/package.json")) self.assertNotIn("scripts", package) self.assertTrue(package["primeAgentRustBridge"])
def test_real_installer_keeps_bridge_but_retires_ts_package(self): installer = (Path(__file__).resolve().parents[2] / "install-rust.sh").read_text() retirement = installer.split("# The TS npm package:", 1)[1].split( "# --- the kernel pre-warm:", 1)[0] retirement = "# The TS npm package:" + retirement fake_bin = self.root / "fake-bin" fake_bin.mkdir() npm_root = self.root / "npm-root" package_dir = npm_root / "prime-agent" package_dir.mkdir(parents=True) uninstalls = self.root / "uninstalls" fake_npm = fake_bin / "npm" fake_npm.write_text('''#!/bin/shif [ "$1" = root ]; then echo "$TEST_NPM_ROOT"; else echo uninstall >> "$TEST_UNINSTALL_LOG"; fi''') fake_npm.chmod(0o755) for flag, bridge, expected in (("1", False, False), ("", True, False), ("", False, True)): with self.subTest(flag=flag, bridge=bridge): uninstalls.unlink(missing_ok=True) (package_dir / "package.json").write_text(json.dumps({ "name": "prime-agent", "version": "1.0.0" if bridge else "0.9.8", "primeAgentRustBridge": bridge})) result = subprocess.run(["sh", "-eu"], input='''WINDOWS=nobin_dir="$TEST_NPM_ROOT"launcher="$bin_dir/launcher"printf '#!/bin/sh\\nexit 0\\n' > "$launcher"chmod 0755 "$launcher"step_start() { :; }; step_ok() { :; }; step_fail() { :; }; note() { :; }; todo() { :; }''' + retirement, text=True, capture_output=True, env={**self.env, "PATH": str(fake_bin) + os.pathsep + self.env["PATH"], "UVPY": sys.executable, "TEST_NPM_ROOT": str(npm_root), "TEST_UNINSTALL_LOG": str(uninstalls), "PRIME_AGENT_PRESERVE_NPM_BRIDGE": flag}) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(uninstalls.exists(), expected)
@unittest.skipUnless(sys.platform in ("darwin", "linux") and shutil.which("npm"), "requires npm on a historical TypeScript platform")class InstallerTakeoverTests(unittest.TestCase): def test_npm_retirement_leaves_working_native_launcher(self): installer = (Path(__file__).resolve().parents[2] / "install-rust.sh").read_text() launcher_section = installer.split("# --- the launcher (the takeover lives here)", 1)[1].split( "# The cmd/PowerShell launcher twin", 1)[0] retirement = installer.split("# The TS npm package:", 1)[1].split( "# --- the kernel pre-warm:", 1)[0] recovery = installer.split("on_exit() {", 1)[1].split(' [ -n "$launcher_tmp" ]', 1)[0] for shared, uninstall_fails in ((True, 0), (False, 0), (True, 1), (True, 2)): with self.subTest(shared=shared, uninstall_fails=uninstall_fails), tempfile.TemporaryDirectory() as temp: root = Path(temp) prefix = root / "native prefix" npm_prefix = prefix if shared else root / "npm prefix" package = npm_prefix / "lib/node_modules/prime-agent" package.mkdir(parents=True) (package / "package.json").write_text(json.dumps({ "name": "prime-agent", "version": "0.9.8", "bin": {"prime-agent": "cli.js"}})) (package / "cli.js").write_text("#!/usr/bin/env node\nconsole.log('0.9.8');\n") (package / "cli.js").chmod(0o755) (npm_prefix / "bin").mkdir() (npm_prefix / "bin/prime-agent").symlink_to("../lib/node_modules/prime-agent/cli.js") (prefix / "bin").mkdir(parents=True, exist_ok=True) payload = prefix / "share/prime-agent" payload.mkdir(parents=True) (payload / "prime-agent").write_text("#!/bin/sh\nprintf '0.10.0\\n'\n") (payload / "prime-agent").chmod(0o755) env = {"PATH": os.environ["PATH"], "HOME": temp, "PREFIX": str(prefix), "UVPY": sys.executable, "TEST_UNINSTALL_FAIL": str(uninstall_fails), "npm_config_prefix": str(npm_prefix), "npm_config_cache": str(root / "cache"), "npm_config_userconfig": str(root / "empty-user-npmrc"), "npm_config_globalconfig": str(root / "empty-global-npmrc")} result = subprocess.run(["sh", "-eu"], input='''bin_dir="$PREFIX/bin"launcher="$bin_dir/prime-agent"WINDOWS=nosay() { :; }; note() { :; }; die() { echo "$*" >&2; exit 1; }step_start() { :; }; step_ok() { :; }; step_fail() { :; }; todo() { :; }ui_stop() { :; }; publish_window_restore() { :; }''' + 'recover() {\n' + recovery + '\n}\n' + '''trap 'recover; [ -z "${launcher_tmp:-}" ] || rm -f "$launcher_tmp"' EXITtrap 'exit 143' TERMnpm() { command npm "$@" || return $? if [ "$1" = uninstall ]; then case "$TEST_UNINSTALL_FAIL" in 1) return 42 ;; 2) kill -TERM "$$" ;; esac fi}''' + '# --- the launcher (the takeover lives here)' + launcher_section + '# The TS npm package:' + retirement + '\n"$launcher" --version\n', env=env, text=True, capture_output=True, timeout=30) self.assertFalse(package.exists(), result.stderr) self.assertEqual(result.returncode, 143 if uninstall_fails == 2 else 0, result.stderr) probe = subprocess.run([str(prefix / "bin/prime-agent"), "--version"], env=env, text=True, capture_output=True, timeout=10) self.assertEqual((probe.returncode, probe.stdout.strip()), (0, "0.10.0"), probe.stderr) self.assertEqual(list((prefix / "bin").glob(".prime-agent.*")), [])
def test_failed_version_probe_is_install_failure(self): installer = (Path(__file__).resolve().parents[2] / "install-rust.sh").read_text() verification = installer.split("# --- verify: the launcher must answer --version", 1)[1].split("\n", 1)[1].split("# --- the summary", 1)[0] result = subprocess.run(["sh", "-eu"], input='''install_probe_launcher=/nonexistent/prime-agentsay() { :; }; die() { echo "$*" >&2; exit 1; }''' + verification, text=True, capture_output=True) self.assertNotEqual(result.returncode, 0, "a broken launcher must not certify installation")
if __name__ == "__main__": unittest.main()