From d86d057b530342f11bc7a20d96ec4894513f27ec Mon Sep 17 00:00:00 2001 From: 4pmtong Date: Wed, 5 Aug 2026 23:14:31 +0800 Subject: [PATCH] fix: make remote commands crash-safe --- .../controller/remote_command_controller.py | 10 +- backend/app/run_journal/store.py | 40 +++- backend/app/run_sync/cloud_sync.py | 39 ++-- backend/app/run_sync/command_sync.py | 144 ++++++++----- .../app/run_journal/test_command_inbox.py | 56 +++++ backend/tests/app/run_sync/test_cloud_sync.py | 25 +++ .../tests/app/run_sync/test_command_sync.py | 72 ++++++- src/hooks/useRemoteControlBridge.test.ts | 45 ++++ src/hooks/useRemoteControlBridge.ts | 195 +++++++++++++++++- 9 files changed, 546 insertions(+), 80 deletions(-) create mode 100644 src/hooks/useRemoteControlBridge.test.ts diff --git a/backend/app/controller/remote_command_controller.py b/backend/app/controller/remote_command_controller.py index 9015f71f..5e03f995 100644 --- a/backend/app/controller/remote_command_controller.py +++ b/backend/app/controller/remote_command_controller.py @@ -94,7 +94,15 @@ async def persist_command_inbox(body: RemoteCommandInboxIn): raise HTTPException( status_code=422, detail=f"missing command field: {exc.args[0]}" ) from exc - return {"command": asdict(record), "may_execute": may_execute} + execution_event = await asyncio.to_thread( + get_default_run_journal().get_latest_command_execution_result, + record.command_id, + ) + return { + "command": asdict(record), + "may_execute": may_execute, + "execution_event": asdict(execution_event) if execution_event else None, + } @router.get("/inbox/pending") diff --git a/backend/app/run_journal/store.py b/backend/app/run_journal/store.py index c70d676b..8ec293a4 100644 --- a/backend/app/run_journal/store.py +++ b/backend/app/run_journal/store.py @@ -2042,6 +2042,24 @@ class SQLiteRunJournal: ).fetchone() return self._command_from_row(row) if row is not None else None + def get_latest_command_execution_result( + self, command_id: str + ) -> CommandResultEvent | None: + """Return the durable terminal execution result used for ACK replay.""" + + with self._lock: + row = self._connection.execute( + """ + SELECT * FROM command_result_events + WHERE command_id = ? + AND event_type IN ('execution.completed', 'execution.failed') + ORDER BY command_event_sequence DESC + LIMIT 1 + """, + (command_id,), + ).fetchone() + return self._command_event_from_row(row) if row is not None else None + def list_reconcilable_commands( self, *, limit: int = 100 ) -> list[RemoteCommandInboxRecord]: @@ -2277,13 +2295,21 @@ class SQLiteRunJournal: ) 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 + WITH heads AS ( + SELECT command_id, + MIN(command_event_sequence) AS head_sequence + FROM command_result_outbox + WHERE status != 'sent' + GROUP BY command_id + ) + SELECT heads.command_id, heads.head_sequence + FROM heads + JOIN command_result_outbox AS head + ON head.command_id = heads.command_id + AND head.command_event_sequence = heads.head_sequence + WHERE head.status = 'pending' + AND head.next_attempt_at <= ? + ORDER BY head.updated_at, heads.command_id LIMIT ? """, (timestamp, max_commands), diff --git a/backend/app/run_sync/cloud_sync.py b/backend/app/run_sync/cloud_sync.py index 06dfaeb4..a8032cca 100644 --- a/backend/app/run_sync/cloud_sync.py +++ b/backend/app/run_sync/cloud_sync.py @@ -40,6 +40,10 @@ class RunEventSyncProtocolError(RuntimeError): pass +class RunSyncInfrastructureError(RuntimeError): + """A device/route control-plane failure, never a poison Run event.""" + + class RunEventSyncTransport(Protocol): async def ingest( self, @@ -124,20 +128,31 @@ class HttpRunEventSyncTransport: 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}}, - ) + try: + await self._json_request( + "POST", + f"{base}/devices/register", + configuration, + { + "capabilities": { + "run_event_sync": 1, + "command_sync": 1, + } + }, + ) + except RunEventSyncHttpError as exc: + raise RunSyncInfrastructureError(str(exc)) from exc 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, - {}, - ) + try: + await self._json_request( + "PUT", + f"{base}/projects/{project_id}/execution-route", + configuration, + {}, + ) + except RunEventSyncHttpError as exc: + raise RunSyncInfrastructureError(str(exc)) from exc self._claimed_routes.add(route_key) async def ingest( diff --git a/backend/app/run_sync/command_sync.py b/backend/app/run_sync/command_sync.py index 854f4cd1..64d19d96 100644 --- a/backend/app/run_sync/command_sync.py +++ b/backend/app/run_sync/command_sync.py @@ -24,6 +24,12 @@ from app.run_sync.cloud_sync import ( logger = logging.getLogger("command_sync") +_EXECUTABLE_INBOX_STATES = frozenset({"received", "dispatched"}) + + +class CommandSyncInfrastructureError(RuntimeError): + """Device registration failure, not a poison command result event.""" + def _timestamp(value: str | float | int) -> float: if isinstance(value, (float, int)): @@ -102,18 +108,21 @@ class HttpCommandSyncTransport: 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, - } - }, - ) + try: + await self._request( + "POST", + f"{key[0]}/devices/register", + configuration, + payload={ + "capabilities": { + "run_event_sync": 1, + "command_inbox": 1, + "command_result_sync": 1, + } + }, + ) + except RunEventSyncHttpError as exc: + raise CommandSyncInfrastructureError(str(exc)) from exc self._registered.add(key) async def pull_pending( @@ -280,7 +289,7 @@ class CommandControlWorker: if command is None: raise KeyError(command_id) if command.receipt_status == "confirmed": - return command, True + return command, command.state in _EXECUTABLE_INBOX_STATES if command.receipt_status == "expired_late": return command, False configuration = self._configuration @@ -288,7 +297,8 @@ class CommandControlWorker: # 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 + command.state in _EXECUTABLE_INBOX_STATES + and not command.requires_online_receipt_confirmation and time.time() <= command.expires_at + 5 ) return command, may_execute @@ -297,6 +307,7 @@ class CommandControlWorker: configuration, command ) except ( + CommandSyncInfrastructureError, httpx.HTTPError, RunEventSyncHttpError, RunEventSyncProtocolError, @@ -306,7 +317,8 @@ class CommandControlWorker: extra={"command_id": command_id}, ) may_execute = ( - not command.requires_online_receipt_confirmation + command.state in _EXECUTABLE_INBOX_STATES + and not command.requires_online_receipt_confirmation and time.time() <= command.expires_at + 5 ) return command, may_execute @@ -317,59 +329,87 @@ class CommandControlWorker: command_id, status, ) - return updated, bool(response.get("may_execute")) + return updated, bool(response.get("may_execute")) and ( + updated.state in _EXECUTABLE_INBOX_STATES + ) async def drain_once(self) -> int: configuration = self._configuration if configuration is None: return 0 - 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) + synced = await self._drain_outbound(configuration) + pulled_count = await self._pull_and_reconcile(configuration) + return synced + pulled_count + async def _drain_outbound( + self, configuration: CloudSyncConfiguration + ) -> int: batches = await asyncio.to_thread( self._journal.claim_command_result_batches, max_commands=self._max_commands, batch_size=self._batch_size, ) if not batches: - return len(pulled) + return 0 results = await asyncio.gather( *(self._sync_batch(batch, configuration) for batch in batches) ) self.notify() - return len(pulled) + sum(results) + return sum(results) + + async def _pull_and_reconcile( + self, configuration: CloudSyncConfiguration + ) -> int: + pulled_count = 0 + try: + pulled = await self._transport.pull_pending( + configuration, limit=self._max_commands + ) + except Exception: + logger.exception("Remote command pull failed; outbound lane remains live") + pulled = [] + for item in pulled: + try: + await self.persist_command( + { + "id": item["command_id"], + "session_id": item["session_id"], + "user_id": item["user_id"], + "project_id": item["project_id"], + "run_id": item.get("run_id"), + "route_version": item["route_version"], + "type": item["command_type"], + "payload": item.get("payload") or {}, + "space_id": item.get("space_id"), + "target_task_id": item.get("run_id"), + "target_brain_session_id": item.get( + "target_brain_session_id" + ), + "source_channel": item.get("source_channel"), + "next_task_id": item.get("next_task_id"), + "expires_at": item["expires_at"], + "receipt_grace_until": item["receipt_grace_until"], + "requires_online_receipt_confirmation": item.get( + "requires_online_receipt_confirmation", False + ), + } + ) + pulled_count += 1 + except Exception: + logger.exception("Ignoring malformed or conflicting remote command") + for command in await asyncio.to_thread( + self._journal.list_reconcilable_commands, + limit=self._max_commands, + ): + if command.receipt_status == "pending": + try: + await self.confirm_receipt(command.command_id) + except Exception: + logger.exception( + "Remote command receipt reconciliation failed", + extra={"command_id": command.command_id}, + ) + return pulled_count async def _sync_batch( self, diff --git a/backend/tests/app/run_journal/test_command_inbox.py b/backend/tests/app/run_journal/test_command_inbox.py index aad389c0..12064e99 100644 --- a/backend/tests/app/run_journal/test_command_inbox.py +++ b/backend/tests/app/run_journal/test_command_inbox.py @@ -140,6 +140,62 @@ def test_poison_event_blocks_only_its_command_lane(tmp_path): assert [batch.command_id for batch in ready] == ["command-2"] +def test_tail_poison_does_not_block_prior_command_prefix(tmp_path): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + _persist(journal) + journal.append_command_result( + "command-1", + event_type="admission.accepted", + event_id="command-1:admission", + occurred_at=101.0, + ) + journal.append_command_result( + "command-1", + event_type="execution.completed", + event_id="command-1:result", + payload={"result": {"ok": True}}, + occurred_at=102.0, + ) + batch = journal.claim_command_result_batches(now=102.0)[0] + journal.block_command_result_batch( + batch, + failed_event_id="command-1:result", + error="poison tail", + now=103.0, + ) + + prefix = journal.claim_command_result_batches(now=104.0)[0] + + assert [event.event_id for event in prefix.events] == [ + "command-1:receipt", + "command-1:admission", + ] + + +def test_latest_execution_result_is_available_for_durable_ack_replay(tmp_path): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + _persist(journal) + journal.append_command_result( + "command-1", + event_type="admission.accepted", + event_id="command-1:admission", + occurred_at=101.0, + ) + journal.append_command_result( + "command-1", + event_type="execution.failed", + event_id="command-1:result", + payload={"error_code": "FAILED", "error": "boom"}, + occurred_at=102.0, + ) + + result = journal.get_latest_command_execution_result("command-1") + + assert result is not None + assert result.event_type == "execution.failed" + assert result.payload["error"] == "boom" + + def test_stale_command_outbox_worker_cannot_overwrite_new_lease(tmp_path): with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: _persist(journal) diff --git a/backend/tests/app/run_sync/test_cloud_sync.py b/backend/tests/app/run_sync/test_cloud_sync.py index 55e6827a..1406c2f6 100644 --- a/backend/tests/app/run_sync/test_cloud_sync.py +++ b/backend/tests/app/run_sync/test_cloud_sync.py @@ -3,12 +3,14 @@ from __future__ import annotations import asyncio from typing import Any +import httpx import pytest from app.run_journal import RunEventDraft, SQLiteRunJournal from app.run_sync import ( CloudSyncConfiguration, CloudSyncWorker, + HttpRunEventSyncTransport, RunEventSyncHttpError, ) @@ -221,3 +223,26 @@ async def test_invalid_success_response_is_retried(journal): assert pending[0].attempt_count == 1 assert "scope does not match" in pending[0].last_error await worker.close() + + +@pytest.mark.asyncio +async def test_device_registration_conflict_retries_without_poisoning_event( + journal, +): + _append(journal, "run-1") + + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path.endswith("/devices/register") + return httpx.Response(409, json={"detail": "route temporarily owned"}) + + transport = HttpRunEventSyncTransport( + transport=httpx.MockTransport(handler) + ) + worker = _worker(journal, transport) + + assert await worker.drain_once() == 0 + pending = journal.list_pending_outbox(now=float("inf")) + assert pending[0].status == "pending" + assert pending[0].attempt_count == 1 + assert "route temporarily owned" in (pending[0].last_error or "") + await worker.close() diff --git a/backend/tests/app/run_sync/test_command_sync.py b/backend/tests/app/run_sync/test_command_sync.py index 00c5c2a8..8e6c4f63 100644 --- a/backend/tests/app/run_sync/test_command_sync.py +++ b/backend/tests/app/run_sync/test_command_sync.py @@ -2,11 +2,15 @@ from __future__ import annotations from typing import Any +import httpx import pytest from app.run_journal import InvalidRunTransitionError, SQLiteRunJournal from app.run_sync.cloud_sync import CloudSyncConfiguration -from app.run_sync.command_sync import CommandControlWorker +from app.run_sync.command_sync import ( + CommandControlWorker, + HttpCommandSyncTransport, +) class FakeCommandTransport: @@ -14,8 +18,11 @@ class FakeCommandTransport: self.pending: list[dict[str, Any]] = [] self.confirmed: list[str] = [] self.ingested: list[Any] = [] + self.pull_error: Exception | None = None async def pull_pending(self, _configuration, *, limit): + if self.pull_error is not None: + raise self.pull_error return self.pending[:limit] async def confirm_receipt(self, _configuration, command): @@ -97,6 +104,69 @@ async def test_high_risk_command_does_not_execute_without_cloud_config( await worker.close() +@pytest.mark.asyncio +async def test_terminal_command_receipt_replay_never_executes_again(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()) + await worker.confirm_receipt(record.command_id) + journal.append_command_result( + record.command_id, + event_type="admission.accepted", + event_id="accepted", + ) + journal.append_command_result( + record.command_id, + event_type="execution.completed", + event_id="completed", + payload={"result": {"ok": True}}, + ) + + replayed, may_execute = await worker.confirm_receipt(record.command_id) + + assert replayed.state == "completed" + assert may_execute is False + assert transport.confirmed == [record.command_id] + await worker.close() + + +@pytest.mark.asyncio +async def test_inbound_pull_failure_does_not_starve_outbound_results(tmp_path): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + transport = FakeCommandTransport() + worker = CommandControlWorker(journal, transport) + worker.configure(_configuration()) + await worker.persist_command(_command()) + transport.pull_error = ValueError("malformed pending command") + + assert await worker.drain_once() == 1 + assert len(transport.ingested) == 1 + assert journal.claim_command_result_batches() == [] + await worker.close() + + +@pytest.mark.asyncio +async def test_device_registration_error_retries_command_lane(tmp_path): + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path.endswith("/devices/register") + return httpx.Response(409, json={"detail": "device route conflict"}) + + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + transport = HttpCommandSyncTransport( + transport=httpx.MockTransport(handler) + ) + worker = CommandControlWorker(journal, transport) + worker.configure(_configuration()) + await worker.persist_command(_command()) + + assert await worker.drain_once() == 0 + batch = journal.claim_command_result_batches(now=float("inf"))[0] + assert batch.attempt_count == 1 + await worker.close() + + def test_command_inbox_terminal_state_cannot_move_backwards(tmp_path): with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: record = journal.persist_remote_command( diff --git a/src/hooks/useRemoteControlBridge.test.ts b/src/hooks/useRemoteControlBridge.test.ts new file mode 100644 index 00000000..97bb0102 --- /dev/null +++ b/src/hooks/useRemoteControlBridge.test.ts @@ -0,0 +1,45 @@ +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + +import { ackFromDurableExecution } from './useRemoteControlBridge'; + +describe('remote command durable ACK replay', () => { + it('replays the canonical completed outcome without executing again', () => { + expect( + ackFromDurableExecution('command-1', { + event_type: 'execution.completed', + payload: { result: { run_id: 'run-1' } }, + }) + ).toEqual({ + type: 'command_ack', + command_id: 'command-1', + status: 'acknowledged', + result: { run_id: 'run-1' }, + replayed_from_cache: true, + }); + }); + + it('replays the canonical failure rather than an upload error', () => { + expect( + ackFromDurableExecution('command-1', { + event_type: 'execution.failed', + payload: { error_code: 'TOOL_FAILED', error: 'original failure' }, + }) + ).toMatchObject({ + status: 'failed', + error_code: 'TOOL_FAILED', + error: 'original failure', + }); + }); +}); diff --git a/src/hooks/useRemoteControlBridge.ts b/src/hooks/useRemoteControlBridge.ts index 619d8208..1cb19e29 100644 --- a/src/hooks/useRemoteControlBridge.ts +++ b/src/hooks/useRemoteControlBridge.ts @@ -36,6 +36,23 @@ type BridgeAck = { replayed_from_cache?: boolean; }; +type CommandResultBody = { + status: 'completed' | 'failed'; + event_id: string; + result: Record; + error_code?: string; + error?: string; +}; + +type DurableCommandEvent = { + event_type?: string; + payload?: { + result?: Record; + error_code?: string; + error?: string; + }; +}; + type RemoteCommand = { id: string; session_id: string; @@ -65,6 +82,7 @@ const CACHE_LIMIT = 200; const COMMAND_TIMEOUT_MS = 10000; const RATE_LIMIT_WINDOW_MS = 1000; const RATE_LIMIT_MAX_COMMANDS = 5; +const PENDING_COMMAND_RESULTS_KEY = 'eigent:remote-command-results:v1'; const remoteHistoryHydrationInFlight = new Set(); const BRIDGE_CAPABILITIES = { bridge_version: 1, @@ -85,6 +103,82 @@ const BRIDGE_CAPABILITIES = { ], }; +type PendingCommandResult = { + command: RemoteCommand; + body: CommandResultBody; +}; + +function readPendingCommandResults(): PendingCommandResult[] { + try { + const value = JSON.parse( + window.localStorage.getItem(PENDING_COMMAND_RESULTS_KEY) || '[]' + ); + return Array.isArray(value) ? value : []; + } catch { + return []; + } +} + +function writePendingCommandResults(items: PendingCommandResult[]) { + try { + window.localStorage.setItem( + PENDING_COMMAND_RESULTS_KEY, + JSON.stringify(items) + ); + } catch (error) { + console.warn( + '[RemoteControlBridge] Could not persist pending command result', + error + ); + } +} + +function queuePendingCommandResult(item: PendingCommandResult) { + const items = readPendingCommandResults().filter( + (candidate) => candidate.command.id !== item.command.id + ); + items.push(item); + writePendingCommandResults(items); +} + +function removePendingCommandResult(commandId: string) { + writePendingCommandResults( + readPendingCommandResults().filter( + (candidate) => candidate.command.id !== commandId + ) + ); +} + +export function ackFromDurableExecution( + commandId: string, + event?: DurableCommandEvent | null +): BridgeAck | null { + if (!event?.event_type) { + return null; + } + const payload = event.payload || {}; + if (event.event_type === 'execution.completed') { + return { + type: 'command_ack', + command_id: commandId, + status: 'acknowledged', + result: payload.result || {}, + replayed_from_cache: true, + }; + } + if (event.event_type === 'execution.failed') { + return { + type: 'command_ack', + command_id: commandId, + status: 'failed', + error_code: payload.error_code || 'COMMAND_FAILED', + error: payload.error || 'Remote command failed', + replayed_from_cache: true, + }; + } + return null; +} + function trimCache(cache: Map) { if (cache.size <= CACHE_LIMIT) { return; @@ -895,6 +989,32 @@ export function useRemoteControlBridge(token: string | null | undefined) { return executeRemoteCommand(command, token); }; + const persistCommandResult = async ( + command: RemoteCommand, + body: CommandResultBody + ) => { + queuePendingCommandResult({ command, body }); + await proxyFetchPost( + `/remote-control/commands/${encodeURIComponent(command.id)}/result`, + body, + brainHeaders(command) + ); + removePendingCommandResult(command.id); + }; + + const flushPendingCommandResults = async () => { + for (const item of readPendingCommandResults()) { + try { + await persistCommandResult(item.command, item.body); + } catch (error) { + console.warn( + '[RemoteControlBridge] Pending command result remains queued', + { command_id: item.command.id, error } + ); + } + } + }; + const persistCommandAndExecute = async ( command: RemoteCommand ): Promise => { @@ -907,6 +1027,61 @@ export function useRemoteControlBridge(token: string | null | undefined) { type: 'command_delivered', command_id: command.id, }); + const durableAck = ackFromDurableExecution( + command.id, + persisted?.execution_event + ); + if (durableAck) { + return durableAck; + } + const inboxState = persisted?.command?.state; + if (inboxState === 'accepted') { + // A previous renderer durably admitted this command. Re-executing after + // restart could duplicate an external side effect, so close the unknown + // outcome explicitly and let the independent result lane retry it. + const unknownAck: BridgeAck = { + type: 'command_ack', + command_id: command.id, + status: 'failed', + error_code: 'COMMAND_OUTCOME_UNKNOWN_AFTER_RESTART', + error: + 'The command was admitted before Desktop restarted; it was not replayed.', + }; + try { + await persistCommandResult(command, { + status: 'failed', + event_id: `${command.id}:result`, + result: {}, + error_code: unknownAck.error_code, + error: unknownAck.error, + }); + } catch (error) { + console.warn( + '[RemoteControlBridge] Outcome-unknown result queued for retry', + error + ); + } + return unknownAck; + } + if (inboxState === 'rejected') { + return { + type: 'command_ack', + command_id: command.id, + status: 'failed', + error_code: 'COMMAND_REJECTED', + error: persisted?.command?.last_error || 'Command was rejected', + replayed_from_cache: true, + }; + } + if (inboxState === 'completed' || inboxState === 'failed') { + return { + type: 'command_ack', + command_id: command.id, + status: 'failed', + error_code: 'COMMAND_RESULT_INTEGRITY_ERROR', + error: 'Durable command state has no replayable execution result', + }; + } if (!persisted?.may_execute) { await proxyFetchPost( `/remote-control/commands/${encodeURIComponent(command.id)}/admission`, @@ -940,17 +1115,23 @@ export function useRemoteControlBridge(token: string | null | undefined) { } catch (error) { ack = commandErrorAck(command.id, error); } - await proxyFetchPost( - `/remote-control/commands/${encodeURIComponent(command.id)}/result`, - { + try { + await persistCommandResult(command, { status: ack.status === 'acknowledged' ? 'completed' : 'failed', event_id: `${command.id}:result`, result: ack.result || {}, error_code: ack.error_code, error: ack.error, - }, - brainHeaders(command) - ); + }); + } catch (error) { + // The execution outcome is authoritative. Keep its exact payload in a + // durable renderer queue and retry it; never replace it with a fake + // execution.failed ACK caused by an upload/notification failure. + console.warn('[RemoteControlBridge] Command result queued for retry', { + command_id: command.id, + error, + }); + } return ack; }; @@ -1066,7 +1247,7 @@ export function useRemoteControlBridge(token: string | null | undefined) { { desktop_instance_id: desktopInstanceId } ); setRemoteControlBridgeConnected(true); - void replayDurableInbox(); + void flushPendingCommandResults().then(replayDurableInbox); return; } if (message?.type === 'auth_expired') {