Fix shell routes on Windows without PTY support

routes/shell_routes.py

@@ -14,9 +14,12 @@ from typing import Dict, Any
 try:
     import fcntl
     import pty
-except ImportError:
+except ImportError as exc:
     fcntl = None
     pty = None
+    _PTY_IMPORT_ERROR = exc
+else:
+    _PTY_IMPORT_ERROR = None
 
 from fastapi import APIRouter, Request, HTTPException
 from fastapi.responses import StreamingResponse
@@ -43,6 +46,8 @@ def _require_admin(request: Request):
 
 logger = logging.getLogger(__name__)
 
+PTY_SUPPORTED = pty is not None and fcntl is not None and hasattr(os, "setsid")
+
 
 def _find_line_break(buf):
     """Find next line terminator in buffer. Returns (index, separator_length) or (-1, 0)."""
@@ -63,6 +68,7 @@ EXEC_TIMEOUT = 30  # seconds — shorter than agent's 60s
 STREAM_TIMEOUT = 120  # default for short commands
 MAX_OUTPUT = 200_000  # truncate limit
 TMUX_LOG_DIR = Path(tempfile.gettempdir()) / "odysseus-tmux"
+PTY_UNSUPPORTED_ERROR = "pty_unsupported"
 
 
 class ShellExecRequest(BaseModel):
@@ -102,9 +108,12 @@ async def _exec_shell(command: str, timeout: int = EXEC_TIMEOUT) -> Dict[str, An
 
 async def _generate_pty(cmd: str, timeout: int, request: Request):
     """Run command in a pseudo-TTY so tqdm/progress bars work natively."""
-    if pty is None or fcntl is None:
-        yield f"data: {json.dumps({'stream': 'stderr', 'data': 'PTY streaming is not available on Windows'})}\n\n"
-        yield f"data: {json.dumps({'exit_code': -1})}\n\n"
+    if not PTY_SUPPORTED:
+        msg = "PTY streaming is not supported on this platform"
+        if _PTY_IMPORT_ERROR:
+            msg += f": {_PTY_IMPORT_ERROR}"
+        yield f"data: {json.dumps({'stream': 'stderr', 'data': msg, 'error': PTY_UNSUPPORTED_ERROR})}\n\n"
+        yield f"data: {json.dumps({'exit_code': -1, 'error': PTY_UNSUPPORTED_ERROR})}\n\n"
         return
 
     loop = asyncio.get_event_loop()

tests/test_shell_routes.py

@@ -1,9 +1,66 @@
-"""Tests for shell_routes.py — _find_line_break helper.
-Imports the function directly since it has no app dependencies."""
+"""Tests for shell_routes.py helpers."""
+
+import builtins
+import importlib.util
+import json
+import sys
+from pathlib import Path
+from types import SimpleNamespace
 
 from routes.shell_routes import _find_line_break
 
 
+def test_shell_routes_import_without_posix_pty_modules(monkeypatch):
+    """Native Windows has no fcntl/termios; importing routes must still work."""
+    real_import = builtins.__import__
+
+    def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
+        if name in {"fcntl", "pty"}:
+            raise ImportError(f"No module named {name!r}")
+        return real_import(name, globals, locals, fromlist, level)
+
+    monkeypatch.setattr(builtins, "__import__", fake_import)
+    cached_modules = {name: sys.modules.pop(name, None) for name in ("fcntl", "pty")}
+
+    module_path = Path(__file__).resolve().parents[1] / "routes" / "shell_routes.py"
+    spec = importlib.util.spec_from_file_location("_shell_routes_without_pty", module_path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    try:
+        spec.loader.exec_module(module)
+    finally:
+        sys.modules.pop(spec.name, None)
+        for name, cached_module in cached_modules.items():
+            if cached_module is not None:
+                sys.modules[name] = cached_module
+
+    assert module.PTY_SUPPORTED is False
+    assert module._find_line_break(b"ok\n") == (2, 1)
+
+
+async def test_generate_pty_reports_explicit_unsupported_error(monkeypatch):
+    """Clients can distinguish unsupported PTY mode from process failures."""
+    import routes.shell_routes as shell_routes
+
+    monkeypatch.setattr(shell_routes, "PTY_SUPPORTED", False)
+    monkeypatch.setattr(shell_routes, "_PTY_IMPORT_ERROR", ImportError("No module named 'termios'"))
+
+    request = SimpleNamespace(is_disconnected=lambda: False)
+    events = [
+        json.loads(chunk.removeprefix("data: ").strip())
+        async for chunk in shell_routes._generate_pty("echo hi", 5, request)
+    ]
+
+    assert events == [
+        {
+            "stream": "stderr",
+            "data": "PTY streaming is not supported on this platform: No module named 'termios'",
+            "error": shell_routes.PTY_UNSUPPORTED_ERROR,
+        },
+        {"exit_code": -1, "error": shell_routes.PTY_UNSUPPORTED_ERROR},
+    ]
+
+
 class TestFindLineBreak:
     """Test line-break detection in byte buffers."""