fix: stop worker-subprocess tests from hanging CI for hours

tests/_worker_io.py

@@ -0,0 +1,31 @@
+"""Timeout-guarded read from a worker subprocess's stdout.
+
+A bare ``proc.stdout.readline()`` blocks forever if the worker never writes the
+expected line (protocol mismatch, crash before flushing, deadlock) — the worker
+stays alive waiting on stdin, so the pipe never hits EOF either. One such stuck
+assertion turned a single flaky test into a CI job stuck for hours (until GitHub's
+default 6h job timeout, or a human, killed it). Every worker-subprocess test should
+read through this instead of calling ``proc.stdout.readline()`` directly.
+
+Reads run in a helper thread rather than gating on ``select()`` on the raw fd:
+``proc.stdout`` is a buffered ``TextIOWrapper``, so a burst of several JSON lines
+delivered in one OS-level read can sit fully in its internal buffer — ``select()``
+on the fd then reports "not ready" (nothing new at the OS level) even though
+``readline()`` would return one of those lines instantly.
+"""
+from __future__ import annotations
+
+import queue
+import subprocess
+import threading
+
+DEFAULT_TIMEOUT = 15.0
+
+
+def readline_timeout(proc: subprocess.Popen, timeout: float = DEFAULT_TIMEOUT) -> str:
+    result: "queue.Queue[str]" = queue.Queue(maxsize=1)
+    threading.Thread(target=lambda: result.put(proc.stdout.readline()), daemon=True).start()
+    try:
+        return result.get(timeout=timeout)
+    except queue.Empty:
+        raise TimeoutError(f"worker produced no output within {timeout}s (pid={proc.pid})") from None

tests/test_debugging.py

@@ -13,6 +13,7 @@
 from pathlib import Path
 
 import protocol as P
+from tests._worker_io import readline_timeout
 
 APP_DIR = Path(__file__).resolve().parent.parent
 
@@ -25,7 +26,7 @@ def _worker(entry: str) -> Iterator[subprocess.Popen]:
         env={**os.environ},
     )
     try:
-        proc.stdout.readline()  # startup graph topology
+        readline_timeout(proc)  # startup graph topology
         yield proc
     finally:
         proc.kill()
@@ -39,7 +40,7 @@ def _send(proc: subprocess.Popen, obj: dict) -> None:
 
 def _read_until(proc: subprocess.Popen, type_: str):
     for _ in range(80):
-        line = proc.stdout.readline()
+        line = readline_timeout(proc)
         if not line:
             break
         ev = P.ServerEventAdapter.validate_python(json.loads(line))
@@ -87,8 +88,8 @@ def test_ac03_fork_time_travel_with_override() -> None:
         _read_until(proc, "state_snapshot")
         # fork from there with an override; re-runs b, c on a new runId
         _send(proc, {"cmd": "fork", "checkpointId": ckpt, "stateOverride": {"log": ["X"]}})
+        final = _read_until(proc, "state_snapshot")              # final state (emitted before run_finished)
         _read_until(proc, "run_finished")                        # fork run finished
-        final = _read_until(proc, "state_snapshot")
         assert "X" in final.snapshot.values["log"]               # override is in the forked state
 
 

tests/test_entry_wizard.py

@@ -15,6 +15,7 @@
 
 import protocol as P
 from graphloupe_sidecar import discover
+from tests._worker_io import readline_timeout
 
 APP_DIR = Path(__file__).resolve().parent.parent
 
@@ -119,7 +120,7 @@ def _load_via_adapter(tmp_path: Path, symbol: str) -> P.ServerEvent:
     gl.mkdir()
     (gl / "entry.py").write_text(ADAPTER.format(symbol=symbol), encoding="utf-8")
     with _worker("entry:build_graph", tmp_path) as proc:
-        return P.ServerEventAdapter.validate_python(json.loads(proc.stdout.readline()))
+        return P.ServerEventAdapter.validate_python(json.loads(readline_timeout(proc)))
 
 
 def test_adapter_loads_compiled_graph_variable(tmp_path: Path) -> None:

tests/test_graph_loading.py

@@ -18,6 +18,7 @@
 
 import protocol as P
 from graphloupe_sidecar.server import app
+from tests._worker_io import readline_timeout
 
 APP_DIR = Path(__file__).resolve().parent.parent
 
@@ -37,7 +38,7 @@ def _worker(entry: str, env_extra: dict[str, str] | None = None) -> Iterator[sub
 
 
 def _first(proc: subprocess.Popen):
-    return P.ServerEventAdapter.validate_python(json.loads(proc.stdout.readline()))
+    return P.ServerEventAdapter.validate_python(json.loads(readline_timeout(proc)))
 
 
 def test_ac01_custom_entry_topology() -> None:

tests/test_health_check.py

@@ -11,6 +11,7 @@
 from pathlib import Path
 
 import protocol as P
+from tests._worker_io import readline_timeout
 
 APP_DIR = Path(__file__).resolve().parent.parent
 
@@ -53,7 +54,7 @@ def _worker(entry: str, root: Path) -> Iterator[subprocess.Popen]:
 def _topology(src: str, tmp_path: Path) -> P.GraphTopology:
     (tmp_path / "g.py").write_text(src, encoding="utf-8")
     with _worker("g:build_graph", tmp_path) as proc:
-        msg = P.ServerEventAdapter.validate_python(json.loads(proc.stdout.readline()))
+        msg = P.ServerEventAdapter.validate_python(json.loads(readline_timeout(proc)))
     assert isinstance(msg, P.GraphTopology)
     return msg
 

tests/test_manual_inference.py

@@ -18,6 +18,7 @@
 
 import protocol as P
 from graphloupe_sidecar.graph import build_manual_graph
+from tests._worker_io import readline_timeout
 
 APP_DIR = Path(__file__).resolve().parent.parent
 
@@ -46,7 +47,7 @@ def _worker(entry: str) -> Iterator[subprocess.Popen]:
         env={**os.environ},
     )
     try:
-        proc.stdout.readline()  # consume the startup graph topology line
+        readline_timeout(proc)  # consume the startup graph topology line
         yield proc
     finally:
         proc.kill()
@@ -60,7 +61,7 @@ def _send(proc: subprocess.Popen, obj: dict) -> None:
 
 def _read_until(proc: subprocess.Popen, type_: str):
     for _ in range(50):
-        line = proc.stdout.readline()
+        line = readline_timeout(proc)
         if not line:
             break
         ev = P.ServerEventAdapter.validate_python(json.loads(line))

tests/test_workspace_load.py

@@ -15,6 +15,7 @@
 from pathlib import Path
 
 import protocol as P
+from tests._worker_io import readline_timeout
 
 APP_DIR = Path(__file__).resolve().parent.parent
 
@@ -54,7 +55,7 @@ def _worker(entry: str, root: Path) -> Iterator[subprocess.Popen]:
 
 
 def _first(proc: subprocess.Popen):
-    return P.ServerEventAdapter.validate_python(json.loads(proc.stdout.readline()))
+    return P.ServerEventAdapter.validate_python(json.loads(readline_timeout(proc)))
 
 
 def test_ac01_loads_graph_from_project_root(tmp_path: Path) -> None: