Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246#!/usr/bin/env python3"""Exercise a shipped TS daemon's post-install coordinator against a Rust candidate.
This launches no model inference. All homes, sockets, and daemon ownershipregistries are temporary. Run in a disposable Prime sandbox."""import argparseimport jsonimport osfrom pathlib import Pathimport signalimport shutilimport socketimport subprocessimport tarfileimport tempfileimport threadingimport time
def connect(path): stream = socket.socket(socket.AF_UNIX) stream.settimeout(30) try: stream.connect(str(path)) reader = stream.makefile("r") hello = json.loads(reader.readline()) return stream, reader, hello except BaseException: stream.close() raise
def request(path, command): stream, reader, hello = connect(path) try: stream.sendall((json.dumps({"type": "command", "id": "probe", "protocol": hello["protocol"], "command": command}) + "\n").encode()) while line := reader.readline(): frame = json.loads(line) if frame.get("type") == "response" and frame.get("id") == "probe": if not frame.get("success"): raise RuntimeError(frame) return frame.get("data") raise RuntimeError("daemon disconnected before response") finally: reader.close() stream.close()
def await_hello(path, child): deadline = time.monotonic() + 30 while True: try: stream, reader, hello = connect(path) reader.close() stream.close() return hello except (OSError, ValueError): if child.poll() is not None or time.monotonic() >= deadline: raise RuntimeError("old daemon did not become ready") time.sleep(0.05)
def stop_fixture_processes(home): """Stop only Linux processes that carry this fixture's unique HOME.""" if not Path("/proc").is_dir(): return expected = f"HOME={home}".encode() deadline = time.monotonic() + 10 while True: matched = [] for entry in Path("/proc").iterdir(): if not entry.name.isdigit() or int(entry.name) == os.getpid(): continue try: if expected in (entry / "environ").read_bytes().split(b"\0"): matched.append(entry) except (FileNotFoundError, ProcessLookupError, PermissionError): continue if not matched: return for entry in matched: try: # Recheck after enumeration so a reused PID cannot be signalled. if expected in (entry / "environ").read_bytes().split(b"\0"): os.kill(int(entry.name), signal.SIGKILL if time.monotonic() > deadline - 5 else signal.SIGTERM) except (FileNotFoundError, ProcessLookupError, PermissionError): pass if time.monotonic() >= deadline: raise RuntimeError("fixture processes did not terminate") time.sleep(0.05)
def main(): parser = argparse.ArgumentParser(description=__doc__) legacy_source = parser.add_mutually_exclusive_group(required=True) legacy_source.add_argument("--legacy-archive", type=Path) legacy_source.add_argument("--legacy-npm-prefix", type=Path) legacy_source.add_argument("--legacy-binary", type=Path, help="invoke an installed legacy executable or wrapper directly") parser.add_argument("--node", default="node") parser.add_argument("--candidate-archive", type=Path) parser.add_argument("--candidate-binary", type=Path) parser.add_argument("--with-session", action="store_true") parser.add_argument("--report", type=Path, required=True) options = parser.parse_args() if bool(options.candidate_archive) == bool(options.candidate_binary): parser.error("choose one candidate archive or candidate binary") evidence = {} with tempfile.TemporaryDirectory(prefix="legacy-runtime-") as directory: root = Path(directory) old_dir, candidate_dir = root / "ts", root / "rust" if options.legacy_archive: old_dir.mkdir() with tarfile.open(options.legacy_archive) as archive: archive.extractall(old_dir, filter="data") old_command = [str(old_dir / "prime-agent")] elif options.legacy_npm_prefix: shutil.copytree(options.legacy_npm_prefix, old_dir, symlinks=True) old_command = [options.node, str(old_dir / "lib/node_modules/prime-agent/dist/bundle/cli.js")] else: # Preserve installed wrappers, including their runtime environment. old_command = [str(options.legacy_binary.resolve())] if options.candidate_archive: candidate_dir.mkdir() with tarfile.open(options.candidate_archive) as archive: archive.extractall(candidate_dir, filter="data") candidate = candidate_dir / "prime-agent" else: candidate = options.candidate_binary.resolve() home, agent, temporary = root / "home", root / "agent", root / "tmp" for path in (home, agent / "update-restarts", temporary): path.mkdir(parents=True) endpoint = temporary / "daemon.sock" env = dict(os.environ) for key in list(env): if key.startswith(("PI_", "PRIME_AGENT_")): del env[key] env.update(HOME=str(home), TMPDIR=str(temporary), DO_NOT_TRACK="1", XDG_CONFIG_HOME=str(home / ".config"), XDG_DATA_HOME=str(home / ".local/share"), XDG_CACHE_HOME=str(home / ".cache"), PRIME_AGENT_CODING_AGENT_DIR=str(agent), PRIME_AGENT_OFFLINE="1", PRIME_AGENT_INTERNAL_DAEMON_SUPERVISOR_REGISTRY_DIR=str(root / "owners")) with (root / "old.log").open("w") as log: old = subprocess.Popen(old_command + ["--mode", "daemon", "--daemon-socket", str(endpoint)], env=env, cwd=root, stdout=log, stderr=subprocess.STDOUT) threading.Thread(target=old.wait, daemon=True).start() successor_pid = None try: evidence["predecessor"] = await_hello(endpoint, old) transcript = None if options.with_session: # Seed an existing transcript: TS deliberately buffers a brand # new session until its first assistant message exists. transcript = agent / "sessions" / "migration.jsonl" transcript.parent.mkdir(exist_ok=True) stamp = "2026-10-09T00:00:00.000Z" rows = [ {"type":"session","version":3,"id":"11111111-2222-4333-8444-555555555555","timestamp":stamp,"cwd":str(root)}, {"type":"message","id":"seed-assistant","parentId":None,"timestamp":stamp,"message":{ "role":"assistant","content":[{"type":"text","text":"durable migration sentinel"}], "api":"openai-completions","provider":"openai","model":"gpt-4o-mini", "usage":{"input":0,"output":0,"cacheRead":0,"cacheWrite":0,"totalTokens":0, "cost":{"input":0,"output":0,"cacheRead":0,"cacheWrite":0,"total":0}}, "stopReason":"stop","timestamp":1791504000000}}, ] transcript.write_text("".join(json.dumps(row)+"\n" for row in rows)) created = request(endpoint, {"type": "create", "sessionPath":str(transcript), "config": { "cwd": str(root), "agentDir": str(agent), "provider": "openai", "model": "gpt-4o-mini", "noExtensions": True, "noSkills": True, "noContextFiles": True, "telemetryDisabled": True}}) evidence["created_session"] = created request(endpoint, {"type": "append_custom_message", "activeSessionId": created["activeSessionId"], "message": {"customType": "migration-test", "content": "live TS checkpoint sentinel", "display": True, "timestamp": int(time.time() * 1000)}}) transcript = Path(created["sessionFile"]) deadline = time.monotonic() + 10 while True: before = transcript.read_text() if "live TS checkpoint sentinel" in before: break if time.monotonic() >= deadline: raise RuntimeError("TS session did not persist the custom message") time.sleep(0.05) status_path = agent / "update-restarts" / "coordinator.json" result = subprocess.run([str(candidate), "update", "--internal-update-restart-coordinator", "--daemon-socket", str(endpoint), "--internal-update-restart-status", str(status_path)], env=env, cwd=root, capture_output=True, text=True, timeout=180) evidence.update(exit_code=result.returncode, stdout=result.stdout, stderr=result.stderr, status=json.loads(status_path.read_text())) stream, reader, hello = connect(endpoint) reader.close() stream.close() evidence["successor"] = hello successor_pid = hello["supervisorPid"] assert result.returncode == 0, evidence assert evidence["status"]["phase"] == "complete", evidence assert successor_pid != old.pid if options.with_session: assert evidence["status"]["counts"]["restored"] == 1, evidence assert evidence["status"]["counts"]["failed"] == 0, evidence assert transcript.read_text().startswith(before), "old transcript must remain intact" evidence["transcript_preserved"] = True evidence["restored_sessions"] = request(endpoint, {"type": "list"}) restored = evidence["restored_sessions"]["sessions"] assert len(restored) == 1, restored assert restored[0]["sessionId"] == created["sessionId"], restored assert restored[0]["sessionFile"] == str(transcript), restored witness = "restored-session-bash-witness" evidence["restored_bash"] = request(endpoint, { "type": "execute_bash_and_wait", "activeSessionId": restored[0]["activeSessionId"], "command": f"printf '%s' '{witness}'"}) bash = evidence["restored_bash"] assert bash["output"] == witness and bash["exitCode"] == 0, bash assert not bash["cancelled"] and not bash["truncated"], bash deadline = time.monotonic() + 10 while True: after = transcript.read_text() rows = [json.loads(line) for line in after.splitlines()] if any(row.get("message", {}).get("role") == "bashExecution" and row["message"].get("output") == witness for row in rows): assert after.startswith(before), "bash execution must preserve old transcript" evidence["restored_bash_persisted"] = True break if time.monotonic() >= deadline: raise RuntimeError("restored worker did not persist the bash result") time.sleep(0.05) evidence["success"] = True finally: evidence["old_log"] = (root / "old.log").read_text() options.report.parent.mkdir(parents=True, exist_ok=True) options.report.write_text(json.dumps(evidence, indent=2) + "\n") try: request(endpoint, {"type": "shutdown", "force": True}) except (OSError, ValueError, RuntimeError): pass if old.poll() is None: old.terminate() old.wait(timeout=30) stop_fixture_processes(home) print(json.dumps(evidence, indent=2))
if __name__ == "__main__": main()