diff --git a/backend/apps/agents/core/ws_manager.py b/backend/apps/agents/core/ws_manager.py index 04a63400..0f043ca8 100644 --- a/backend/apps/agents/core/ws_manager.py +++ b/backend/apps/agents/core/ws_manager.py @@ -18,6 +18,7 @@ _BROWSER_CMD_TIMEOUTS = { "replay_route": 20.0, # an API fetch can be slow "wait": 12.0, # smart-wait already caps itself well under this } +_BROWSER_CMD_REBROADCAST_S = 3.0 class ConnectionManager: @@ -242,16 +243,17 @@ class ConnectionManager: if not self.global_connections: return {"error": "No dashboard is connected. Open the dashboard to use browser tools."} - future = asyncio.get_event_loop().create_future() + loop = asyncio.get_event_loop() + future = loop.create_future() self.browser_futures[request_id] = future - await self.broadcast_global("browser:command", { + payload = { "request_id": request_id, "action": action, "browser_id": browser_id, "tab_id": tab_id, "params": params, - }) + } try: # Bound each command so a wedged tab can't block for 30s (the cost @@ -261,10 +263,21 @@ class ConnectionManager: # command just times out and the next success resets the agent's streak, # so only a SUSTAINED hang trips the fast-fail abort. timeout = _BROWSER_CMD_TIMEOUTS.get(action, _BROWSER_CMD_TIMEOUT_DEFAULT) - result = await asyncio.wait_for(future, timeout=timeout) - return result - except asyncio.TimeoutError: - return {"error": "Browser command timed out"} + deadline = loop.time() + timeout + # Re-broadcast until a client answers: a silently-dead dashboard + # socket takes up to ~35s of heartbeat to notice, and a command + # sent into that gap is lost forever (broadcast skips seq_log). + # The renderer dedupes by request_id so re-sends can't double-act. + while True: + await self.broadcast_global("browser:command", payload) + remaining = deadline - loop.time() + if remaining <= 0: + return {"error": "Browser command timed out"} + done, _ = await asyncio.wait( + {future}, timeout=min(_BROWSER_CMD_REBROADCAST_S, remaining) + ) + if done: + return future.result() finally: self.browser_futures.pop(request_id, None) diff --git a/backend/tests/test_browser_command_timeout.py b/backend/tests/test_browser_command_timeout.py index 19dce3ae..270a35ce 100644 --- a/backend/tests/test_browser_command_timeout.py +++ b/backend/tests/test_browser_command_timeout.py @@ -58,6 +58,28 @@ async def test_navigate_gets_the_longer_leash(monkeypatch): assert elapsed > 0.5, "navigate should use its longer bound, not the default" +@pytest.mark.asyncio +async def test_lost_first_delivery_heals_via_rebroadcast(monkeypatch): + # a silently-dead socket eats the first broadcast; the re-send after the + # rebroadcast interval must reach the (reconnected) client and succeed + monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUT_DEFAULT", 5.0) + monkeypatch.setattr(wsm, "_BROWSER_CMD_REBROADCAST_S", 0.1) + m = _mgr() + sends = [] + + class _CountingSock: + async def send_text(self, payload): + sends.append(payload) + if len(sends) >= 2: # first delivery "lost", second lands + rid = next(iter(m.browser_futures)) + m.resolve_browser_command(rid, {"text": "ok"}) + + m.global_connections = [_CountingSock()] + res = await m.send_browser_command("rid4", "get_text", "b1", {}) + assert res == {"text": "ok"} + assert len(sends) >= 2, "command must be re-broadcast until a client answers" + + @pytest.mark.asyncio async def test_a_resolved_command_returns_immediately(monkeypatch): # a healthy command returns the moment the renderer resolves it, not at the bound diff --git a/frontend/src/app/pages/Dashboard/hooks/lifecycle/useDashboardLifecycle.ts b/frontend/src/app/pages/Dashboard/hooks/lifecycle/useDashboardLifecycle.ts index 759bbbd6..362ff93b 100644 --- a/frontend/src/app/pages/Dashboard/hooks/lifecycle/useDashboardLifecycle.ts +++ b/frontend/src/app/pages/Dashboard/hooks/lifecycle/useDashboardLifecycle.ts @@ -83,6 +83,12 @@ export function useDashboardLifecycle({ dispatch(fetchSessions({ dashboardId })); dispatch(fetchLayout(dashboardId)); const cleanupBrowserHandler = initBrowserCommandHandler(); + // Global broadcasts (spawned browser cards) skip the replay log, so a + // socket gap loses them; a reconnect refetch is the only way they return. + const unsubReconnect = dashboardWs.on('dashboard:reconnected', () => { + dispatch(fetchSessions({ dashboardId })); + dispatch(fetchLayout(dashboardId)); + }); // DEFERRABLE: history list (for the search palette) and outputs // (for the apps panel) aren't on the first-paint path. Same for the // dashboard WS connection (it carries cross-session events; opens @@ -136,6 +142,7 @@ export function useDashboardLifecycle({ clearTimeout(warmTimer); warmAbort.abort(); cleanupBrowserHandler(); + unsubReconnect(); dashboardWs.disconnect(); // Cancel any not-yet-fired idle work; the cleanup handler can't // run partially if the dashboard switches before idle fired. diff --git a/frontend/src/shared/ws/WebSocketManager.ts b/frontend/src/shared/ws/WebSocketManager.ts index e6c79c0b..7e3b9e31 100644 --- a/frontend/src/shared/ws/WebSocketManager.ts +++ b/frontend/src/shared/ws/WebSocketManager.ts @@ -107,6 +107,7 @@ class WebSocketManager { // Set to true by `disconnect()` so we don't reconnect after an // explicit close (component unmount / user clicks Close). private explicitlyClosed: boolean = false; + private hasConnectedOnce: boolean = false; // Heartbeat. We send a ping on a fixed cadence and arm a timeout // for the pong; if the timeout fires, we force-close the socket so @@ -224,7 +225,13 @@ class WebSocketManager { // Dashboard / global WS: no resume, queue can flush right away. this.resumeAcked = true; this.flushQueue(); + // Global broadcasts skip the replay log, so anything missed during + // a socket gap only reappears if subscribers refetch on reconnect. + if (this.hasConnectedOnce) { + this.listeners.get('dashboard:reconnected')?.forEach((fn) => fn({})); + } } + this.hasConnectedOnce = true; }; this.ws.onmessage = (event) => {