Codex application override added.

tests/native_codex.py

@@ -0,0 +1,159 @@
+#!/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
+        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")
+            action("Insert+a", "Application Codex")
+            action("d", "Default")
+            action("Return", "Default settings restored")
+            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 and status")
+        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()