fix(frontend-smoke): honor observation timeout during CDP setup

moli-frontend-smoke/README.md

@@ -96,7 +96,10 @@ concurrency, but a release conclusion should also repeat the reference run and c
 hash across scheduling modes.
 
 Each worker keeps one browser-level CDP connection and creates an isolated browser context per
-case. CI may pass `--reference-infrastructure-retries 1` to re-observe a case after a Chromium
+case. Discovery, the WebSocket handshake, and context/target setup use the configured observation
+timeout plus protocol headroom, just like the later observation commands. Discovery and the
+handshake share one connection deadline. Timeout diagnostics identify the command or event that
+was pending. CI may pass `--reference-infrastructure-retries 1` to re-observe a case after a Chromium
 connection or protocol failure. The retry does not apply to fixture errors, normalization errors,
 Moli failures, or DOM differences, and the original failure remains recorded in `summary.json`.
 

moli-frontend-smoke/moli_frontend_smoke/cdp.py

@@ -31,17 +31,19 @@ class CdpCommandError(CdpError):
 _DOM_ENABLE_PARAMS = {"includeWhitespace": "all"}
 
 
-def _read_json_url(url: str) -> dict[str, Any]:
+def _read_json_url(url: str, *, timeout: float = 2.0) -> dict[str, Any]:
     opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
-    with opener.open(url, timeout=2) as response:
+    with opener.open(url, timeout=timeout) as response:
         value = json.loads(response.read().decode("utf-8"))
     if not isinstance(value, dict):
         raise CdpError(f"unexpected CDP discovery response: {value!r}")
     return value
 
 
-async def discover_websocket_url(endpoint: str) -> str:
-    payload = await asyncio.to_thread(_read_json_url, endpoint.rstrip("/") + "/json/version")
+async def discover_websocket_url(endpoint: str, *, timeout: float = 2.0) -> str:
+    payload = await asyncio.to_thread(
+        _read_json_url, endpoint.rstrip("/") + "/json/version", timeout=timeout,
+    )
     url = payload.get("webSocketDebuggerUrl")
     if not isinstance(url, str) or not url:
         raise CdpError(f"CDP discovery did not include webSocketDebuggerUrl: {payload}")
@@ -90,11 +92,15 @@ class RawCdpClient:
     ) -> dict[str, Any]:
         message_id = await self.send(method, params, session_id=session_id)
         deadline = asyncio.get_running_loop().time() + timeout
+        timeout_message = f"timed out waiting for {method} response after {timeout:g}s"
         while True:
             remaining = deadline - asyncio.get_running_loop().time()
             if remaining <= 0:
-                raise CdpError(f"timed out waiting for {method} response")
-            message = await asyncio.wait_for(self.recv(), timeout=remaining)
+                raise CdpError(timeout_message)
+            try:
+                message = await asyncio.wait_for(self.recv(), timeout=remaining)
+            except TimeoutError as error:
+                raise CdpError(timeout_message) from error
             if message.get("id") == message_id:
                 if "error" in message:
                     raise CdpCommandError(method, message["error"])
@@ -121,24 +127,29 @@ class RawCdpClient:
             if matches(message):
                 return message
         deadline = asyncio.get_running_loop().time() + timeout
+        timeout_message = f"timed out waiting for {method} event after {timeout:g}s"
         while True:
             remaining = deadline - asyncio.get_running_loop().time()
             if remaining <= 0:
-                raise CdpError(f"timed out waiting for {method}")
-            message = await asyncio.wait_for(self.recv(), timeout=remaining)
+                raise CdpError(timeout_message)
+            try:
+                message = await asyncio.wait_for(self.recv(), timeout=remaining)
+            except TimeoutError as error:
+                raise CdpError(timeout_message) from error
             self.events.append(message)
             if matches(message):
                 return message
 
 
-async def connect(endpoint: str) -> RawCdpClient:
-    url = await discover_websocket_url(endpoint)
-    websocket = await websockets.connect(
-        url,
-        open_timeout=5,
-        max_size=32 * 1024 * 1024,
-        proxy=None,
-    )
+async def connect(endpoint: str, *, timeout: float = 10.0) -> RawCdpClient:
+    async with asyncio.timeout(timeout):
+        url = await discover_websocket_url(endpoint, timeout=timeout)
+        websocket = await websockets.connect(
+            url,
+            open_timeout=timeout,
+            max_size=32 * 1024 * 1024,
+            proxy=None,
+        )
     return RawCdpClient(websocket=websocket)
 
 
