vrr/eigent
2
0
Fork 0
mirror of https://github.com/eigent-ai/eigent.git synced 2026-08-25 16:42:11 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

feat: add durable remote command inbox

Browse source
This commit is contained in:
4pmtong 2026-08-05 18:36:24 +08:00
parent c6ceaa1ea7
commit 57778d84fe
11 changed files with 1767 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

156
backend/app/controller/remote_command_controller.py Normal file
View file

@ -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)

6
backend/app/router.py
View file

@ -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"],

6
backend/app/run_journal/__init__.py
View file

@ -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",

40
backend/app/run_journal/models.py
View file

@ -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, ...]

652
backend/app/run_journal/store.py
View file

@ -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(

81
backend/app/run_sync/cloud_sync.py
View file

@ -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()

463
backend/app/run_sync/command_sync.py Normal file
View file

@ -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()

43
backend/app/run_sync/runtime.py
View file

@ -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()

155
backend/tests/app/run_journal/test_command_inbox.py Normal file
View file

@ -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)

97
backend/tests/app/run_sync/test_command_sync.py Normal file
View file

@ -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()

93
src/hooks/useRemoteControlBridge.ts
View file

@ -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<string, any>;
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<BridgeAck> => {
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') {