fix: make remote commands crash-safe

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

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

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(

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,

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)

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

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(

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',
+    });
+  });
+});

src/hooks/useRemoteControlBridge.ts

@@ -36,6 +36,23 @@ type BridgeAck = {
   replayed_from_cache?: boolean;
 };
 
+type CommandResultBody = {
+  status: 'completed' | 'failed';
+  event_id: string;
+  result: Record<string, any>;
+  error_code?: string;
+  error?: string;
+};
+
+type DurableCommandEvent = {
+  event_type?: string;
+  payload?: {
+    result?: Record<string, any>;
+    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<string>();
 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<string, CacheEntry>) {
   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<BridgeAck> => {
@@ -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') {