@@ -425,18 +436,20 @@ async def observe_case(
     frames: list[DomFrameObservation] = []
     final_root: dict[str, Any] | None = None
     diagnostics: dict[str, Any] = {}
+    command_timeout = max(10.0, timeout_ms / 1000 + 2.0)
     try:
         if client is None:
-            client = await connect(endpoint)
+            client = await connect(endpoint, timeout=command_timeout)
         else:
             client.events.clear()
-        context = await client.command("Target.createBrowserContext")
+        context = await client.command("Target.createBrowserContext", timeout=command_timeout)
         context_id = context.get("browserContextId")
         if not isinstance(context_id, str) or not context_id:
             raise CdpError(f"Target.createBrowserContext returned {context}")
         target = await client.command(
             "Target.createTarget",
             {"browserContextId": context_id, "url": "about:blank"},
+            timeout=command_timeout,
         )
         target_id = target.get("targetId")
         if not isinstance(target_id, str) or not target_id:
@@ -444,12 +457,12 @@ async def observe_case(
         attached = await client.command(
             "Target.attachToTarget",
             {"targetId": target_id, "flatten": True},
+            timeout=command_timeout,
         )
         session_id = attached.get("sessionId")
         if not isinstance(session_id, str) or not session_id:
             raise CdpError(f"Target.attachToTarget returned {attached}")
 
-        command_timeout = max(10.0, timeout_ms / 1000 + 2.0)
         await client.command("Runtime.enable", session_id=session_id, timeout=command_timeout)
         await client.command("Page.enable", session_id=session_id, timeout=command_timeout)
         await client.command(

moli-frontend-smoke/moli_frontend_smoke/runner.py

@@ -296,7 +296,9 @@ async def _observe_phase_case(
         if client is None:
             connect_started = time.perf_counter()
             try:
-                client = await connect(endpoint)
+                client = await connect(
+                    endpoint, timeout=max(10.0, timeout_ms / 1000 + 2.0),
+                )
             except Exception as error:
                 observation = EngineObservation(
                     engine=engine,

moli-frontend-smoke/tests/test_cdp_timeouts.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import asyncio
+from contextlib import suppress
+import json
+
+import pytest
+from websockets.asyncio.server import serve
+
+from moli_frontend_smoke.cdp import CdpError, RawCdpClient, connect
+
+
+@pytest.mark.parametrize("timeout", [0.25, 5.0])
+def test_connection_deadline_covers_discovery_and_websocket_handshake(timeout: float) -> None:
+    async def scenario() -> None:
+        async def handshake(_connection, _request):
+            await asyncio.sleep(0.15)
+
+        async def websocket_handler(websocket):
+            await websocket.wait_closed()
+
+        async with serve(
+            websocket_handler, "127.0.0.1", 0, process_request=handshake,
+        ) as websocket_server:
+            port = websocket_server.sockets[0].getsockname()[1]
+            websocket_url = f"ws://127.0.0.1:{port}/devtools/browser/probe"
+            handlers: set[asyncio.Task] = set()
+
+            async def discovery(reader, writer):
+                task = asyncio.current_task()
+                handlers.add(task)
+                try:
+                    await reader.readuntil(b"\r\n\r\n")
+                    await asyncio.sleep(0.15)
+                    body = json.dumps({"webSocketDebuggerUrl": websocket_url}).encode()
+                    writer.write(
+                        b"HTTP/1.1 200 OK\r\nConnection: close\r\n"
+                        + f"Content-Length: {len(body)}\r\n\r\n".encode()
+                        + body
+                    )
+                    await writer.drain()
+                finally:
+                    writer.close()
+                    with suppress(ConnectionError):
+                        await writer.wait_closed()
+                    handlers.discard(task)
+
+            async with await asyncio.start_server(discovery, "127.0.0.1", 0) as server:
+                port = server.sockets[0].getsockname()[1]
+                endpoint = f"http://127.0.0.1:{port}"
+                if timeout < 0.3:
+                    # Each phase fits individually, but their combined duration
+                    # exceeds the one connection deadline.
+                    with pytest.raises(TimeoutError):
+                        await connect(endpoint, timeout=timeout)
+                else:
+                    client = await connect(endpoint, timeout=timeout)
+                    await client.websocket.close()
+            await asyncio.gather(*handlers)
+
+    asyncio.run(scenario())
+
+
+class _SilentSocket:
+    async def send(self, _message):
+        pass
+
+    async def recv(self):
+        await asyncio.Event().wait()
+
+
+@pytest.mark.parametrize("operation", ["command", "event"])
+def test_timeout_diagnostic_identifies_the_pending_operation(operation: str) -> None:
+    async def scenario() -> None:
+        client = RawCdpClient(websocket=_SilentSocket())  # type: ignore[arg-type]
+        if operation == "command":
+            with pytest.raises(CdpError, match="Target.createTarget response after 0.01s"):
+                await client.command("Target.createTarget", timeout=0.01)
+        else:
+            with pytest.raises(CdpError, match="Page.loadEventFired event after 0.01s"):
+                await client.wait_event(
+                    "Page.loadEventFired", session_id="page", timeout=0.01,
+                )
+
+    asyncio.run(scenario())

moli-frontend-smoke/tests/test_runner.py

@@ -106,7 +106,7 @@ def test_observation_worker_reuses_one_cdp_connection(monkeypatch) -> None:
     clients: list[_FakeClient] = []
     observed_clients: list[_FakeClient] = []
 
-    async def fake_connect(_endpoint):
+    async def fake_connect(_endpoint, **_kwargs):
         client = _FakeClient()
         clients.append(client)
         return client
@@ -141,7 +141,7 @@ def test_chromium_retries_only_infrastructure_failures(monkeypatch) -> None:
     clients: list[_FakeClient] = []
     attempts = 0
 
-    async def fake_connect(_endpoint):
+    async def fake_connect(_endpoint, **_kwargs):
         client = _FakeClient()
         clients.append(client)
         return client
@@ -188,7 +188,7 @@ def test_chromium_retries_only_infrastructure_failures(monkeypatch) -> None:
 def test_fixture_failure_is_never_retried(monkeypatch) -> None:
     attempts = 0
 
-    async def fake_connect(_endpoint):
+    async def fake_connect(_endpoint, **_kwargs):
         return _FakeClient()
 
     async def fake_observe_case(*, engine, **_kwargs):
@@ -228,7 +228,7 @@ def test_fixture_failure_is_never_retried(monkeypatch) -> None:
 def test_moli_failure_is_never_retried(monkeypatch) -> None:
     attempts = 0
 
-    async def fake_connect(_endpoint):
+    async def fake_connect(_endpoint, **_kwargs):
         return _FakeClient()
 
     async def fake_observe_case(*, engine, **_kwargs):