fix(tui): allow a longer handshake when launching through go run

A source checkout starts the TUI with `go run`, which can download a
toolchain and modules and then compile before the child can answer the
protocol handshake. The 10 second limit expires first, so the first launch
on a cold Go cache fails with "Timed out waiting for TUI protocol ready".

Give `go run` launches a 300 second handshake limit. Packaged binaries
keep the 10 second limit.

Fixes #1191

strix/interface/tui/backend/server.py

@@ -70,15 +70,18 @@ class TuiBackendServer:
         self.activated = False
         controller.set_change_callback(self.notify_changed)
 
-    async def start(self, connection: socket.socket) -> None:
+    async def start(
+        self, connection: socket.socket, handshake_timeout: float | None = None
+    ) -> None:
         """Negotiate protocol v3 before activating command or state traffic."""
         if self._socket is not None:
             raise RuntimeError("TUI backend is already started")
         connection.setblocking(False)  # noqa: FBT003
         self._socket = connection
+        timeout = _HANDSHAKE_TIMEOUT if handshake_timeout is None else handshake_timeout
         try:
             await self._send(envelope("hello", {"capabilities": list(PROTOCOL_CAPABILITIES)}))
-            await asyncio.wait_for(self._receive_ready(), timeout=_HANDSHAKE_TIMEOUT)
+            await asyncio.wait_for(self._receive_ready(), timeout=timeout)
         except TimeoutError as exc:
             raise ProtocolHandshakeError("Timed out waiting for TUI protocol ready") from exc
         except (EOFError, ConnectionError, OSError) as exc:

strix/interface/tui/runtime.py

@@ -48,6 +48,10 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+# A cold go run can download a toolchain and modules, then compile, before the
+# child answers the handshake.
+_GO_RUN_HANDSHAKE_TIMEOUT = 300.0
+
 
 def _revision_count(report: dict[str, Any]) -> int:
     history = report.get("update_history")
@@ -432,7 +436,10 @@ class GoTuiRuntime:
                     flush=True,
                 )
             process, backend_socket = await launch_tui_process(command, env, cwd)
-            await self.server.start(backend_socket)
+            await self.server.start(
+                backend_socket,
+                handshake_timeout=_GO_RUN_HANDSHAKE_TIMEOUT if cwd is not None else None,
+            )
             prepare_task = self._start_preparation()
             sync_task = asyncio.create_task(self.sync_state())
             return_code = await wait_process(process)

tests/test_go_tui_runtime.py

@@ -18,6 +18,8 @@ import pytest
 from strix.config.settings import DEFAULT_MAX_TURNS
 from strix.interface.tui import runtime as go_tui
 from strix.interface.tui import sidecar
+from strix.interface.tui.backend import server as server_module
+from strix.interface.tui.backend.protocol import PROTOCOL_CAPABILITIES
 from strix.interface.tui.runtime import GoTuiRuntime
 from strix.report.state import ReportState
 
@@ -265,6 +267,97 @@ async def test_runtime_prepares_state_before_launch_and_scans_only_after_ready(
             run_task.cancel()
 
 
+def _stub_launch(
+    monkeypatch: pytest.MonkeyPatch,
+    runtime: GoTuiRuntime,
+    command: list[str],
+    backend: socket.socket,
+) -> None:
+    async def launch(
+        _command: list[str], _env: dict[str, str], _cwd: str | None
+    ) -> tuple[SimpleNamespace, socket.socket]:
+        return SimpleNamespace(returncode=0), backend
+
+    async def wait_process(_process: object) -> int:
+        return 0
+
+    monkeypatch.setattr(runtime, "binary_command", lambda: command)
+    monkeypatch.setattr(runtime, "init_run_state", lambda: None)
+    monkeypatch.setattr(runtime, "start_scan", lambda: None)
+    monkeypatch.setattr(go_tui, "launch_tui_process", launch)
+    monkeypatch.setattr(go_tui, "wait_process", wait_process)
+    monkeypatch.setattr(go_tui, "persist_current", lambda: None)
+    monkeypatch.setattr(server_module, "_HANDSHAKE_TIMEOUT", 0.05)
+
+
+async def _reply_ready_after(child: socket.socket, delay: float) -> None:
+    await _receive_message(child)
+    await asyncio.sleep(delay)
+    await _send_message(
+        child,
+        {"version": 3, "type": "ready", "payload": {"capabilities": list(PROTOCOL_CAPABILITIES)}},
+    )
+
+
+@pytest.mark.asyncio
+async def test_go_run_child_may_compile_past_the_packaged_handshake_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    runtime_args = args()
+    runtime_args.needs_setup = False
+    runtime = GoTuiRuntime(runtime_args)
+    backend, child = socket.socketpair()
+    child.setblocking(False)  # noqa: FBT003
+    _stub_launch(monkeypatch, runtime, ["go", "run", "./cmd/strix-tui"], backend)
+
+    run_task = asyncio.create_task(runtime.run())
+    try:
+        await _reply_ready_after(child, 0.2)
+        await asyncio.wait_for(run_task, timeout=2)
+        assert runtime.server.activated is True
+    finally:
+        child.close()
+        if not run_task.done():
+            run_task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_packaged_child_slower_than_handshake_timeout_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    runtime_args = args()
+    runtime_args.needs_setup = False
+    runtime = GoTuiRuntime(runtime_args)
+    backend, child = socket.socketpair()
+    child.setblocking(False)  # noqa: FBT003
+    _stub_launch(monkeypatch, runtime, ["test-sidecar"], backend)
+
+    try:
+        with pytest.raises(go_tui.GoTuiPreActivationError, match="Timed out"):
+            await asyncio.wait_for(runtime.run(), timeout=2)
+    finally:
+        child.close()
+
+
+@pytest.mark.asyncio
+async def test_silent_go_run_child_still_times_out(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    runtime_args = args()
+    runtime_args.needs_setup = False
+    runtime = GoTuiRuntime(runtime_args)
+    backend, child = socket.socketpair()
+    child.setblocking(False)  # noqa: FBT003
+    _stub_launch(monkeypatch, runtime, ["go", "run", "./cmd/strix-tui"], backend)
+    monkeypatch.setattr(go_tui, "_GO_RUN_HANDSHAKE_TIMEOUT", 0.05)
+
+    try:
+        with pytest.raises(go_tui.GoTuiPreActivationError, match="Timed out"):
+            await asyncio.wait_for(runtime.run(), timeout=2)
+    finally:
+        child.close()
+
+
 @pytest.mark.asyncio
 async def test_pre_activation_failure_propagates_to_dispatcher(
     monkeypatch: pytest.MonkeyPatch,