diff --git a/backend/app/run_sync/command_sync.py b/backend/app/run_sync/command_sync.py index 463b925e4..60fc3c294 100644 --- a/backend/app/run_sync/command_sync.py +++ b/backend/app/run_sync/command_sync.py @@ -17,6 +17,7 @@ from __future__ import annotations import asyncio +import hashlib import logging import time from datetime import UTC, datetime @@ -44,6 +45,18 @@ _EXECUTABLE_INBOX_STATES = frozenset({"received", "dispatched"}) class CommandSyncInfrastructureError(RuntimeError): """Device registration failure, not a poison command result event.""" + def __init__(self, message: str, *, code: str | None = None) -> None: + super().__init__(message) + self.code = code + + +def _sync_error_code(detail: Any) -> str | None: + body = detail.get("detail", detail) if isinstance(detail, dict) else None + if not isinstance(body, dict): + return None + code = body.get("code") + return str(code) if isinstance(code, str) and code else None + def _timestamp(value: str | float | int) -> float: if isinstance(value, (float, int)): @@ -65,7 +78,7 @@ class HttpCommandSyncTransport: timeout=timeout_seconds, transport=transport, ) - self._registered: set[tuple[str, str]] = set() + self._registered: set[tuple[str, str, str]] = set() self._lock = asyncio.Lock() @staticmethod @@ -116,7 +129,14 @@ class HttpCommandSyncTransport: async def ensure_registered( self, configuration: CloudSyncConfiguration ) -> None: - key = (self._base(configuration), configuration.desktop_instance_id) + authorization_digest = hashlib.sha256( + configuration.authorization.encode("utf-8") + ).hexdigest() + key = ( + self._base(configuration), + configuration.desktop_instance_id, + authorization_digest, + ) if key in self._registered: return async with self._lock: @@ -136,7 +156,9 @@ class HttpCommandSyncTransport: }, ) except RunEventSyncHttpError as exc: - raise CommandSyncInfrastructureError(str(exc)) from exc + raise CommandSyncInfrastructureError( + str(exc), code=_sync_error_code(exc.detail) + ) from exc self._registered.add(key) async def pull_pending( @@ -227,6 +249,8 @@ class CommandControlWorker: self._wake = asyncio.Event() self._task: asyncio.Task[None] | None = None self._closed = False + self._inbound_retry_attempt = 0 + self._next_inbound_attempt_at = 0.0 def configure(self, configuration: CloudSyncConfiguration) -> None: self._configuration = configuration @@ -377,16 +401,49 @@ class CommandControlWorker: async def _pull_and_reconcile( self, configuration: CloudSyncConfiguration ) -> int: + if time.monotonic() < self._next_inbound_attempt_at: + return 0 pulled_count = 0 try: pulled = await self._transport.pull_pending( configuration, limit=self._max_commands ) - except Exception: - logger.exception( - "Remote command pull failed; outbound lane remains live" + except Exception as exc: + self._inbound_retry_attempt += 1 + delay = min( + 2 ** min(self._inbound_retry_attempt, 8), + self._max_retry_seconds, ) - pulled = [] + if isinstance(exc, CommandSyncInfrastructureError): + error_code = exc.code + elif isinstance(exc, RunEventSyncHttpError): + error_code = _sync_error_code(exc.detail) + else: + error_code = None + if error_code == "device_owner_mismatch": + # This cannot heal through a one-second retry loop. Keep the + # outbound lane live, but slow the device registration path + # enough to avoid a local/Cloud request and log storm. + delay = max(delay, min(60.0, self._max_retry_seconds)) + logger.warning( + "Remote command pull paused: Desktop device belongs to " + "another account; explicit device transfer or reset is " + "required", + extra={ + "error_code": error_code, + "retry_in_seconds": delay, + }, + ) + else: + logger.exception( + "Remote command pull failed; outbound lane remains live; " + "retrying with backoff", + extra={"retry_in_seconds": delay}, + ) + self._next_inbound_attempt_at = time.monotonic() + delay + return 0 + self._inbound_retry_attempt = 0 + self._next_inbound_attempt_at = 0.0 for item in pulled: try: await self.persist_command( diff --git a/backend/tests/app/run_sync/test_command_sync.py b/backend/tests/app/run_sync/test_command_sync.py index bd87f42ad..4582cc5b5 100644 --- a/backend/tests/app/run_sync/test_command_sync.py +++ b/backend/tests/app/run_sync/test_command_sync.py @@ -20,7 +20,10 @@ import httpx import pytest from app.run_journal import InvalidRunTransitionError, SQLiteRunJournal -from app.run_sync.cloud_sync import CloudSyncConfiguration +from app.run_sync.cloud_sync import ( + CloudSyncConfiguration, + RunEventSyncHttpError, +) from app.run_sync.command_sync import ( CommandControlWorker, HttpCommandSyncTransport, @@ -33,8 +36,10 @@ class FakeCommandTransport: self.confirmed: list[str] = [] self.ingested: list[Any] = [] self.pull_error: Exception | None = None + self.pull_count = 0 async def pull_pending(self, _configuration, *, limit): + self.pull_count += 1 if self.pull_error is not None: raise self.pull_error return self.pending[:limit] @@ -166,6 +171,36 @@ async def test_inbound_pull_failure_does_not_starve_outbound_results(tmp_path): await worker.close() +@pytest.mark.asyncio +async def test_device_owner_mismatch_backs_off_inbound_without_hot_loop( + tmp_path, +): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + transport = FakeCommandTransport() + transport.pull_error = RunEventSyncHttpError( + 409, + { + "detail": { + "code": "device_owner_mismatch", + "message": "Desktop device belongs to another user", + } + }, + ) + worker = CommandControlWorker( + journal, + transport, + poll_interval_seconds=0.01, + max_retry_seconds=300, + ) + worker.configure(_configuration()) + + assert await worker.drain_once() == 0 + assert await worker.drain_once() == 0 + assert transport.pull_count == 1 + assert worker._next_inbound_attempt_at > 0 + await worker.close() + + @pytest.mark.asyncio async def test_device_registration_error_retries_command_lane(tmp_path): def handler(request: httpx.Request) -> httpx.Response: @@ -186,6 +221,35 @@ async def test_device_registration_error_retries_command_lane(tmp_path): await worker.close() +@pytest.mark.asyncio +async def test_transport_reregisters_when_authenticated_credential_changes(): + registrations: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/devices/register"): + registrations.append(request.headers["authorization"]) + return httpx.Response(200, json={}) + assert request.url.path.endswith("/commands/pending") + return httpx.Response(200, json={"items": []}) + + transport = HttpCommandSyncTransport( + transport=httpx.MockTransport(handler) + ) + first = _configuration() + second = CloudSyncConfiguration( + endpoint_url=first.endpoint_url, + authorization="Bearer another-account-token", + desktop_instance_id=first.desktop_instance_id, + ) + + assert await transport.pull_pending(first, limit=1) == [] + assert await transport.pull_pending(first, limit=1) == [] + assert await transport.pull_pending(second, limit=1) == [] + + assert registrations == ["Bearer token", "Bearer another-account-token"] + await transport.close() + + def test_command_inbox_terminal_state_cannot_move_backwards(tmp_path): with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: record = journal.persist_remote_command( diff --git a/src/components/Dispatch/index.tsx b/src/components/Dispatch/index.tsx index ac599fbe9..badac5f17 100644 --- a/src/components/Dispatch/index.tsx +++ b/src/components/Dispatch/index.tsx @@ -20,6 +20,7 @@ import { SESSION_SIDE_PANEL_CONTENT_WIDTH_CLASS } from '@/components/Session/ses import { Button } from '@/components/ui/button'; import { createRemoteControlSession, + getRemoteControlBridgeError, getRemoteControlDesktopInstanceId, isRemoteControlAlreadyGoneError, parseRemoteControlLinkToken, @@ -437,10 +438,19 @@ export function WorkspaceDispatch() { try { const bridgeReady = await waitForRemoteControlBridgeConnected(); if (!bridgeReady) { - toast.error('Remote control is still connecting.', { - description: - 'Keep Eigent Desktop open and try again in a few seconds.', - }); + const bridgeError = getRemoteControlBridgeError(); + toast.error( + bridgeError?.retryable === false + ? 'Remote control needs attention.' + : 'Remote control is still connecting.', + { + description: + bridgeError?.code === 'device_owner_mismatch' + ? 'This Desktop is registered to another Eigent account. Sign in with that account or explicitly reset/transfer the Desktop device registration.' + : bridgeError?.message || + 'Keep Eigent Desktop open and try again in a few seconds.', + } + ); return; } diff --git a/src/hooks/useRemoteControlBridge.test.ts b/src/hooks/useRemoteControlBridge.test.ts index d0d4f3153..f057df4fa 100644 --- a/src/hooks/useRemoteControlBridge.test.ts +++ b/src/hooks/useRemoteControlBridge.test.ts @@ -22,6 +22,21 @@ describe('remote command durable ACK replay', () => { window.localStorage.clear(); }); + it('treats a device owner mismatch as a terminal bridge error', () => { + expect( + __remoteControlBridgeTestHooks.serverBridgeError({ + type: 'error', + code: 'device_owner_mismatch', + message: 'Desktop device belongs to another user', + retryable: false, + }) + ).toEqual({ + code: 'device_owner_mismatch', + message: 'Desktop device belongs to another user', + retryable: false, + }); + }); + it('replays the canonical completed outcome without executing again', () => { expect( ackFromDurableExecution('command-1', { diff --git a/src/hooks/useRemoteControlBridge.ts b/src/hooks/useRemoteControlBridge.ts index 0bee2c369..d9dd41d1b 100644 --- a/src/hooks/useRemoteControlBridge.ts +++ b/src/hooks/useRemoteControlBridge.ts @@ -24,6 +24,7 @@ import { getRemoteControlDesktopInstanceId, getRemoteControlWebSocketUrl, setRemoteControlBridgeConnected, + type RemoteControlBridgeError, } from '@/lib/remoteControl'; import { createFollowUpRequest, @@ -1119,8 +1120,28 @@ export const __remoteControlBridgeTestHooks = { pendingCommandResult, queuePendingCommandResult, removePendingCommandResult, + serverBridgeError, }; +function serverBridgeError(message: any): RemoteControlBridgeError | null { + if (message?.type !== 'error') { + return null; + } + const code = + typeof message.code === 'string' && message.code + ? message.code + : 'bridge_error'; + const text = + typeof message.message === 'string' && message.message + ? message.message + : 'Remote control bridge registration failed.'; + const retryable = + typeof message.retryable === 'boolean' + ? message.retryable + : code !== 'device_owner_mismatch'; + return { code, message: text, retryable }; +} + export function useRemoteControlBridge(token: string | null | undefined) { const cacheRef = useRef>(new Map()); const rateLimitRef = useRef([]); @@ -1132,10 +1153,12 @@ export function useRemoteControlBridge(token: string | null | undefined) { useEffect(() => { if (!token || !isDesktop()) { - setRemoteControlBridgeConnected(false); + setRemoteControlBridgeConnected(false, null); return; } + setRemoteControlBridgeConnected(false, null); + let stopped = false; let ws: WebSocket | null = null; let reconnectTimer: number | null = null; @@ -1561,7 +1584,6 @@ export function useRemoteControlBridge(token: string | null | undefined) { attempt: reconnectAttempt, }); ws.onopen = () => { - reconnectAttempt = 0; send({ type: 'subscribe', desktop_instance_id: desktopInstanceId, @@ -1580,6 +1602,9 @@ export function useRemoteControlBridge(token: string | null | undefined) { try { const message = JSON.parse(event.data); if (message?.type === 'connected') { + // Reset backoff only after application-level registration. A raw + // WebSocket open followed by a policy rejection is not success. + reconnectAttempt = 0; console.info( '[RemoteControlBridge][RC-TRACE] bridge registered on server', { desktop_instance_id: desktopInstanceId } @@ -1590,6 +1615,19 @@ export function useRemoteControlBridge(token: string | null | undefined) { .then(reconcileRemoteFollowUps); return; } + const bridgeError = serverBridgeError(message); + if (bridgeError) { + console.error( + '[RemoteControlBridge] Bridge registration rejected', + bridgeError + ); + setRemoteControlBridgeConnected(false, bridgeError); + if (!bridgeError.retryable) { + stopped = true; + ws?.close(1000, bridgeError.code.slice(0, 120)); + } + return; + } if (message?.type === 'auth_expired') { // JWT expired or rejected. Stop attempting; the auth store will // refresh on its own schedule and the hook will rerun with the @@ -1624,7 +1662,18 @@ export function useRemoteControlBridge(token: string | null | undefined) { stopped, attempt: reconnectAttempt, }); - setRemoteControlBridgeConnected(false); + if (!stopped && event?.code === 1008) { + stopped = true; + setRemoteControlBridgeConnected(false, { + code: 'bridge_policy_rejected', + message: + event?.reason || + 'Remote control bridge registration was rejected by policy.', + retryable: false, + }); + } else { + setRemoteControlBridgeConnected(false); + } if (pingTimer) { window.clearInterval(pingTimer); pingTimer = null; diff --git a/src/lib/remoteControl.bridge.test.ts b/src/lib/remoteControl.bridge.test.ts new file mode 100644 index 000000000..fb08db830 --- /dev/null +++ b/src/lib/remoteControl.bridge.test.ts @@ -0,0 +1,50 @@ +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + +import { + getRemoteControlBridgeError, + setRemoteControlBridgeConnected, + waitForRemoteControlBridgeConnected, +} from './remoteControl'; + +describe('remote control bridge state', () => { + beforeEach(() => { + setRemoteControlBridgeConnected(false, null); + }); + + it('resolves waiters immediately when registration is terminally rejected', async () => { + const waiting = waitForRemoteControlBridgeConnected(5_000); + + setRemoteControlBridgeConnected(false, { + code: 'device_owner_mismatch', + message: 'Desktop device belongs to another user', + retryable: false, + }); + + await expect(waiting).resolves.toBe(false); + expect(getRemoteControlBridgeError()?.code).toBe('device_owner_mismatch'); + }); + + it('clears a previous bridge error after registration succeeds', () => { + setRemoteControlBridgeConnected(false, { + code: 'device_owner_mismatch', + message: 'Desktop device belongs to another user', + retryable: false, + }); + + setRemoteControlBridgeConnected(true); + + expect(getRemoteControlBridgeError()).toBeNull(); + }); +}); diff --git a/src/lib/remoteControl.ts b/src/lib/remoteControl.ts index 79f8fc55f..c95e8976f 100644 --- a/src/lib/remoteControl.ts +++ b/src/lib/remoteControl.ts @@ -28,6 +28,13 @@ const PENDING_COMMAND_REQUEST_TTL_MS = 24 * 60 * 60 * 1000; const PENDING_COMMAND_REQUEST_LIMIT = 32; let remoteControlBridgeConnected = false; +let remoteControlBridgeError: RemoteControlBridgeError | null = null; + +export type RemoteControlBridgeError = { + code: string; + message: string; + retryable: boolean; +}; type PendingCommandRequest = { fingerprint: string; @@ -130,10 +137,20 @@ export function getRemoteControlDesktopInstanceId(): Promise { return getDesktopInstanceId(); } -export function setRemoteControlBridgeConnected(connected: boolean): void { +export function setRemoteControlBridgeConnected( + connected: boolean, + error?: RemoteControlBridgeError | null +): void { remoteControlBridgeConnected = connected; + if (connected) { + remoteControlBridgeError = null; + } else if (error !== undefined) { + remoteControlBridgeError = error; + } window.dispatchEvent( - new CustomEvent(BRIDGE_READY_EVENT, { detail: { connected } }) + new CustomEvent(BRIDGE_READY_EVENT, { + detail: { connected, error: remoteControlBridgeError }, + }) ); } @@ -141,6 +158,10 @@ export function isRemoteControlBridgeConnected(): boolean { return remoteControlBridgeConnected; } +export function getRemoteControlBridgeError(): RemoteControlBridgeError | null { + return remoteControlBridgeError; +} + export function waitForRemoteControlBridgeConnected( timeoutMs = 2500 ): Promise { @@ -154,8 +175,14 @@ export function waitForRemoteControlBridgeConnected( }, timeoutMs); const onReady = (event: Event) => { - const connected = Boolean((event as CustomEvent).detail?.connected); + const detail = (event as CustomEvent).detail; + const connected = Boolean(detail?.connected); if (!connected) { + if (detail?.error?.retryable === false) { + window.clearTimeout(timeout); + window.removeEventListener(BRIDGE_READY_EVENT, onReady); + resolve(false); + } return; } window.clearTimeout(timeout);