# ========= 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. ========= """Durable Desktop command Inbox and independent command-result sync lane.""" from __future__ import annotations import asyncio import hashlib import logging import time from datetime import UTC, datetime from typing import Any import httpx from app.run_journal import ( CommandResultSyncBatch, OutboxLeaseLostError, RemoteCommandInboxRecord, SQLiteRunJournal, ) from app.run_sync.cloud_sync import ( CloudSyncConfiguration, RunEventSyncHttpError, RunEventSyncProtocolError, ) logger = logging.getLogger("command_sync") _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)): return float(value) parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=UTC) return parsed.timestamp() class HttpCommandSyncTransport: def __init__( self, *, timeout_seconds: float = 15.0, transport: httpx.AsyncBaseTransport | None = None, ) -> None: self._client = httpx.AsyncClient( timeout=timeout_seconds, transport=transport, ) self._registered: set[tuple[str, str, str]] = set() self._lock = asyncio.Lock() @staticmethod def _base(configuration: CloudSyncConfiguration) -> str: return configuration.endpoint_url.rsplit("/", 1)[0] @staticmethod def _headers( configuration: CloudSyncConfiguration, ) -> dict[str, str]: return { "Authorization": configuration.authorization, "X-Desktop-Instance-ID": configuration.desktop_instance_id, } async def _request( self, method: str, url: str, configuration: CloudSyncConfiguration, *, payload: dict[str, Any] | None = None, ) -> dict[str, Any]: response = await self._client.request( method, url, json=payload, headers=self._headers(configuration), ) if response.is_error: try: detail: Any = response.json() except ValueError: detail = response.text[:2000] raise RunEventSyncHttpError(response.status_code, detail) try: result = response.json() except ValueError as exc: raise RunEventSyncProtocolError( "Command sync returned non-JSON success response" ) from exc if not isinstance(result, dict): raise RunEventSyncProtocolError( "Command sync response must be a JSON object" ) return result async def ensure_registered( self, configuration: CloudSyncConfiguration ) -> None: 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: if key in self._registered: return try: await self._request( "POST", f"{key[0]}/devices/register", configuration, payload={ "capabilities": { "run_event_sync": 1, "command_inbox": 1, "command_result_sync": 1, } }, ) except RunEventSyncHttpError as exc: raise CommandSyncInfrastructureError( str(exc), code=_sync_error_code(exc.detail) ) from exc self._registered.add(key) async def pull_pending( self, configuration: CloudSyncConfiguration, *, limit: int ) -> list[dict[str, Any]]: await self.ensure_registered(configuration) result = await self._request( "GET", f"{self._base(configuration)}/commands/pending?limit={limit}", configuration, ) items = result.get("items") if not isinstance(items, list): raise RunEventSyncProtocolError( "Pending command response must contain an items array" ) return [item for item in items if isinstance(item, dict)] async def confirm_receipt( self, configuration: CloudSyncConfiguration, command: RemoteCommandInboxRecord, ) -> dict[str, Any]: await self.ensure_registered(configuration) return await self._request( "POST", f"{self._base(configuration)}/commands/{command.command_id}/confirm-receipt", configuration, payload={ "event_id": command.receipt_event_id, "desktop_event_sequence": 1, "occurred_at": datetime.fromtimestamp( command.created_at, tz=UTC ).isoformat(), "lease_token": command.delivery_lease_token, }, ) async def ingest_events( self, configuration: CloudSyncConfiguration, batch: CommandResultSyncBatch, ) -> dict[str, Any]: await self.ensure_registered(configuration) return await self._request( "POST", f"{self._base(configuration)}/commands/{batch.command_id}/events", configuration, payload={ "delivery_lease_token": batch.delivery_lease_token, "events": [ { "event_id": event.event_id, "desktop_event_sequence": event.command_event_sequence, "event_type": event.event_type, "payload": event.payload, "occurred_at": datetime.fromtimestamp( event.occurred_at, tz=UTC ).isoformat(), } for event in batch.events ], }, ) async def close(self) -> None: await self._client.aclose() class CommandControlWorker: def __init__( self, journal: SQLiteRunJournal, transport: HttpCommandSyncTransport, *, max_commands: int = 8, batch_size: int = 100, poll_interval_seconds: float = 1.0, max_retry_seconds: float = 300.0, ) -> None: self._journal = journal self._transport = transport self._max_commands = max_commands self._batch_size = batch_size self._poll_interval_seconds = poll_interval_seconds self._max_retry_seconds = max_retry_seconds self._configuration: CloudSyncConfiguration | None = None 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 self.notify() def start(self) -> None: if self._closed: raise RuntimeError("CommandControlWorker is closed") if self._task is None: self._task = asyncio.create_task( self._run(), name="remote-command-control" ) self.notify() def notify(self) -> None: if not self._closed: self._wake.set() async def persist_command( self, command: dict[str, Any] ) -> RemoteCommandInboxRecord: project_id = str( command.get("target_project_id") or command.get("project_id") ) run_id = command.get("target_task_id") or command.get("run_id") expires_at = _timestamp(command["expires_at"]) receipt_grace_until = _timestamp(command["receipt_grace_until"]) canonical_envelope = { "id": str(command["id"]), "session_id": str(command["session_id"]), "user_id": int(command["user_id"]), "space_id": command.get("space_id"), "project_id": project_id, "target_project_id": project_id, "active_task_id": command.get("active_task_id"), "target_task_id": run_id, "brain_session_id": command.get("brain_session_id"), "target_brain_session_id": command.get("target_brain_session_id"), "source_channel": command.get("source_channel", "remote_web"), "type": str(command["type"]), "payload": dict(command.get("payload") or {}), "next_task_id": command.get("next_task_id"), "route_version": int(command.get("route_version") or 1), "expires_at": expires_at, "receipt_grace_until": receipt_grace_until, "requires_online_receipt_confirmation": bool( command.get("requires_online_receipt_confirmation", False) ), } record = await asyncio.to_thread( self._journal.persist_remote_command, command_id=str(command["id"]), session_id=str(command["session_id"]), user_id=int(command["user_id"]), project_id=project_id, run_id=run_id, route_version=int(command.get("route_version") or 1), command_type=str(command["type"]), payload=canonical_envelope, expires_at=expires_at, receipt_grace_until=receipt_grace_until, requires_online_receipt_confirmation=bool( command.get("requires_online_receipt_confirmation", False) ), delivery_lease_token=command.get("lease_token"), receipt_event_id=command.get("receipt_event_id"), ) self.notify() return record async def confirm_receipt( self, command_id: str ) -> tuple[RemoteCommandInboxRecord, bool]: command = await asyncio.to_thread( self._journal.get_remote_command, command_id ) if command is None: raise KeyError(command_id) if command.receipt_status == "confirmed": return command, command.state in _EXECUTABLE_INBOX_STATES if command.receipt_status == "expired_late": return command, False configuration = self._configuration if configuration is None: # Low-risk commands may proceed during a Cloud outage inside a small # skew window. High-risk actions remain gated on the Cloud CAS. may_execute = ( command.state in _EXECUTABLE_INBOX_STATES and not command.requires_online_receipt_confirmation and time.time() <= command.expires_at + 5 ) return command, may_execute try: response = await self._transport.confirm_receipt( configuration, command ) except ( CommandSyncInfrastructureError, httpx.HTTPError, RunEventSyncHttpError, RunEventSyncProtocolError, ): logger.exception( "Command receipt confirmation failed", extra={"command_id": command_id}, ) may_execute = ( command.state in _EXECUTABLE_INBOX_STATES and not command.requires_online_receipt_confirmation and time.time() <= command.expires_at + 5 ) return command, may_execute result = response.get("result") status = "expired_late" if result == "expired_late" else "confirmed" updated = await asyncio.to_thread( self._journal.set_command_receipt_status, command_id, status, ) return updated, bool(response.get("may_execute")) and ( updated.state in _EXECUTABLE_INBOX_STATES ) async def drain_once(self) -> int: configuration = self._configuration if configuration is None: return 0 synced = await self._drain_outbound(configuration) pulled_count = await self._pull_and_reconcile(configuration) return synced + pulled_count async def _drain_outbound( self, configuration: CloudSyncConfiguration ) -> int: batches = await asyncio.to_thread( self._journal.claim_command_result_batches, max_commands=self._max_commands, batch_size=self._batch_size, ) if not batches: return 0 results = await asyncio.gather( *(self._sync_batch(batch, configuration) for batch in batches) ) self.notify() return sum(results) 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 as exc: self._inbound_retry_attempt += 1 delay = min( 2 ** min(self._inbound_retry_attempt, 8), self._max_retry_seconds, ) 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( { "id": item["command_id"], "session_id": item["session_id"], "user_id": item["user_id"], "project_id": item["project_id"], "run_id": item.get("run_id"), "route_version": item["route_version"], "type": item["command_type"], "payload": item.get("payload") or {}, "space_id": item.get("space_id"), "target_task_id": item.get("run_id"), "target_brain_session_id": item.get( "target_brain_session_id" ), "source_channel": item.get("source_channel"), "next_task_id": item.get("next_task_id"), "expires_at": item["expires_at"], "receipt_grace_until": item["receipt_grace_until"], "requires_online_receipt_confirmation": item.get( "requires_online_receipt_confirmation", False ), "lease_token": item.get("lease_token"), } ) pulled_count += 1 except Exception: logger.exception( "Ignoring malformed or conflicting remote command" ) for command in await asyncio.to_thread( self._journal.list_reconcilable_commands, limit=self._max_commands, ): if command.receipt_status == "pending": try: await self.confirm_receipt(command.command_id) except Exception: logger.exception( "Remote command receipt reconciliation failed", extra={"command_id": command.command_id}, ) return pulled_count async def _sync_batch( self, batch: CommandResultSyncBatch, configuration: CloudSyncConfiguration, ) -> int: try: response = await self._transport.ingest_events( configuration, batch ) expected = response.get("expected_next_desktop_event_sequence") if not isinstance(expected, int): raise RunEventSyncProtocolError( "Command ingest omitted expected next sequence" ) await asyncio.to_thread( self._journal.mark_command_result_batch_sent, batch ) return len(batch.events) except RunEventSyncHttpError as exc: if exc.status_code in {400, 409, 413, 422}: failed = self._failed_event_id(exc.detail, batch) await asyncio.to_thread( self._journal.block_command_result_batch, batch, failed_event_id=failed, error=str(exc), ) else: await self._retry(batch, str(exc)) except OutboxLeaseLostError: logger.info( "Ignoring stale command sync result", extra={"command_id": batch.command_id}, ) except Exception as exc: await self._retry(batch, f"{type(exc).__name__}: {exc}") return 0 async def _retry(self, batch: CommandResultSyncBatch, error: str) -> None: delay = min( 2 ** min(batch.attempt_count + 1, 8), self._max_retry_seconds ) try: await asyncio.to_thread( self._journal.retry_command_result_batch, batch, error=error, next_attempt_at=time.time() + delay, ) except OutboxLeaseLostError: logger.info("Command retry lost its lease") @staticmethod def _failed_event_id(detail: Any, batch: CommandResultSyncBatch) -> str: body = detail.get("detail", detail) if isinstance(detail, dict) else {} if isinstance(body, dict): failed = body.get("first_failed_event_id") if isinstance(failed, str) and any( event.event_id == failed for event in batch.events ): return failed return batch.events[0].event_id async def _run(self) -> None: while not self._closed: self._wake.clear() try: await self.drain_once() except asyncio.CancelledError: raise except Exception: logger.exception("Unexpected command control drain failure") try: await asyncio.wait_for( self._wake.wait(), timeout=self._poll_interval_seconds ) except TimeoutError: pass async def close(self) -> None: self._closed = True self._wake.set() if self._task is not None: self._task.cancel() try: await self._task except asyncio.CancelledError: pass self._task = None await self._transport.close()