#!/usr/bin/env python3 # Copyright (c) 2026 Storm Dragon # SPDX-License-Identifier: GPL-3.0-or-later """Private X11 bridge/PTY test, with fake speech and no live Codex requests.""" import importlib.util import json import os from pathlib import Path import subprocess import sys import tempfile import time import tty def fixture_shell(): tty.setraw(0) os.write(1, "\x1b[2J\x1b[HOpenAI Codex\x1b[22;1H› ".encode()) while True: data = os.read(0, 4096) if not data: return if data == b"\x04": os.write(1, b"\x1b[24;1H\x1b[2K[Username@host directory] $ ") continue os.write(1, data) def main(): from native_terminal import KeyboardFixture, SpeechServer, wait_for if os.environ.get("FENRIR_PRIVATE_X11") != "1" or not os.environ.get("DISPLAY"): raise SystemExit("Use xvfb-run -a env FENRIR_PRIVATE_X11=1") repo = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location("bridge", repo / "tools/codex_bridge.py") bridge = importlib.util.module_from_spec(spec) spec.loader.exec_module(bridge) with tempfile.TemporaryDirectory(prefix="fenrir-codex-native-") as directory: root = Path(directory) (root / "keyboard").mkdir() (root / "keyboard/custom.toml").write_bytes((repo / "config/keyboard/desktop.toml").read_bytes()) (root / "home").mkdir() (root / "settings.toml").write_text('version=3\n[presentation.sound]\nenabled=false\n' f'[remote]\nsocket_file="{root / "remote.sock"}"\n') shell = root / "shell" shell.write_text(f'#!{sys.executable}\nimport runpy, sys\nsys.argv=[{str(Path(__file__).resolve())!r}, "--shell"]\nrunpy.run_path(sys.argv[0], run_name="__main__")\n') shell.chmod(0o700) server = SpeechServer(root / "speech.sock") keyboard = KeyboardFixture() env = dict(os.environ, HOME=str(root / "home"), TMPDIR=str(root), FENRIR_TEST_BINARY=os.environ.get("FENRIR_TEST_BINARY", str(repo / "fenrir")), SPEECHD_ADDRESS="unix_socket:" + str(root / "speech.sock"), LOGNAME="Username", USER="Username", LC_ALL="C", LANGUAGE="C") process = subprocess.Popen(["xterm", "-geometry", "100x24", "-e", sys.executable, str(repo / "tests/native_terminal.py"), "--outer", str(root)], env=env) watcher = None try: wait_for(lambda: (root / "started").exists(), "reader startup") state = json.loads((root / "started").read_text()) subprocess.run(["xdotool", "windowfocus", "--sync", str(state["window"])], check=True) wait_for(lambda: any("OpenAI Codex" in item for item in server.events), "fixture screen ready") def spoken(text, start): wait_for(lambda: any(text in item for item in server.events[start:]), f"speech {text!r}") def action(chord, expected): start = len(server.events) keyboard.key(chord) spoken(expected, start) action("Insert+a", "Application Default") action("c", "Codex") action("Return", "Codex application override selected") action("Insert+0", "Waiting for Codex bridge") session = "fixture-session" transcript = root / "session.jsonl" def append_record(kind, payload): with transcript.open("ab") as stream: stream.write((json.dumps({"type": kind, "payload": payload}) + "\n").encode()) append_record("session_meta", {"id": session}) append_record("event_msg", {"type": "agent_message", "message": "Existing reply"}) statusPath = root / "status.json" bridge.write_status(statusPath, "working") activePath = root / "active.json" bridge.write_json(activePath, {"key":"fixture", "epoch":1}) config = {"pid": os.getpid(), "started": bridge.process_info(os.getpid())[2], "window": str(state["window"]), "terminal": 1, "session": session, "transcript": str(transcript), "status_path": str(statusPath), "lock_path": str(root / "watch.lock"), "active_path": str(activePath), "key":"fixture"} watcher = subprocess.Popen([sys.executable, str(repo / "tools/codex_bridge.py"), "--watch", json.dumps(config)], env=env) wait_for(lambda: (root / "watch.lock").exists(), "bridge startup") # Wait for baseline delivery by polling the manual history command. start = len(server.events) for _ in range(20): keyboard.key("Insert+1") time.sleep(0.2) if "Existing reply" in server.events[start:]: break else: raise AssertionError("baseline history unavailable") action("Insert+0", "Codex working") # The raw fixture echoes terminal cursor escapes just like a TUI. # Actual XTest arrows must speak the unchanged composer character. keyboard.key("h", "e", "l", "l", "o") action("Left", "o") action("Left", "l") action("Right", "o") start = len(server.events) append_record("event_msg", {"type": "item_completed", "item": {"type": "CommandExecution", "aggregated_output": "Private tool result"}}) message = json.dumps({"type": "event_msg", "payload": {"type": "item_completed", "item": {"type": "AgentMessage", "content": [{"type": "Text", "text": "A completed progress update."}]}}}) + "\n" with transcript.open("ab") as stream: stream.write(message[:-3].encode()) time.sleep(0.5) assert not any("completed progress" in item or "Private tool result" in item for item in server.events[start:]) with transcript.open("ab") as stream: stream.write(message[-3:].encode()) spoken("A completed progress update.", start) time.sleep(2.3) # Include a complete heartbeat/retry interval. assert server.events[start:].count("A completed progress update.") == 1, server.events[start:] action("Insert+1", "A completed progress update.") action("Insert+2", "Existing reply") start = len(server.events) longReply = "Long reply " + "word " * 2000 append_record("event_msg", {"type": "agent_message", "message": longReply}) spoken("Long reply", start) assert any(item.strip() == longReply.strip() for item in server.events[start:]), "long reply truncated" watcher.terminate() watcher.wait(timeout=3) bridge.write_status(statusPath, "working") start = len(server.events) watcher = subprocess.Popen([sys.executable, str(repo / "tools/codex_bridge.py"), "--watch", json.dumps(config)], env=env) time.sleep(0.6) assert not any("Long reply" in item for item in server.events[start:]), "replay spoke old history" bridge.write_status(statusPath, "done") time.sleep(0.4) action("Insert+0", "Codex finished") # Leave old Codex/composer text visible and keep the bridge alive: # the fresh shell prompt itself must restore ordinary behavior. action("ctrl+d", "Default settings restored") keyboard.key("h", "e", "l", "l", "o") action("Left", "o") action("Left", "l") action("Right", "o") keyboard.key("Insert+q") wait_for(lambda: (root / "finished").exists(), "reader exit") assert not server.errors, server.errors print("PASS: real private PTY/override/socket bridge, composer Left/Right feedback, baseline history, completed messages only, one announcement, long replies, replay, status, shell exit without bridge closure and restored shell arrows") except BaseException: print("Recent private speech:", server.events[-15:], file=sys.stderr) if (root / "status.json").exists(): print("Bridge status:", (root / "status.json").read_text(), file=sys.stderr) if (root / "reader-errors").exists(): print("Reader errors:", (root / "reader-errors").read_text()[-2500:], file=sys.stderr) raise finally: if watcher is not None and watcher.poll() is None: watcher.terminate() watcher.wait(timeout=3) process.terminate() try: process.wait(timeout=3) except subprocess.TimeoutExpired: process.kill() keyboard.close() server.close() if __name__ == "__main__": fixture_shell() if "--shell" in sys.argv else main()