ci: Drive Windows integration TUIs through ConPTY

pyproject.toml

@@ -60,6 +60,7 @@ dev = [
     "pytest>=9.0.3",
     "pexpect>=4.9.0",
     "pyte>=0.8.2",
+    "pywinpty==3.0.5; sys_platform == 'win32'",
     # Pinned so local and CI lint with the identical formatter/type-checker;
     # ruff/ty output shifts between releases, so bump these deliberately.
     "ruff==0.16.5",

scripts/run_integration.py

@@ -7,7 +7,6 @@
 from __future__ import annotations
 
 import argparse
-import ast
 import base64
 import contextlib
 import datetime as dt
@@ -33,10 +32,14 @@
     "opencode": "opencode-ai",
 }
 WINDOWS_PATHEXT = ".COM;.EXE;.BAT;.CMD"
-PTY_MODULES = {"pexpect", "pyte"}
-PTY_HELPERS = {"utils.terminal", "utils.mcp"}
 # Claude exports no spans on Windows, where ug writes no machine-wide Claude settings.
 WINDOWS_UNSUPPORTED_MODULES = {"test_ug_claude_tracing.py"}
+WINDOWS_UNSUPPORTED_NODES = {
+    # Seeds machine-wide managed settings, which ug has no Windows path for.
+    "test_ug_configure_claude_lifecycle.py::test_unmanaged_claude_preserves_preexisting_family_defaults",
+    # Codex's Windows sandbox can't read ug's owner-only session-env dir yet.
+    "test_ug_smart_routing_hooks.py::test_smart_router_skill_toggles_codex_subagent_routing",
+}
 HEADLESS_TEST_NODES = {
     "claude": "test_ug_claude_headless.py::test_ug_claude_headless_prompt_argument",
     "codex": "test_ug_codex_headless.py::test_ug_codex_headless_prompt_argument",
@@ -128,34 +131,17 @@ def integration_test_targets(
     if platform_name == "nt" and installation_only:
         return [str(suite / "test_installation.py")]
     if platform_name == "nt":
-        return [
+        modules = [
             str(module)
             for module in sorted(suite.glob("test_*.py"))
-            if module.name not in WINDOWS_UNSUPPORTED_MODULES and not uses_pty(module)
+            if module.name not in WINDOWS_UNSUPPORTED_MODULES
         ]
+        # pytest matches --deselect against node IDs relative to the suite's pytest.ini.
+        deselected = [f"--deselect={node}" for node in sorted(WINDOWS_UNSUPPORTED_NODES)]
+        return [*modules, *deselected]
     return [str(suite)]
 
 
-def uses_pty(module: Path) -> bool:
-    """Whether a suite module drives agents through the POSIX-only PTY helpers."""
-    imported_names = []
-    for node in ast.walk(ast.parse(module.read_text(encoding="utf-8"))):
-        if isinstance(node, ast.ImportFrom):
-            imported_names.append(node.module or "")
-            imported_names.extend(
-                f"{node.module}.{alias.name}" if node.module else alias.name
-                for alias in node.names
-                if alias.name != "*"
-            )
-        elif isinstance(node, ast.Import):
-            imported_names.extend(alias.name for alias in node.names)
-    return any(
-        name.split(".")[0] in PTY_MODULES
-        or any(name == helper or name.endswith(f".{helper}") for helper in PTY_HELPERS)
-        for name in imported_names
-    )
-
-
 def process_group_options() -> dict:
     if os.name == "posix":
         return {"start_new_session": True}
@@ -786,9 +772,8 @@ def run(command, *, cwd=output, env=base_env, timeout=600) -> str:
             if client_id and client_secret:
                 bearer = mint_m2m_token(args.workspace, client_id, client_secret)
 
-        test_dependencies = ["pytest==9.0.3", "httpx==0.28.1"]
-        if os.name == "posix":
-            test_dependencies.extend(["pexpect==4.9.0", "pyte==0.8.2"])
+        test_dependencies = ["pytest==9.0.3", "httpx==0.28.1", "pyte==0.8.2"]
+        test_dependencies.append("pexpect==4.9.0" if os.name == "posix" else "pywinpty==3.0.5")
         run(
             [
                 uv,

tests/integration/README.md

@@ -202,7 +202,7 @@ through them without changing existing journey commands or completion assertions
 ## Run a specific combination
 
 Prerequisites: Python 3.12+, uv, and Node/npm. Live runs also require Databricks
-CLI 1.17.0. Full live/TUI runs require a POSIX host; Windows supports the explicit
+CLI 1.17.0. Windows drives TUIs through ConPTY (pywinpty) and supports the explicit
 headless subset described below. The runner installs the requested agents into a new
 npm prefix and ug into a new virtualenv. Pytest, the recorder's pinned `httpx`, and,
 for live runs, the PTY/screen libraries (pexpect and pyte) live in a different

tests/integration/test_ug_claude_custom_oauth.py

@@ -2,10 +2,12 @@
 
 import configparser
 import json
-import shlex
+import os
+from pathlib import Path
 from urllib.parse import urlparse
 
 import pytest
+from utils.harness import split_command
 from utils.terminal import AgentTerminal
 
 pytestmark = [pytest.mark.live, pytest.mark.tui, pytest.mark.claude]
@@ -49,11 +51,16 @@ def test_ug_claude_custom_oauth_cli_boots(live_session, workspace):
     assert profile == f"ug-oauth-{hostname}-{CLIENT_ID}"
     assert profiles[profile]["client_id"] == CLIENT_ID
 
-    managed_settings = json.loads(
-        session.run(MANAGED_SETTINGS_PATH, binary=session.which("cat"), timeout=30).stdout
-    )
-    assert shlex.split(managed_settings["apiKeyHelper"]) == [
-        str(session.binary.with_name("ug")),
+    if os.name == "nt":
+        # ug has no Windows managed-settings writer, so the helper lands in user settings.
+        settings_text = (session.home / ".claude/ucode-settings.json").read_text()
+    else:
+        settings_text = session.run(
+            MANAGED_SETTINGS_PATH, binary=session.which("cat"), timeout=30
+        ).stdout
+    helper = split_command(json.loads(settings_text)["apiKeyHelper"])
+    assert Path(helper[0]).with_suffix("") == session.binary.with_name("ug")
+    assert helper[1:] == [
         "auth-token",
         "--host",
         workspace,

tests/integration/test_ug_configure_claude.py

@@ -1,11 +1,11 @@
 """CUJs: configure Claude through ug, then use its real interactive session."""
 
 import json
-import shlex
 from pathlib import Path
 
 import pytest
 from utils.evidence import FileTask
+from utils.harness import split_command
 from utils.terminal import AgentTerminal, ConfigureTerminal
 
 pytestmark = [pytest.mark.live, pytest.mark.tui, pytest.mark.claude]
@@ -39,8 +39,8 @@ def test_ug_configure_claude_databricks(live_session, workspace):
     assert not session.workspace_state().get("provider_services", {}).get("claude")
 
     settings = json.loads((session.home / ".claude/ucode-settings.json").read_text())
-    helper = shlex.split(settings["apiKeyHelper"])
-    assert Path(helper[0]) == session.binary.with_name("ug")
+    helper = split_command(settings["apiKeyHelper"])
+    assert Path(helper[0]).with_suffix("") == session.binary.with_name("ug")
     assert helper[1] == "auth-token"
     token_result = session.run(*helper[1:], binary=helper[0], strip_ansi=False, timeout=30)
     assert token_result.stdout == "<redacted>\n"

tests/integration/test_ug_configure_codex.py

@@ -40,7 +40,7 @@ def test_ug_configure_codex_databricks(live_session, workspace):
 
     config = tomllib.loads((session.home / ".codex/ucode.config.toml").read_text())
     helper = config["model_providers"][config["model_provider"]]["auth"]
-    assert Path(helper["command"]) == session.binary.with_name("ug")
+    assert Path(helper["command"]).with_suffix("") == session.binary.with_name("ug")
     assert helper["args"][0] == "auth-token"
     token_result = session.run(
         *helper["args"], binary=helper["command"], strip_ansi=False, timeout=30

tests/integration/test_ug_configure_codex_lifecycle.py

@@ -1,5 +1,6 @@
 """CUJs for changing and undoing codex's ug setup."""
 
+import os
 import tomllib
 
 import pytest
@@ -22,6 +23,9 @@ def test_ug_configure_codex_repeat_and_revert(live_session, workspace):
     # Ordinary pre-existing user settings, not manufactured gateway state.
     user_path = session.home / ".codex/config.toml"
     original = "# user comment\n[notice]\nhide_rate_limit_model_nudge = true\n"
+    if os.name == "nt":
+        # A Windows user has picked a sandbox; `codex exec` rejects shell commands without one.
+        original += '[windows]\nsandbox = "unelevated"\n'
     user_path.parent.mkdir(parents=True)
     user_path.write_text(original)
 

tests/integration/test_ug_smart_routing_hooks.py

@@ -185,10 +185,17 @@ def completion_counts():
     before_answers, before_confirmations = completion_counts()
     tui.submit(invocation)
 
-    def toggled(_screen):
+    def toggled(screen):
+        if tui.approve_windows_command(screen):
+            return False
+        try:
+            controls_now = json.loads(controls[0].read_text())
+        except PermissionError:
+            # Windows denies reads while ug replaces the controls file; try again next frame.
+            return False
         answers, confirmations = completion_counts()
         return (
-            json.loads(controls[0].read_text()) == expected
+            controls_now == expected
             and confirmations > before_confirmations
             and answers > before_answers
         )

tests/integration/utils/constants.py

@@ -1,7 +1,10 @@
 """Shared constants for the integration CUJs."""
 
+import os
+
 CLAUDE_TEST_MODEL = "system.ai.claude-haiku-4-5"
-CODEX_TEST_MODEL = "system.ai.gpt-5-4-nano"
+# nano sometimes nests powershell inside Codex's pwsh, breaking `$` expansion in its file reads.
+CODEX_TEST_MODEL = "system.ai.gpt-5-6-luna" if os.name == "nt" else "system.ai.gpt-5-4-nano"
 
 MANAGED_CLAUDE_PROVIDER_SERVICE = "main.default.ci_e2e_anthropic_mps"
 MANAGED_CODEX_PROVIDER_SERVICE = "main.default.ci_e2e_openai_mps"

tests/integration/utils/harness.py

@@ -3,10 +3,12 @@
 from __future__ import annotations
 
 import contextlib
+import ctypes
 import json
 import os
 import queue
 import re
+import shlex
 import shutil
 import signal
 import subprocess
@@ -19,6 +21,21 @@
 ANSI = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]")
 
 
+def split_command(line: str) -> list[str]:
+    """Split a command line the way the OS will when an agent runs it."""
+    if os.name != "nt":
+        return shlex.split(line)
+    # ug writes Windows commands with list2cmdline; CreateProcess splits them like this.
+    split = ctypes.windll.shell32.CommandLineToArgvW
+    split.restype = ctypes.POINTER(ctypes.c_wchar_p)
+    count = ctypes.c_int()
+    argv = split(line, ctypes.byref(count))
+    try:
+        return [argv[index] for index in range(count.value)]
+    finally:
+        ctypes.windll.kernel32.LocalFree(argv)
+
+
 def process_group_options() -> dict:
     if os.name == "posix":
         return {"start_new_session": True}
@@ -182,7 +199,9 @@ def choose_codex_windows_sandbox(self) -> None:
             return
         config = self.home / ".codex/config.toml"
         config.parent.mkdir(parents=True, exist_ok=True)
-        config.write_text('[windows]\nsandbox = "unelevated"\n')
+        existing = config.read_text(encoding="utf-8") if config.is_file() else ""
+        if "[windows]" not in existing:
+            config.write_text(existing + '\n[windows]\nsandbox = "unelevated"\n', encoding="utf-8")
 
     def record(self, name: str, value: object) -> None:
         (self.artifacts / name).write_text(self.redact(json.dumps(value, indent=2)))