Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477#!/usr/bin/env python3"""Installed Prime Agent native-scrollback regression (Linux, stdlib only).
Usage: python3 nix/checks/transcript_pty.py /nix/store/.../bin/prime-agentNo live daemon, provider, credentials or host session files are used. A localJSONL peer injects deterministic restore/reconnect/branch faults. Every assertionabout native history inspects bytes emitted by the installed CLI on a REAL PTY,strictly between alternate-screen leave and reentry, not a headless render seam.Logs remain in a private scratch directory on success and failure."""import argparseimport errnoimport fcntlimport hashlibimport jsonimport osfrom pathlib import Pathimport ptyimport reimport selectorsimport signalimport socketimport structimport sysimport tempfileimport termiosimport timeimport traceback
SCHEMA = "protocol-7-schema-30-8e4b17c2a9f5"COUNT = 1205OMITTED = 1200CSI = re.compile(rb"\x1b\[[0-?]*[ -/]*[@-~]")OSC = re.compile(rb"\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)")TOKEN = re.compile(rb"(?:restore|selected)-\d{4}-end")
def plain(data): return CSI.sub(b"", OSC.sub(b"", bytes(data)))
def history(prefix): return [{"role": "user", "content": f"{prefix}-{i:04}-end", "timestamp": i + 1} for i in range(COUNT)]
def check_history(data, prefix): # Main-screen wire bytes are the actual native scrollback, not viewport rows. tokens = TOKEN.findall(plain(data)) expected = [f"{prefix}-{i:04}-end".encode() for i in range(COUNT)] assert tokens == expected, ( f"{prefix}: expected {COUNT} ordered unique native messages, got {len(tokens)}; " f"first={tokens[:3]!r}, last={tokens[-3:]!r}") assert b"\x1b[3J" not in data, "native scrollback was erased"
class TerminalBytes: """Split the real byte stream at terminal mode changes; retain all raw bytes.""" def __init__(self): self.raw = bytearray() self.pending = bytearray() self.alt = False self.main = bytearray() self.replays = [] self.enters = 0 self.leaves = 0 self.segment = bytearray()
def feed(self, chunk): self.raw.extend(chunk) self.pending.extend(chunk) replies = bytearray() while self.pending: if self.pending[0] != 27: end = self.pending.find(27) if end < 0: end = len(self.pending) text = self.pending[:end] if not self.alt: self.main.extend(text) self.segment.extend(text) del self.pending[:end] continue if len(self.pending) < 2: break if self.pending[1] == ord('['): match = CSI.match(self.pending) if match is None: break seq = bytes(match[0]) elif self.pending[1] == ord(']'): match = OSC.match(self.pending) if match is None: break seq = bytes(match[0]) else: seq = bytes(self.pending[:2]) del self.pending[:len(seq)] if seq == b"\x1b[?1049h": if self.segment: self.replays.append(bytes(self.segment)) self.segment.clear() self.alt = True self.enters += 1 elif seq == b"\x1b[?1049l": self.alt = False self.leaves += 1 else: if not self.alt: self.main.extend(seq) self.segment.extend(seq) if seq == b"\x1b[6n": replies.extend(b"\x1b[1;1R") elif seq == b"\x1b[?u": replies.extend(b"\x1b[?0u") elif seq in (b"\x1b[c", b"\x1b[0c"): replies.extend(b"\x1b[?1;2c") return replies
class Probe: def __init__(self, args, scratch): self.args, self.root = args, scratch self.deadline = time.monotonic() + args.timeout self.sel = selectors.DefaultSelector() self.term = TerminalBytes() self.clients = {} self.requests = [] self.active = None self.selected = False self.reject_selected = True self.generation = "g1" self.pid = None self.status = None self.master = None self.sockpath = scratch / "d.sock" self.listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self.listener.bind(str(self.sockpath)) self.listener.listen(16) self.listener.setblocking(False) self.sel.register(self.listener, selectors.EVENT_READ, "listen") self.log = (scratch / "protocol.jsonl").open("w", buffering=1) self.rawlog = (scratch / "terminal.bin").open("wb", buffering=0)
def send(self, client, data): self.log.write(json.dumps({"out": data}) + "\n") self.clients[client][1].extend(json.dumps(data).encode() + b"\n") self.sel.modify(client, selectors.EVENT_READ | selectors.EVENT_WRITE, "client")
def close_client(self, client): self.sel.unregister(client) client.close() del self.clients[client] if self.active is client: self.active = None
def state(self): return {"sessionId": "saved", "activeSessionId": "active", "leafId": "selected-leaf" if self.selected else "old-leaf", "compactionCount": 1, "cwd": str(self.root / "cwd"), "sessionFile": str(self.fixture), "isStreaming": False, "isCompacting": False}
def command(self, client, envelope): command = envelope["command"] kind = command["type"] self.requests.append(command) self.log.write(json.dumps({"in": envelope}) + "\n") state = self.state() messages = history("selected" if self.selected else "restore") data = {} success = True if kind == "create": data = {"id": "active", "activeSessionId": "active", "sessionId": "saved", "cwd": str(self.root / "cwd")} elif kind == "attach": self.active = client caps = command.get("capabilities", []) self.attach_caps = caps snapshot = {"activeSessionId": "active", "state": state, "summary": {"id": "active", "cwd": state["cwd"]}, "messages": messages[OMITTED:], "historyBefore": OMITTED, "displayTranscript": True, "lastEventSequence": 0, "lastEventCursor": {"generation": self.generation, "sequence": 0}} data = {"activeSessionId": "active", "snapshot": snapshot, "replay": {"status": "complete", "toSequence": 0}} elif kind == "get_connection_state": data = state elif kind == "get_messages": if self.selected and self.reject_selected and "before" not in command: self.reject_selected = False success = False else: data = {"messages": messages[:command["before"]] if "before" in command else messages, "displayTranscript": True, "sessionId": "saved", "leafId": state["leafId"], "generation": self.generation} if "historyLeafId" in command: data["historyLeafId"] = command["historyLeafId"] elif kind == "get_session_context": data = {"context": {"messages": messages[-1:]}} elif kind == "get_session_tree": nodes = [] for i, (entry_id, parent, content) in enumerate([ ("root", None, "shared root"), ("old-leaf", "root", "old branch"), ("selected-leaf", "root", "selected destination")]): nodes.append({"entry": {"type": "message", "id": entry_id, "parentId": parent, "timestamp": f"2026-01-01T00:00:0{i}Z", "message": {"role": "user", "content": content, "timestamp": i + 1}}}) data = {"leafId": state["leafId"], "flatNodes": nodes} elif kind == "navigate_tree": assert command["targetId"] == "selected-leaf", command self.selected = True data = {"editorText": ""} elif kind == "get_session_stats": data = {"sessionId": "saved", "sessionFile": str(self.fixture), "contextUsage": {"tokens": 1, "contextWindow": 200000}} elif kind == "get_rlm_children": data = {"children": [], "eventSequence": 0} elif kind == "list_heartbeats": data = {"heartbeats": []} elif kind == "get_commands": data = {"commands": []} elif kind == "reload": data = {"message": "Reloaded keybindings, skills, prompts, themes"} response = {"type": "response", "id": envelope["id"], "command": kind, "success": success, "data": data} if not success: response["error"] = "injected selected history failure" self.send(client, response)
def tick(self): remaining = self.deadline - time.monotonic() if remaining <= 0: raise TimeoutError("hard probe deadline exceeded") for key, mask in self.sel.select(remaining): obj = key.fileobj if key.data == "listen": client, _ = self.listener.accept() client.setblocking(False) self.clients[client] = [bytearray(), bytearray()] self.sel.register(client, selectors.EVENT_READ, "client") self.send(client, {"type": "daemon_hello", "socketPath": str(self.sockpath), "protocol": {"name": "prime-agent.daemon", "version": 7}, "schemaId": self.args.schema_id, "appVersion": "0.10.0", "clientId": "pty-probe", "serverCapabilities": ["display_transcript"]}) elif key.data == "pty": try: chunk = os.read(self.master, 65536) except OSError as error: if error.errno != errno.EIO: raise chunk = b"" if not chunk: self.sel.unregister(self.master) _, self.status = os.waitpid(self.pid, 0) continue self.rawlog.write(chunk) replies = self.term.feed(chunk) if replies: os.write(self.master, replies) else: if mask & selectors.EVENT_WRITE: output = self.clients[obj][1] if output: del output[:obj.send(output)] if not output: self.sel.modify(obj, selectors.EVENT_READ, "client") if mask & selectors.EVENT_READ: chunk = obj.recv(65536) if not chunk: self.close_client(obj) continue incoming = self.clients[obj][0] incoming.extend(chunk) while b"\n" in incoming: line, _, rest = incoming.partition(b"\n") incoming[:] = rest if line: self.command(obj, json.loads(line))
def wait(self, predicate, label): self.phase = label print(f"WAIT {label}", flush=True) while not predicate(): if self.status is not None: raise AssertionError(f"CLI exited early ({self.status}) during {label}") self.tick() print(f"OK {label}", flush=True)
def text_since(self, offset, needle): return needle.encode() in plain(self.term.raw[offset:])
def write(self, text): os.write(self.master, text)
def replay_count(self, prefix): return sum(f"{prefix}-0000-end".encode() in plain(s) for s in self.term.replays)
def launch(self): self.fixture = self.root / "saved.jsonl" self.fixture.write_text(json.dumps({"type": "session", "version": 3, "id": "saved", "timestamp": "2026-01-01T00:00:00Z", "cwd": str(self.root / "cwd")}) + "\n") self.fixture_hash = hashlib.sha256(self.fixture.read_bytes()).hexdigest() env = isolated_env(self.root) agent = Path(env["PRIME_AGENT_CODING_AGENT_DIR"]) (agent / "settings.json").write_text(json.dumps({"onboardingShown": True, "telemetryEnabled": False, "agentTraces": {"enabled": False}, "branchSummary": {"skipPrompt": True}})) pid, master = pty.fork() if pid == 0: os.chdir(self.root / "cwd") os.execve(self.args.binary, [self.args.binary, "--daemon-socket", str(self.sockpath), "--resume", str(self.fixture), "--provider", "openai", "--model", "gpt-4o", "--api-key", "fixture-not-a-credential"], env) self.pid, self.master = pid, master os.set_blocking(master, False) fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 100, 0, 0)) self.sel.register(master, selectors.EVENT_READ, "pty")
def run(self): self.launch() self.wait(lambda: self.replay_count("restore") == 1, "1205-message restore native replay + alt reentry") assert "display_transcript" in self.attach_caps, self.attach_caps assert any(r.get("before") == OMITTED for r in self.requests), "backfill boundary not exercised" check_history(next(s for s in self.term.replays if b"restore-0000-end" in plain(s)), "restore") offset = len(self.term.raw) self.write(b"\x0f") # Ctrl+O: repaint detail, not restore. fcntl.ioctl(self.master, termios.TIOCSWINSZ, struct.pack("HHHH", 25, 101, 0, 0)) os.kill(self.pid, signal.SIGWINCH) self.write(b"/reload\r") self.wait(lambda: self.text_since(offset, "Reloaded keybindings"), "redraw and ordinary reload") assert self.replay_count("restore") == 1, "ordinary redraw duplicated native history" # Force transport loss, but keep our bound listener. No free-port race. self.generation = "g2" assert self.active is not None self.close_client(self.active) self.wait(lambda: self.replay_count("restore") == 2, "reconnect authoritative native replay") restore = [s for s in self.term.replays if b"restore-0000-end" in plain(s)] check_history(restore[1], "restore") offset = len(self.term.raw) self.write(b"/tree\r") self.wait(lambda: self.text_since(offset, "Session Tree"), "tree picker") offset = len(self.term.raw) self.write(b"selected") # Wait for visible filter handling rather than sending Enter in the same paste burst. self.wait(lambda: self.text_since(offset, "selected"), "selected branch filter") offset = len(self.term.raw) self.write(b"\r") self.wait(lambda: self.selected, "branch navigate request") self.wait(lambda: self.text_since(offset, "Full session history unavailable"), "injected branch failure disclosed") assert self.text_since(offset, "selected-1204-end"), "selected context fallback not rendered" assert self.replay_count("selected") == 0, "incomplete branch replayed as complete history" offset = len(self.term.raw) self.write(b"/reload\r") self.wait(lambda: self.replay_count("selected") == 1, "selected history recovery + native replay") check_history(next(s for s in self.term.replays if b"selected-0000-end" in plain(s)), "selected") assert b"\x1b[3J" not in self.term.raw, "native scrollback erase" assert hashlib.sha256(self.fixture.read_bytes()).hexdigest() == self.fixture_hash, "fixture changed" self.write(b"\x04") self.wait(lambda: self.status is not None, "CLI clean exit") assert os.waitstatus_to_exitcode(self.status) == 0, self.status assert not self.term.alt, "alternate screen not restored on exit" (self.root / "result.json").write_text(json.dumps({"binary": self.args.binary, "sha256": hashlib.sha256(Path(self.args.binary).read_bytes()).hexdigest(), "schemaId": self.args.schema_id, "messages": COUNT, "restoreReplays": len(restore), "selectedReplays": self.replay_count("selected"), "fixtureUnchanged": True, "exitCode": 0}, indent=2) + "\n")
def close(self): if self.pid is not None: try: os.killpg(self.pid, signal.SIGKILL) except ProcessLookupError: pass if self.status is None: try: os.waitpid(self.pid, 0) except ChildProcessError: pass for client in list(self.clients): self.close_client(client) self.listener.close() if self.master is not None: os.close(self.master) self.sel.close() self.log.close() self.rawlog.close() self.sockpath.unlink(missing_ok=True)
def isolated_env(root): # Construct, never filter inherited env: no credentials, PRIME_AGENT_*, RLM_*, # PI_*, proxy, SSH agent, project hooks, user config, Python path or loader env. paths = {name: root / name for name in ["home", "config", "cache", "data", "state", "runtime", "tmp", "agent", "sessions", "kernel-venv", "cwd"]} for path in paths.values(): path.mkdir(mode=0o700, parents=True, exist_ok=True) return {"PATH": "/run/current-system/sw/bin:/usr/bin:/bin", "TERM": "xterm-256color", "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "TZ": "UTC", "DO_NOT_TRACK": "1", "HOME": str(paths["home"]), "XDG_CONFIG_HOME": str(paths["config"]), "XDG_CACHE_HOME": str(paths["cache"]), "XDG_DATA_HOME": str(paths["data"]), "XDG_STATE_HOME": str(paths["state"]), "XDG_RUNTIME_DIR": str(paths["runtime"]), "TMPDIR": str(paths["tmp"]), "PRIME_AGENT_CODING_AGENT_DIR": str(paths["agent"]), "PRIME_AGENT_AGENT_DIR": str(paths["agent"]), "PI_CODING_AGENT_DIR": str(paths["agent"]), "PRIME_AGENT_SESSION_DIR": str(paths["sessions"]), "PRIME_AGENT_CODING_AGENT_SESSION_DIR": str(paths["sessions"]), "PRIME_AGENT_KERNEL_VENV": str(paths["kernel-venv"]), "RLM_SESSION_DIR": str(paths["sessions"]), "PRIME_AGENT_INTERNAL_DAEMON_SUPERVISOR_REGISTRY_DIR": str(paths["runtime"]), "PRIME_AGENT_DAEMON_SOCKET": str(root / "d.sock")}
def self_test(): data = b"\x1b[?1049hviewport\x1b[?1049l" + b"\n".join( f"restore-{i:04}-end".encode() for i in range(COUNT)) + b"\x1b[?1049h" terminal = TerminalBytes() for byte in data: terminal.feed(bytes([byte])) assert terminal.alt and terminal.enters == 2 and terminal.leaves == 1 assert len(terminal.replays) == 1 check_history(terminal.replays[0], "restore") for mutant in [terminal.replays[0] + b"restore-0000-end", terminal.replays[0][30:], terminal.replays[0] + b"\x1b[3J"]: try: check_history(mutant, "restore") except AssertionError: continue raise AssertionError("oracle accepted missing/duplicate/erased history") env = isolated_env(Path(tempfile.mkdtemp(prefix="env-", dir=SCRATCH))) assert "OPENAI_API_KEY" not in env and "PYTHONPATH" not in env print("PASS fixture parser, chunk-boundary handling, negative mutants, isolated environment")
def alarm(_signum, _frame): raise TimeoutError("hard wall-clock deadline exceeded")
def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("binary", nargs="?", help="absolute path to installed Prime Agent executable") parser.add_argument("--timeout", type=int, default=90, help="whole-run hard deadline in seconds") parser.add_argument("--schema-id", default=SCHEMA, help="installed daemon schema ID (hello compatibility)") parser.add_argument("--self-test", action="store_true") args = parser.parse_args() if args.timeout < 1: parser.error("--timeout must be positive") os.umask(0o077) global SCRATCH base = Path.home() / ".cache" / "agent-scratch" / "prime-rust-migration" base.mkdir(mode=0o700, parents=True, exist_ok=True) SCRATCH = Path(tempfile.mkdtemp(prefix="pty-", dir=base)) print(f"Diagnostics: {SCRATCH}", flush=True) signal.signal(signal.SIGALRM, alarm) signal.alarm(args.timeout) if args.self_test: self_test() return 0 if not args.binary or not Path(args.binary).is_absolute() or not os.access(args.binary, os.X_OK): parser.error("an absolute installed executable path is required") probe = Probe(args, SCRATCH) try: probe.run() print("PASS installed CLI native-scrollback PTY regression", flush=True) return 0 except BaseException: diagnostic = traceback.format_exc() (SCRATCH / "failure.txt").write_text(f"phase: {getattr(probe, 'phase', 'launch')}\n{diagnostic}") print(diagnostic, file=sys.stderr, flush=True) return 1 finally: signal.alarm(0) probe.close()
if __name__ == "__main__": sys.exit(main())