diff --git a/backend/app/controller/remote_command_controller.py b/backend/app/controller/remote_command_controller.py new file mode 100644 index 000000000..9015f71f9 --- /dev/null +++ b/backend/app/controller/remote_command_controller.py @@ -0,0 +1,156 @@ +"""Durable Remote Control Inbox endpoints used by the Desktop renderer.""" + +from __future__ import annotations + +import asyncio +import time +from dataclasses import asdict +from datetime import UTC, datetime, timedelta +from typing import Any, Literal + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, Field + +from app.run_journal import get_default_run_journal +from app.run_sync.runtime import ( + notify_default_cloud_sync_worker, + persist_and_confirm_remote_command, +) + +router = APIRouter(prefix="/remote-control/commands") + +_HIGH_RISK_COMMAND_TYPES = { + "stop", + "stop_task", + "remove_task", + "space_apply_project_run", + "space_discard_project_overlays", +} + + +class RemoteCommandInboxIn(BaseModel): + id: str = Field(min_length=1, max_length=64) + session_id: str = Field(min_length=1, max_length=64) + user_id: int + project_id: str | None = None + target_project_id: str | None = None + run_id: str | None = None + target_task_id: str | None = None + route_version: int = Field(default=1, ge=1) + type: str = Field(min_length=1, max_length=64) + payload: dict[str, Any] = Field(default_factory=dict) + expires_at: datetime | None = None + receipt_grace_until: datetime | None = None + requires_online_receipt_confirmation: bool = False + receipt_event_id: str | None = Field(default=None, max_length=64) + + +class CommandAdmissionIn(BaseModel): + status: Literal["accepted", "rejected"] + event_id: str | None = Field(default=None, max_length=64) + reason: str | None = None + + +class CommandExecutionResultIn(BaseModel): + status: Literal["completed", "failed"] + event_id: str | None = Field(default=None, max_length=64) + result: dict[str, Any] = Field(default_factory=dict) + error_code: str | None = None + error: str | None = None + + +def _normalized_command(body: RemoteCommandInboxIn) -> dict[str, Any]: + now = datetime.now(UTC) + expires_at = body.expires_at or now + timedelta(minutes=2) + receipt_grace_until = body.receipt_grace_until or expires_at + timedelta( + seconds=30 + ) + project_id = body.target_project_id or body.project_id + if not project_id: + raise HTTPException(status_code=422, detail="project_id is required") + return { + **body.model_dump(mode="json"), + "project_id": project_id, + "target_project_id": project_id, + "run_id": body.target_task_id or body.run_id, + "expires_at": expires_at.isoformat(), + "receipt_grace_until": receipt_grace_until.isoformat(), + "requires_online_receipt_confirmation": ( + body.requires_online_receipt_confirmation + or body.type in _HIGH_RISK_COMMAND_TYPES + ), + } + + +@router.post("/inbox") +async def persist_command_inbox(body: RemoteCommandInboxIn): + try: + record, may_execute = await persist_and_confirm_remote_command( + _normalized_command(body) + ) + except RuntimeError as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc + except KeyError as exc: + raise HTTPException( + status_code=422, detail=f"missing command field: {exc.args[0]}" + ) from exc + return {"command": asdict(record), "may_execute": may_execute} + + +@router.get("/inbox/pending") +async def list_pending_command_inbox(limit: int = 100): + journal = get_default_run_journal() + records = await asyncio.to_thread( + journal.list_reconcilable_commands, limit=min(max(limit, 1), 500) + ) + dispatched = [] + for record in records: + dispatched.append( + await asyncio.to_thread( + journal.mark_command_dispatched, record.command_id + ) + ) + return {"items": [asdict(record) for record in dispatched]} + + +@router.post("/{command_id}/admission") +async def record_command_admission(command_id: str, body: CommandAdmissionIn): + journal = get_default_run_journal() + record = await asyncio.to_thread(journal.get_remote_command, command_id) + if record is None: + raise HTTPException(status_code=404, detail="Command not found") + if body.status == "accepted" and record.receipt_status == "expired_late": + raise HTTPException( + status_code=409, detail="Expired command cannot be admitted" + ) + event = await asyncio.to_thread( + journal.append_command_result, + command_id, + event_type=f"admission.{body.status}", + event_id=body.event_id, + payload={"reason": body.reason} if body.reason else {}, + occurred_at=time.time(), + ) + notify_default_cloud_sync_worker() + return asdict(event) + + +@router.post("/{command_id}/result") +async def record_command_result( + command_id: str, body: CommandExecutionResultIn +): + journal = get_default_run_journal() + event = await asyncio.to_thread( + journal.append_command_result, + command_id, + event_type=f"execution.{body.status}", + event_id=body.event_id, + payload={ + "result": body.result, + "error_code": body.error_code, + "error": body.error, + }, + occurred_at=time.time(), + ) + notify_default_cloud_sync_worker() + return asdict(event) diff --git a/backend/app/router.py b/backend/app/router.py index aeb4ac4bd..eccf03db8 100644 --- a/backend/app/router.py +++ b/backend/app/router.py @@ -29,6 +29,7 @@ from app.controller import ( mcp_controller, message_controller, model_controller, + remote_command_controller, remote_sub_agent_controller, run_controller, skill_controller, @@ -89,6 +90,11 @@ def register_routers(app: FastAPI, prefix: str = "") -> None: "tags": ["Runs"], "description": "Durable Run snapshots, event replay, and live streams", }, + { + "router": remote_command_controller.router, + "tags": ["Remote Command Inbox"], + "description": "Durable Remote Control Inbox and command-result lane", + }, { "router": model_controller.router, "tags": ["model"], diff --git a/backend/app/run_journal/__init__.py b/backend/app/run_journal/__init__.py index 92e49022d..93e27f333 100644 --- a/backend/app/run_journal/__init__.py +++ b/backend/app/run_journal/__init__.py @@ -13,7 +13,10 @@ # ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= from app.run_journal.models import ( + CommandResultEvent, + CommandResultSyncBatch, CommittedRunEvent, + RemoteCommandInboxRecord, RunEventDraft, RunEventSyncBatch, RunEventSyncOutboxRecord, @@ -41,6 +44,8 @@ from app.run_journal.store import ( __all__ = [ "SCHEMA_VERSION", "CommittedRunEvent", + "CommandResultEvent", + "CommandResultSyncBatch", "EventRecorder", "IdempotencyConflictError", "OptimisticConcurrencyError", @@ -51,6 +56,7 @@ __all__ = [ "RunJournalError", "RunNotFoundError", "RunRecord", + "RemoteCommandInboxRecord", "SQLiteRunJournal", "UnsupportedSchemaVersionError", "close_default_run_journal", diff --git a/backend/app/run_journal/models.py b/backend/app/run_journal/models.py index 2ea5051b5..3b93a27d0 100644 --- a/backend/app/run_journal/models.py +++ b/backend/app/run_journal/models.py @@ -79,3 +79,43 @@ class RunEventSyncBatch: lease_token: str attempt_count: int events: tuple[CommittedRunEvent, ...] + + +@dataclass(frozen=True) +class RemoteCommandInboxRecord: + command_id: str + session_id: str + user_id: int + project_id: str + run_id: str | None + route_version: int + command_type: str + payload: dict[str, Any] + expires_at: float + receipt_grace_until: float + requires_online_receipt_confirmation: bool + receipt_event_id: str + receipt_status: str + state: str + dispatch_attempt_count: int + last_error: str | None + created_at: float + updated_at: float + + +@dataclass(frozen=True) +class CommandResultEvent: + event_id: str + command_id: str + command_event_sequence: int + event_type: str + payload: dict[str, Any] + occurred_at: float + + +@dataclass(frozen=True) +class CommandResultSyncBatch: + command_id: str + lease_token: str + attempt_count: int + events: tuple[CommandResultEvent, ...] diff --git a/backend/app/run_journal/store.py b/backend/app/run_journal/store.py index db0ab57bc..41bcdec54 100644 --- a/backend/app/run_journal/store.py +++ b/backend/app/run_journal/store.py @@ -33,7 +33,10 @@ from pathlib import Path from typing import Any from app.run_journal.models import ( + CommandResultEvent, + CommandResultSyncBatch, CommittedRunEvent, + RemoteCommandInboxRecord, RunEventDraft, RunEventSyncBatch, RunEventSyncOutboxRecord, @@ -41,7 +44,7 @@ from app.run_journal.models import ( ) from app.run_journal.paths import default_run_journal_path -SCHEMA_VERSION = 2 +SCHEMA_VERSION = 3 _MIGRATION_V1 = """ BEGIN IMMEDIATE; @@ -173,6 +176,81 @@ PRAGMA user_version = 2; COMMIT; """ +_MIGRATION_V3 = """ +BEGIN IMMEDIATE; + +CREATE TABLE remote_command_inbox ( + command_id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + user_id INTEGER NOT NULL, + project_id TEXT NOT NULL, + run_id TEXT, + route_version INTEGER NOT NULL CHECK (route_version > 0), + command_type TEXT NOT NULL, + payload_json TEXT NOT NULL, + expires_at REAL NOT NULL, + receipt_grace_until REAL NOT NULL CHECK (receipt_grace_until >= expires_at), + requires_online_receipt_confirmation INTEGER NOT NULL DEFAULT 0 CHECK ( + requires_online_receipt_confirmation IN (0, 1) + ), + receipt_event_id TEXT NOT NULL UNIQUE, + receipt_status TEXT NOT NULL DEFAULT 'pending' CHECK ( + receipt_status IN ('pending', 'confirmed', 'expired_late') + ), + state TEXT NOT NULL DEFAULT 'received' CHECK ( + state IN ('received', 'dispatched', 'accepted', 'rejected', 'completed', 'failed') + ), + dispatch_attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (dispatch_attempt_count >= 0), + last_error TEXT, + created_at REAL NOT NULL, + updated_at REAL NOT NULL +); + +CREATE INDEX remote_command_inbox_dispatch_idx +ON remote_command_inbox(state, updated_at, command_id); + +CREATE TABLE command_result_events ( + event_id TEXT PRIMARY KEY, + command_id TEXT NOT NULL REFERENCES remote_command_inbox(command_id) ON DELETE CASCADE, + command_event_sequence INTEGER NOT NULL CHECK (command_event_sequence > 0), + event_type TEXT NOT NULL, + payload_json TEXT NOT NULL, + occurred_at REAL NOT NULL, + UNIQUE(command_id, command_event_sequence) +); + +CREATE INDEX command_result_events_replay_idx +ON command_result_events(command_id, command_event_sequence); + +CREATE TABLE command_result_outbox ( + event_id TEXT PRIMARY KEY REFERENCES command_result_events(event_id) ON DELETE CASCADE, + command_id TEXT NOT NULL, + command_event_sequence INTEGER NOT NULL CHECK (command_event_sequence > 0), + status TEXT NOT NULL DEFAULT 'pending' CHECK ( + status IN ('pending', 'sending', 'sent', 'dead_letter') + ), + attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0), + next_attempt_at REAL NOT NULL, + lease_token TEXT, + lease_until REAL, + last_error TEXT, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + UNIQUE(command_id, command_event_sequence) +); + +CREATE INDEX command_result_outbox_pending_idx +ON command_result_outbox( + status, next_attempt_at, lease_until, command_id, command_event_sequence +); + +INSERT OR IGNORE INTO run_journal_migrations(version, applied_at) +VALUES (3, CAST(strftime('%s', 'now') AS REAL)); + +PRAGMA user_version = 3; +COMMIT; +""" + class RunJournalError(RuntimeError): """Base error for local RunJournal operations.""" @@ -464,6 +542,483 @@ class SQLiteRunJournal: rows = self._connection.execute(query, parameters).fetchall() return [self._event_from_row(row) for row in rows] + def persist_remote_command( + self, + *, + command_id: str, + session_id: str, + user_id: int, + project_id: str, + run_id: str | None, + route_version: int, + command_type: str, + payload: dict[str, Any], + expires_at: float, + receipt_grace_until: float, + requires_online_receipt_confirmation: bool, + receipt_event_id: str | None = None, + now: float | None = None, + ) -> RemoteCommandInboxRecord: + """Commit the Inbox row and receipt event/outbox in one transaction.""" + + if route_version < 1: + raise ValueError("route_version must be positive") + if receipt_grace_until < expires_at: + raise ValueError("receipt grace must not precede expiry") + timestamp = now if now is not None else time.time() + receipt_id = receipt_event_id or str(uuid.uuid4()) + payload_json = json.dumps( + payload, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + with self._write_transaction() as connection: + existing = connection.execute( + "SELECT * FROM remote_command_inbox WHERE command_id = ?", + (command_id,), + ).fetchone() + if existing is not None: + canonical = ( + existing["session_id"] == session_id + and int(existing["user_id"]) == user_id + and existing["project_id"] == project_id + and existing["run_id"] == run_id + and int(existing["route_version"]) == route_version + and existing["command_type"] == command_type + and existing["payload_json"] == payload_json + and float(existing["expires_at"]) == expires_at + and float(existing["receipt_grace_until"]) + == receipt_grace_until + and bool(existing["requires_online_receipt_confirmation"]) + == requires_online_receipt_confirmation + ) + if not canonical: + raise IdempotencyConflictError( + f"command_id {command_id!r} was reused with different data" + ) + return self._command_from_row(existing) + + connection.execute( + """ + INSERT INTO remote_command_inbox( + command_id, session_id, user_id, project_id, run_id, + route_version, command_type, payload_json, expires_at, + receipt_grace_until, + requires_online_receipt_confirmation, receipt_event_id, + receipt_status, state, dispatch_attempt_count, last_error, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, + 'pending', 'received', 0, NULL, ?, ?) + """, + ( + command_id, + session_id, + user_id, + project_id, + run_id, + route_version, + command_type, + payload_json, + expires_at, + receipt_grace_until, + int(requires_online_receipt_confirmation), + receipt_id, + timestamp, + timestamp, + ), + ) + receipt_payload = "{}" + connection.execute( + """ + INSERT INTO command_result_events( + event_id, command_id, command_event_sequence, event_type, + payload_json, occurred_at + ) VALUES (?, ?, 1, 'receipt.durably_received', ?, ?) + """, + (receipt_id, command_id, receipt_payload, timestamp), + ) + connection.execute( + """ + INSERT INTO command_result_outbox( + event_id, command_id, command_event_sequence, status, + attempt_count, next_attempt_at, lease_token, lease_until, + last_error, created_at, updated_at + ) VALUES (?, ?, 1, 'pending', 0, ?, NULL, NULL, NULL, ?, ?) + """, + (receipt_id, command_id, timestamp, timestamp, timestamp), + ) + row = connection.execute( + "SELECT * FROM remote_command_inbox WHERE command_id = ?", + (command_id,), + ).fetchone() + assert row is not None + return self._command_from_row(row) + + def get_remote_command( + self, command_id: str + ) -> RemoteCommandInboxRecord | None: + with self._lock: + row = self._connection.execute( + "SELECT * FROM remote_command_inbox WHERE command_id = ?", + (command_id,), + ).fetchone() + return self._command_from_row(row) if row is not None else None + + def list_reconcilable_commands( + self, *, limit: int = 100 + ) -> list[RemoteCommandInboxRecord]: + with self._lock: + rows = self._connection.execute( + """ + SELECT * FROM remote_command_inbox + WHERE state IN ('received', 'dispatched', 'accepted') + ORDER BY updated_at, command_id + LIMIT ? + """, + (limit,), + ).fetchall() + return [self._command_from_row(row) for row in rows] + + def mark_command_dispatched( + self, command_id: str, *, now: float | None = None + ) -> RemoteCommandInboxRecord: + timestamp = now if now is not None else time.time() + with self._write_transaction() as connection: + updated = connection.execute( + """ + UPDATE remote_command_inbox + SET state = 'dispatched', + dispatch_attempt_count = dispatch_attempt_count + 1, + updated_at = ? + WHERE command_id = ? AND state IN ('received', 'dispatched') + """, + (timestamp, command_id), + ) + if updated.rowcount != 1: + row = connection.execute( + "SELECT * FROM remote_command_inbox WHERE command_id = ?", + (command_id,), + ).fetchone() + if row is None: + raise RunNotFoundError( + f"command_id {command_id!r} does not exist" + ) + return self._command_from_row(row) + row = connection.execute( + "SELECT * FROM remote_command_inbox WHERE command_id = ?", + (command_id,), + ).fetchone() + assert row is not None + return self._command_from_row(row) + + def set_command_receipt_status( + self, + command_id: str, + status: str, + *, + error: str | None = None, + now: float | None = None, + ) -> RemoteCommandInboxRecord: + if status not in {"confirmed", "expired_late"}: + raise ValueError("invalid command receipt status") + timestamp = now if now is not None else time.time() + with self._write_transaction() as connection: + updated = connection.execute( + """ + UPDATE remote_command_inbox + SET receipt_status = ?, last_error = ?, updated_at = ? + WHERE command_id = ? + """, + (status, error, timestamp, command_id), + ) + if updated.rowcount != 1: + raise RunNotFoundError( + f"command_id {command_id!r} does not exist" + ) + row = connection.execute( + "SELECT * FROM remote_command_inbox WHERE command_id = ?", + (command_id,), + ).fetchone() + assert row is not None + return self._command_from_row(row) + + def append_command_result( + self, + command_id: str, + *, + event_type: str, + payload: dict[str, Any] | None = None, + event_id: str | None = None, + occurred_at: float | None = None, + ) -> CommandResultEvent: + state_by_event = { + "admission.accepted": "accepted", + "admission.rejected": "rejected", + "execution.completed": "completed", + "execution.failed": "failed", + } + if ( + event_type not in state_by_event + and event_type != "execution.started" + ): + raise ValueError("unsupported command result event type") + timestamp = occurred_at if occurred_at is not None else time.time() + result_event_id = event_id or str(uuid.uuid4()) + payload_value = payload or {} + payload_json = json.dumps( + payload_value, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + with self._write_transaction() as connection: + duplicate = connection.execute( + "SELECT * FROM command_result_events WHERE event_id = ?", + (result_event_id,), + ).fetchone() + if duplicate is not None: + if ( + duplicate["command_id"] != command_id + or duplicate["event_type"] != event_type + or duplicate["payload_json"] != payload_json + ): + raise IdempotencyConflictError( + f"command event {result_event_id!r} was reused" + ) + return self._command_event_from_row(duplicate) + inbox = connection.execute( + "SELECT * FROM remote_command_inbox WHERE command_id = ?", + (command_id,), + ).fetchone() + if inbox is None: + raise RunNotFoundError( + f"command_id {command_id!r} does not exist" + ) + sequence = int( + connection.execute( + """ + SELECT COALESCE(MAX(command_event_sequence), 0) + 1 + FROM command_result_events WHERE command_id = ? + """, + (command_id,), + ).fetchone()[0] + ) + connection.execute( + """ + INSERT INTO command_result_events( + event_id, command_id, command_event_sequence, event_type, + payload_json, occurred_at + ) VALUES (?, ?, ?, ?, ?, ?) + """, + ( + result_event_id, + command_id, + sequence, + event_type, + payload_json, + timestamp, + ), + ) + connection.execute( + """ + INSERT INTO command_result_outbox( + event_id, command_id, command_event_sequence, status, + attempt_count, next_attempt_at, lease_token, lease_until, + last_error, created_at, updated_at + ) VALUES (?, ?, ?, 'pending', 0, ?, NULL, NULL, NULL, ?, ?) + """, + ( + result_event_id, + command_id, + sequence, + timestamp, + timestamp, + timestamp, + ), + ) + next_state = state_by_event.get(event_type) + if next_state is not None: + connection.execute( + """ + UPDATE remote_command_inbox + SET state = ?, last_error = ?, updated_at = ? + WHERE command_id = ? + """, + ( + next_state, + payload_value.get("error"), + timestamp, + command_id, + ), + ) + return CommandResultEvent( + event_id=result_event_id, + command_id=command_id, + command_event_sequence=sequence, + event_type=event_type, + payload=payload_value, + occurred_at=timestamp, + ) + + def claim_command_result_batches( + self, + *, + now: float | None = None, + max_commands: int = 8, + batch_size: int = 100, + lease_seconds: float = 30.0, + ) -> list[CommandResultSyncBatch]: + if max_commands < 1 or batch_size < 1 or lease_seconds <= 0: + raise ValueError("command outbox claim limits must be positive") + timestamp = now if now is not None else time.time() + batches: list[CommandResultSyncBatch] = [] + with self._write_transaction() as connection: + connection.execute( + """ + UPDATE command_result_outbox + SET status = 'pending', lease_token = NULL, + lease_until = NULL, updated_at = ? + WHERE status = 'sending' + AND (lease_until IS NULL OR lease_until <= ?) + """, + (timestamp, timestamp), + ) + candidates = connection.execute( + """ + SELECT command_id, MIN(command_event_sequence) AS head_sequence + FROM command_result_outbox + WHERE status != 'sent' + GROUP BY command_id + HAVING SUM(CASE WHEN status = 'dead_letter' THEN 1 ELSE 0 END) = 0 + AND MIN(CASE WHEN status = 'pending' THEN next_attempt_at END) <= ? + ORDER BY MIN(updated_at), command_id + LIMIT ? + """, + (timestamp, max_commands), + ).fetchall() + for candidate in candidates: + rows = connection.execute( + """ + SELECT e.*, o.status, o.attempt_count, o.next_attempt_at + FROM command_result_outbox AS o + JOIN command_result_events AS e ON e.event_id = o.event_id + WHERE o.command_id = ? AND o.command_event_sequence >= ? + AND o.status != 'sent' + ORDER BY o.command_event_sequence + LIMIT ? + """, + ( + candidate["command_id"], + candidate["head_sequence"], + batch_size, + ), + ).fetchall() + ready: list[sqlite3.Row] = [] + expected = int(candidate["head_sequence"]) + for row in rows: + if ( + int(row["command_event_sequence"]) != expected + or row["status"] != "pending" + or float(row["next_attempt_at"]) > timestamp + ): + break + ready.append(row) + expected += 1 + if not ready: + continue + token = uuid.uuid4().hex + event_ids = [row["event_id"] for row in ready] + placeholders = ",".join("?" for _ in event_ids) + updated = connection.execute( + f""" + UPDATE command_result_outbox + SET status = 'sending', lease_token = ?, lease_until = ?, + updated_at = ? + WHERE status = 'pending' AND event_id IN ({placeholders}) + """, + ( + token, + timestamp + lease_seconds, + timestamp, + *event_ids, + ), + ) + if updated.rowcount != len(event_ids): + raise OutboxLeaseLostError( + f"failed to lease command lane {candidate['command_id']!r}" + ) + batches.append( + CommandResultSyncBatch( + command_id=candidate["command_id"], + lease_token=token, + attempt_count=int(ready[0]["attempt_count"]), + events=tuple( + self._command_event_from_row(row) for row in ready + ), + ) + ) + return batches + + def mark_command_result_batch_sent( + self, + batch: CommandResultSyncBatch, + *, + now: float | None = None, + ) -> None: + self._finish_command_result_batch(batch, "sent", now=now) + + def retry_command_result_batch( + self, + batch: CommandResultSyncBatch, + *, + error: str, + next_attempt_at: float, + now: float | None = None, + ) -> None: + self._finish_command_result_batch( + batch, + "pending", + error=error, + next_attempt_at=next_attempt_at, + increment_attempt=True, + now=now, + ) + + def block_command_result_batch( + self, + batch: CommandResultSyncBatch, + *, + failed_event_id: str, + error: str, + now: float | None = None, + ) -> None: + timestamp = now if now is not None else time.time() + event_ids = [event.event_id for event in batch.events] + if failed_event_id not in event_ids: + raise ValueError("failed event must belong to command batch") + with self._write_transaction() as connection: + self._assert_command_batch_lease(connection, batch, event_ids) + placeholders = ",".join("?" for _ in event_ids) + connection.execute( + f""" + UPDATE command_result_outbox + SET status = 'pending', lease_token = NULL, + lease_until = NULL, updated_at = ? + WHERE event_id IN ({placeholders}) + """, + (timestamp, *event_ids), + ) + connection.execute( + """ + UPDATE command_result_outbox + SET status = 'dead_letter', attempt_count = attempt_count + 1, + last_error = ?, updated_at = ? + WHERE event_id = ? + """, + (error[:4000], timestamp, failed_event_id), + ) + def list_pending_outbox( self, *, now: float | None = None, limit: int = 100 ) -> list[RunEventSyncOutboxRecord]: @@ -683,6 +1238,8 @@ class SQLiteRunJournal: self._connection.executescript(_MIGRATION_V1) if version < 2: self._connection.executescript(_MIGRATION_V2) + if version < 3: + self._connection.executescript(_MIGRATION_V3) @contextmanager def _write_transaction(self) -> Iterator[sqlite3.Connection]: @@ -741,6 +1298,63 @@ class SQLiteRunJournal: f"outbox lease for {batch.run_id!r} is stale" ) + @staticmethod + def _assert_command_batch_lease( + connection: sqlite3.Connection, + batch: CommandResultSyncBatch, + event_ids: list[str], + ) -> None: + placeholders = ",".join("?" for _ in event_ids) + count = int( + connection.execute( + f""" + SELECT COUNT(*) FROM command_result_outbox + WHERE status = 'sending' AND lease_token = ? + AND event_id IN ({placeholders}) + """, + (batch.lease_token, *event_ids), + ).fetchone()[0] + ) + if count != len(event_ids): + raise OutboxLeaseLostError( + f"command outbox lease for {batch.command_id!r} is stale" + ) + + def _finish_command_result_batch( + self, + batch: CommandResultSyncBatch, + status: str, + *, + error: str | None = None, + next_attempt_at: float | None = None, + increment_attempt: bool = False, + now: float | None = None, + ) -> None: + timestamp = now if now is not None else time.time() + event_ids = [event.event_id for event in batch.events] + with self._write_transaction() as connection: + self._assert_command_batch_lease(connection, batch, event_ids) + placeholders = ",".join("?" for _ in event_ids) + connection.execute( + f""" + UPDATE command_result_outbox + SET status = ?, + attempt_count = attempt_count + ?, + next_attempt_at = COALESCE(?, next_attempt_at), + lease_token = NULL, lease_until = NULL, + last_error = ?, updated_at = ? + WHERE event_id IN ({placeholders}) + """, + ( + status, + int(increment_attempt), + next_attempt_at, + error[:4000] if error else None, + timestamp, + *event_ids, + ), + ) + @staticmethod def _run_from_row(row: sqlite3.Row) -> RunRecord: return RunRecord( @@ -768,6 +1382,42 @@ class SQLiteRunJournal: run_version=int(row["run_version"]), ) + @staticmethod + def _command_from_row(row: sqlite3.Row) -> RemoteCommandInboxRecord: + return RemoteCommandInboxRecord( + command_id=row["command_id"], + session_id=row["session_id"], + user_id=int(row["user_id"]), + project_id=row["project_id"], + run_id=row["run_id"], + route_version=int(row["route_version"]), + command_type=row["command_type"], + payload=json.loads(row["payload_json"]), + expires_at=float(row["expires_at"]), + receipt_grace_until=float(row["receipt_grace_until"]), + requires_online_receipt_confirmation=bool( + row["requires_online_receipt_confirmation"] + ), + receipt_event_id=row["receipt_event_id"], + receipt_status=row["receipt_status"], + state=row["state"], + dispatch_attempt_count=int(row["dispatch_attempt_count"]), + last_error=row["last_error"], + created_at=float(row["created_at"]), + updated_at=float(row["updated_at"]), + ) + + @staticmethod + def _command_event_from_row(row: sqlite3.Row) -> CommandResultEvent: + return CommandResultEvent( + event_id=row["event_id"], + command_id=row["command_id"], + command_event_sequence=int(row["command_event_sequence"]), + event_type=row["event_type"], + payload=json.loads(row["payload_json"]), + occurred_at=float(row["occurred_at"]), + ) + @staticmethod def _outbox_from_row(row: sqlite3.Row) -> RunEventSyncOutboxRecord: return RunEventSyncOutboxRecord( diff --git a/backend/app/run_sync/cloud_sync.py b/backend/app/run_sync/cloud_sync.py index c40adc531..06dfaeb48 100644 --- a/backend/app/run_sync/cloud_sync.py +++ b/backend/app/run_sync/cloud_sync.py @@ -61,19 +61,35 @@ class HttpRunEventSyncTransport: timeout=timeout_seconds, transport=transport, ) + self._registered_devices: set[tuple[str, str]] = set() + self._claimed_routes: set[tuple[str, str, str]] = set() + self._registration_lock = asyncio.Lock() - async def ingest( + @staticmethod + def _headers( + configuration: CloudSyncConfiguration, + ) -> dict[str, str]: + return { + "Authorization": configuration.authorization, + "X-Desktop-Instance-ID": configuration.desktop_instance_id, + } + + @staticmethod + def _sync_base(configuration: CloudSyncConfiguration) -> str: + return configuration.endpoint_url.rsplit("/", 1)[0] + + async def _json_request( self, + method: str, + url: str, configuration: CloudSyncConfiguration, payload: dict[str, Any], ) -> dict[str, Any]: - response = await self._client.post( - configuration.endpoint_url, + response = await self._client.request( + method, + url, json=payload, - headers={ - "Authorization": configuration.authorization, - "X-Desktop-Instance-ID": configuration.desktop_instance_id, - }, + headers=self._headers(configuration), ) if response.is_error: try: @@ -85,14 +101,61 @@ class HttpRunEventSyncTransport: result = response.json() except ValueError as exc: raise RunEventSyncProtocolError( - "Run event ingest returned non-JSON success response" + "Run sync endpoint returned non-JSON success response" ) from exc if not isinstance(result, dict): raise RunEventSyncProtocolError( - "Run event ingest response must be a JSON object" + "Run sync response must be a JSON object" ) return result + async def _ensure_device_and_route( + self, + configuration: CloudSyncConfiguration, + project_id: str, + ) -> None: + base = self._sync_base(configuration) + device_key = ( + base, + configuration.desktop_instance_id, + ) + route_key = (*device_key, project_id) + if route_key in self._claimed_routes: + return + async with self._registration_lock: + if device_key not in self._registered_devices: + await self._json_request( + "POST", + f"{base}/devices/register", + configuration, + {"capabilities": {"run_event_sync": 1, "command_sync": 1}}, + ) + self._registered_devices.add(device_key) + if route_key not in self._claimed_routes: + await self._json_request( + "PUT", + f"{base}/projects/{project_id}/execution-route", + configuration, + {}, + ) + self._claimed_routes.add(route_key) + + async def ingest( + self, + configuration: CloudSyncConfiguration, + payload: dict[str, Any], + ) -> dict[str, Any]: + await self._ensure_device_and_route( + configuration, + str(payload["project_id"]), + ) + return await self._json_request( + "POST", + configuration.endpoint_url, + configuration, + payload, + ) + async def close(self) -> None: await self._client.aclose() diff --git a/backend/app/run_sync/command_sync.py b/backend/app/run_sync/command_sync.py new file mode 100644 index 000000000..854f4cd15 --- /dev/null +++ b/backend/app/run_sync/command_sync.py @@ -0,0 +1,463 @@ +"""Durable Desktop command Inbox and independent command-result sync lane.""" + +from __future__ import annotations + +import asyncio +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") + + +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]] = 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: + key = (self._base(configuration), configuration.desktop_instance_id) + if key in self._registered: + return + async with self._lock: + if key in self._registered: + return + await self._request( + "POST", + f"{key[0]}/devices/register", + configuration, + payload={ + "capabilities": { + "run_event_sync": 1, + "command_inbox": 1, + "command_result_sync": 1, + } + }, + ) + 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(), + }, + ) + + 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={ + "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 + + 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) + ), + 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, True + 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 = ( + 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 ( + httpx.HTTPError, + RunEventSyncHttpError, + RunEventSyncProtocolError, + ): + logger.exception( + "Command receipt confirmation failed", + extra={"command_id": command_id}, + ) + may_execute = ( + 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")) + + async def drain_once(self) -> int: + configuration = self._configuration + if configuration is None: + return 0 + pulled = await self._transport.pull_pending( + configuration, limit=self._max_commands + ) + for item in pulled: + 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 + ), + } + ) + for command in await asyncio.to_thread( + self._journal.list_reconcilable_commands, + limit=self._max_commands, + ): + if command.receipt_status == "pending": + await self.confirm_receipt(command.command_id) + + 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 len(pulled) + results = await asyncio.gather( + *(self._sync_batch(batch, configuration) for batch in batches) + ) + self.notify() + return len(pulled) + sum(results) + + 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() diff --git a/backend/app/run_sync/runtime.py b/backend/app/run_sync/runtime.py index fa5735b42..ff9eb0283 100644 --- a/backend/app/run_sync/runtime.py +++ b/backend/app/run_sync/runtime.py @@ -12,10 +12,15 @@ from app.run_sync.cloud_sync import ( CloudSyncWorker, HttpRunEventSyncTransport, ) +from app.run_sync.command_sync import ( + CommandControlWorker, + HttpCommandSyncTransport, +) logger = logging.getLogger("run_sync.runtime") _default_worker: CloudSyncWorker | None = None +_default_command_worker: CommandControlWorker | None = None _worker_loop: asyncio.AbstractEventLoop | None = None @@ -42,7 +47,7 @@ def configure_default_cloud_sync_worker( authorization: str | None, desktop_instance_id: str | None, ) -> bool: - global _default_worker, _worker_loop + global _default_command_worker, _default_worker, _worker_loop endpoint = _sync_endpoint(server_url) auth = str(authorization or "").strip() instance_id = str( @@ -64,16 +69,22 @@ def configure_default_cloud_sync_worker( ) _worker_loop = loop _default_worker.start() + _default_command_worker = CommandControlWorker( + get_default_run_journal(), + HttpCommandSyncTransport(), + ) + _default_command_worker.start() elif _worker_loop is not loop: logger.error("Refusing to move CloudSyncWorker across event loops") return False - _default_worker.configure( - CloudSyncConfiguration( - endpoint_url=endpoint, - authorization=auth, - desktop_instance_id=instance_id, - ) + configuration = CloudSyncConfiguration( + endpoint_url=endpoint, + authorization=auth, + desktop_instance_id=instance_id, ) + _default_worker.configure(configuration) + if _default_command_worker is not None: + _default_command_worker.configure(configuration) return True @@ -83,12 +94,28 @@ def notify_default_cloud_sync_worker() -> None: if worker is None or loop is None or loop.is_closed(): return loop.call_soon_threadsafe(worker.notify) + if _default_command_worker is not None: + loop.call_soon_threadsafe(_default_command_worker.notify) + + +async def persist_and_confirm_remote_command( + command: dict, +) -> tuple[object, bool]: + worker = _default_command_worker + if worker is None: + raise RuntimeError("Command control worker is not configured") + record = await worker.persist_command(command) + return await worker.confirm_receipt(record.command_id) async def close_default_cloud_sync_worker() -> None: - global _default_worker, _worker_loop + global _default_command_worker, _default_worker, _worker_loop worker = _default_worker + command_worker = _default_command_worker _default_worker = None + _default_command_worker = None _worker_loop = None if worker is not None: await worker.close() + if command_worker is not None: + await command_worker.close() diff --git a/backend/tests/app/run_journal/test_command_inbox.py b/backend/tests/app/run_journal/test_command_inbox.py new file mode 100644 index 000000000..aad389c05 --- /dev/null +++ b/backend/tests/app/run_journal/test_command_inbox.py @@ -0,0 +1,155 @@ +from __future__ import annotations + +import pytest + +from app.run_journal import ( + IdempotencyConflictError, + OutboxLeaseLostError, + SQLiteRunJournal, +) + + +def _persist(journal: SQLiteRunJournal, command_id: str = "command-1"): + return journal.persist_remote_command( + command_id=command_id, + session_id="session-1", + user_id=7, + project_id="project-1", + run_id=None, + route_version=1, + command_type="user_message", + payload={"id": command_id, "type": "user_message", "payload": {}}, + expires_at=200.0, + receipt_grace_until=230.0, + requires_online_receipt_confirmation=False, + receipt_event_id=f"{command_id}:receipt", + now=100.0, + ) + + +def test_inbox_commit_also_persists_receipt_outbox(tmp_path): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + command = _persist(journal) + assert command.state == "received" + assert command.receipt_status == "pending" + + batches = journal.claim_command_result_batches(now=100.0) + assert len(batches) == 1 + assert batches[0].command_id == "command-1" + assert [event.event_type for event in batches[0].events] == [ + "receipt.durably_received" + ] + assert batches[0].events[0].command_event_sequence == 1 + + +def test_duplicate_command_requires_identical_canonical_envelope(tmp_path): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + first = _persist(journal) + duplicate = _persist(journal) + assert duplicate.receipt_event_id == first.receipt_event_id + with pytest.raises(IdempotencyConflictError): + journal.persist_remote_command( + command_id="command-1", + session_id="session-1", + user_id=7, + project_id="project-1", + run_id=None, + route_version=1, + command_type="user_message", + payload={"changed": True}, + expires_at=200.0, + receipt_grace_until=230.0, + requires_online_receipt_confirmation=False, + now=100.0, + ) + + +def test_startup_reconciliation_rescans_received_and_dispatched(tmp_path): + path = tmp_path / "journal.sqlite3" + with SQLiteRunJournal(path) as journal: + _persist(journal) + journal.mark_command_dispatched("command-1", now=101.0) + + with SQLiteRunJournal(path) as reopened: + pending = reopened.list_reconcilable_commands() + assert [record.command_id for record in pending] == ["command-1"] + assert pending[0].state == "dispatched" + assert pending[0].dispatch_attempt_count == 1 + + +def test_admission_and_execution_share_command_fifo_not_run_fifo(tmp_path): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + _persist(journal) + accepted = journal.append_command_result( + "command-1", + event_type="admission.accepted", + event_id="command-1:admission", + occurred_at=101.0, + ) + assert [ + command.command_id + for command in journal.list_reconcilable_commands() + ] == ["command-1"] + completed = journal.append_command_result( + "command-1", + event_type="execution.completed", + event_id="command-1:result", + payload={"result": {"ok": True}}, + occurred_at=102.0, + ) + assert accepted.command_event_sequence == 2 + assert completed.command_event_sequence == 3 + assert journal.get_remote_command("command-1").state == "completed" + + batch = journal.claim_command_result_batches(now=102.0)[0] + assert [event.command_event_sequence for event in batch.events] == [ + 1, + 2, + 3, + ] + journal.mark_command_result_batch_sent(batch, now=103.0) + assert journal.claim_command_result_batches(now=104.0) == [] + + +def test_poison_event_blocks_only_its_command_lane(tmp_path): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + _persist(journal, "command-1") + _persist(journal, "command-2") + batches = journal.claim_command_result_batches( + now=100.0, max_commands=2 + ) + first = next( + batch for batch in batches if batch.command_id == "command-1" + ) + second = next( + batch for batch in batches if batch.command_id == "command-2" + ) + journal.block_command_result_batch( + first, + failed_event_id=first.events[0].event_id, + error="poison", + now=101.0, + ) + journal.retry_command_result_batch( + second, + error="temporary", + next_attempt_at=102.0, + now=101.0, + ) + ready = journal.claim_command_result_batches(now=102.0) + assert [batch.command_id for batch in ready] == ["command-2"] + + +def test_stale_command_outbox_worker_cannot_overwrite_new_lease(tmp_path): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + _persist(journal) + old = journal.claim_command_result_batches( + now=100.0, lease_seconds=1.0 + )[0] + new = journal.claim_command_result_batches( + now=102.0, lease_seconds=30.0 + )[0] + assert old.lease_token != new.lease_token + with pytest.raises(OutboxLeaseLostError): + journal.mark_command_result_batch_sent(old, now=103.0) + journal.mark_command_result_batch_sent(new, now=103.0) diff --git a/backend/tests/app/run_sync/test_command_sync.py b/backend/tests/app/run_sync/test_command_sync.py new file mode 100644 index 000000000..5fe2652a4 --- /dev/null +++ b/backend/tests/app/run_sync/test_command_sync.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from app.run_journal import SQLiteRunJournal +from app.run_sync.cloud_sync import CloudSyncConfiguration +from app.run_sync.command_sync import CommandControlWorker + + +class FakeCommandTransport: + def __init__(self) -> None: + self.pending: list[dict[str, Any]] = [] + self.confirmed: list[str] = [] + self.ingested: list[Any] = [] + + async def pull_pending(self, _configuration, *, limit): + return self.pending[:limit] + + async def confirm_receipt(self, _configuration, command): + self.confirmed.append(command.command_id) + return { + "result": "confirmed", + "receipt_state": "durably_received", + "may_execute": True, + } + + async def ingest_events(self, _configuration, batch): + self.ingested.append(batch) + return {"expected_next_desktop_event_sequence": len(batch.events) + 1} + + async def close(self): + return None + + +def _configuration() -> CloudSyncConfiguration: + return CloudSyncConfiguration( + endpoint_url="https://example.test/api/v1/sync/events:ingest", + authorization="Bearer token", + desktop_instance_id="device-1", + ) + + +def _command() -> dict[str, Any]: + return { + "id": "command-1", + "session_id": "session-1", + "user_id": 7, + "project_id": "project-1", + "route_version": 1, + "type": "user_message", + "payload": {"content": "hello"}, + "expires_at": "2030-01-01T00:00:00+00:00", + "receipt_grace_until": "2030-01-01T00:00:30+00:00", + "requires_online_receipt_confirmation": False, + } + + +@pytest.mark.asyncio +async def test_worker_confirms_receipt_and_drains_independent_lane(tmp_path): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + transport = FakeCommandTransport() + worker = CommandControlWorker(journal, transport) + worker.configure(_configuration()) + record = await worker.persist_command(_command()) + + confirmed, may_execute = await worker.confirm_receipt( + record.command_id + ) + assert may_execute is True + assert confirmed.receipt_status == "confirmed" + assert transport.confirmed == ["command-1"] + + assert await worker.drain_once() == 1 + assert len(transport.ingested) == 1 + assert transport.ingested[0].events[0].event_type == ( + "receipt.durably_received" + ) + assert journal.claim_command_result_batches() == [] + await worker.close() + + +@pytest.mark.asyncio +async def test_high_risk_command_does_not_execute_without_cloud_config( + tmp_path, +): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + transport = FakeCommandTransport() + worker = CommandControlWorker(journal, transport) + command = _command() + command["requires_online_receipt_confirmation"] = True + record = await worker.persist_command(command) + + _record, may_execute = await worker.confirm_receipt(record.command_id) + assert may_execute is False + await worker.close() diff --git a/src/hooks/useRemoteControlBridge.ts b/src/hooks/useRemoteControlBridge.ts index 091da58ba..619d82089 100644 --- a/src/hooks/useRemoteControlBridge.ts +++ b/src/hooks/useRemoteControlBridge.ts @@ -12,7 +12,7 @@ // limitations under the License. // ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= -import { getBaseURL, proxyFetchGet } from '@/api/http'; +import { getBaseURL, proxyFetchGet, proxyFetchPost } from '@/api/http'; import { isDesktop } from '@/client/platform'; import { getRemoteControlDesktopInstanceId, @@ -51,6 +51,10 @@ type RemoteCommand = { type: string; payload: Record; next_task_id?: string | null; + route_version?: number; + expires_at?: string; + receipt_grace_until?: string; + requires_online_receipt_confirmation?: boolean; }; type CacheEntry = @@ -891,6 +895,65 @@ export function useRemoteControlBridge(token: string | null | undefined) { return executeRemoteCommand(command, token); }; + const persistCommandAndExecute = async ( + command: RemoteCommand + ): Promise => { + const persisted = await proxyFetchPost( + '/remote-control/commands/inbox', + command, + brainHeaders(command) + ); + send({ + type: 'command_delivered', + command_id: command.id, + }); + if (!persisted?.may_execute) { + await proxyFetchPost( + `/remote-control/commands/${encodeURIComponent(command.id)}/admission`, + { + status: 'rejected', + event_id: `${command.id}:admission`, + reason: 'expired_or_receipt_not_confirmed', + }, + brainHeaders(command) + ); + return { + type: 'command_ack', + command_id: command.id, + status: 'failed', + error_code: 'COMMAND_RECEIPT_NOT_CONFIRMED', + error: 'Command expired or could not pass its receipt gate', + }; + } + + await proxyFetchPost( + `/remote-control/commands/${encodeURIComponent(command.id)}/admission`, + { + status: 'accepted', + event_id: `${command.id}:admission`, + }, + brainHeaders(command) + ); + let ack: BridgeAck; + try { + ack = await executeCommand(command); + } catch (error) { + ack = commandErrorAck(command.id, error); + } + await proxyFetchPost( + `/remote-control/commands/${encodeURIComponent(command.id)}/result`, + { + status: ack.status === 'acknowledged' ? 'completed' : 'failed', + event_id: `${command.id}:result`, + result: ack.result || {}, + error_code: ack.error_code, + error: ack.error, + }, + brainHeaders(command) + ); + return ack; + }; + const handleCommand = (command: RemoteCommand) => { console.info('[RemoteControlBridge][RC-TRACE] command received', { command_id: command.id, @@ -924,12 +987,7 @@ export function useRemoteControlBridge(token: string | null | undefined) { cache.set(command.id, { state: 'in_progress', promise }); trimCache(cache); - send({ - type: 'command_delivered', - command_id: command.id, - }); - - executeCommand(command) + persistCommandAndExecute(command) .then(resolveAck!) .catch((error) => resolveAck!(commandErrorAck(command.id, error))); @@ -950,6 +1008,26 @@ export function useRemoteControlBridge(token: string | null | undefined) { }); }; + const replayDurableInbox = async () => { + try { + const response = await proxyFetchGet( + '/remote-control/commands/inbox/pending', + { limit: 100 } + ); + const items = Array.isArray(response?.items) ? response.items : []; + for (const item of items) { + if (item?.payload?.id) { + handleCommand(item.payload as RemoteCommand); + } + } + } catch (error) { + console.warn( + '[RemoteControlBridge] Durable Inbox reconciliation failed', + error + ); + } + }; + const connect = async () => { const url = await getRemoteControlWebSocketUrl( '/api/v1/remote-control/bridge/subscribe' @@ -988,6 +1066,7 @@ export function useRemoteControlBridge(token: string | null | undefined) { { desktop_instance_id: desktopInstanceId } ); setRemoteControlBridgeConnected(true); + void replayDurableInbox(); return; } if (message?.type === 'auth_expired') {