fix(peer-device): recover from weak network stalls

Author: bobleer

docs/architecture/peer-device-mode.md

@@ -49,13 +49,19 @@ FS) and must not be mixed with Peer Device Mode.
 
 - Controller: `PeerDeviceTransportAdapter` wraps product `invoke` as
   `RemoteCommand::HostInvoke` over `account_device_rpc`.
-- HostInvoke on the controller is **priority-queued** (effectively unbounded
-  concurrency, `i32::MAX` in flight). Session restore / session-list / dialog /
-  workspace-startup commands outrank background `git_*` / `ssh_*` / `lsp_*` /
-  `search_*` / FS / canvas / editor RPCs so hydrate is not starved into relay
-  HTTP 504s. Terminal commands are always interactive priority, and one slot is
-  kept free from low-priority background work so input cannot be trapped behind
-  slow polling requests.
+- HostInvoke on the controller is **priority-queued** with four requests in
+  flight. Session restore / session-list / dialog / workspace-startup commands
+  outrank background `git_*` / `ssh_*` / `lsp_*` / `search_*` / FS / canvas /
+  editor RPCs so hydrate is not starved into relay HTTP 504s. Terminal commands
+  are always interactive priority, and one slot is kept free from normal and
+  low-priority work so input cannot be trapped behind slow polling requests.
+- Idempotent read HostInvokes use a 10s per-attempt deadline and at most two
+  exponential-backoff retries. Mutating commands use a 30s deadline without
+  automatic replay because a timed-out mutation has an unknown outcome. The
+  desktop `account_device_rpc` command enforces the requested deadline around
+  the native HTTP future; the controller's Promise deadline is not merely a UI
+  timer. Failed session-list loads leave the spinner and expose an explicit
+  retry action.
 - While Peer Mode is active, background noise is reduced further:
   - controller-local SSH heartbeats and remote-workspace auto-reconnect pause
   - Git / FilesPanel window-focus refresh pauses
@@ -82,7 +88,10 @@ FS) and must not be mixed with Peer Device Mode.
   controller that attached after turn/round lifecycle events or crossed a
   transient relay gap. The host overlays its authoritative in-memory session
   state onto the persisted view so an executing turn is not misclassified as
-  interrupted history.
+  interrupted history. Continuous host output is checkpointed at least once
+  per 2s coalescing window. A controller accepts an active snapshot before the
+  stale-stream deadline only when its rounds, streams, and tools provably move
+  forward; an older persisted snapshot cannot overwrite newer DeviceEvents.
 - CLI Peer Host forwards only turns submitted through Peer Host and linked
   child turns. A background-result follow-up inherits ownership only when its
   Core-internal metadata identifies the exact tracked parent and source child
@@ -105,9 +114,10 @@ FS) and must not be mixed with Peer Device Mode.
   enqueue attempt. An explicit disconnect still restores the local controller
   UI, but reports a warning when host cancellation was not confirmed. This
   boundary does not change the Relay envelope or add ACK or replay.
-- Relay `POST /api/devices/:id/rpc` waits up to **120s** for the peer response;
-  reverse proxies in front of the relay must use a matching (or higher) read
-  timeout or they will return 504 first.
+- Relay `POST /api/devices/:id/rpc` still permits up to **120s** for generic
+  callers. Peer controllers normally cancel earlier through their per-command
+  10s/30s deadlines; reverse proxies must still accommodate any other caller
+  that relies on the Relay maximum.
 
 ## Workspace directory picking
 

src/apps/desktop/src/api/remote_connect_api.rs

@@ -3405,18 +3405,35 @@ pub async fn account_delete_device(targetDeviceId: String) -> Result<(), String>
 /// which routes it to the target device's WS. The target executes it
 /// and the response is returned (decrypted).
 /// Returns the decrypted response JSON.
+const ACCOUNT_DEVICE_RPC_DEFAULT_TIMEOUT_MS: u64 = 120_000;
+const ACCOUNT_DEVICE_RPC_MIN_TIMEOUT_MS: u64 = 1_000;
+
+fn account_device_rpc_timeout_ms(requested: Option<u64>) -> u64 {
+    requested
+        .unwrap_or(ACCOUNT_DEVICE_RPC_DEFAULT_TIMEOUT_MS)
+        .clamp(
+            ACCOUNT_DEVICE_RPC_MIN_TIMEOUT_MS,
+            ACCOUNT_DEVICE_RPC_DEFAULT_TIMEOUT_MS,
+        )
+}
+
 #[tauri::command]
 pub async fn account_device_rpc(
     target_device_id: String,
     command_json: String,
+    timeout_ms: Option<u64>,
 ) -> Result<String, String> {
     let account_generation = account_context_generation();
     let (session, relay_url) = read_account_context_for_generation(account_generation).await?;
     let client = AccountClient::new();
-    let response = client
-        .device_rpc(&relay_url, &session, &target_device_id, &command_json)
-        .await
-        .map_err(|e| format!("{e}"))?;
+    let timeout_ms = account_device_rpc_timeout_ms(timeout_ms);
+    let response = tokio::time::timeout(
+        std::time::Duration::from_millis(timeout_ms),
+        client.device_rpc(&relay_url, &session, &target_device_id, &command_json),
+    )
+    .await
+    .map_err(|_| format!("device RPC timed out after {timeout_ms}ms"))?
+    .map_err(|e| format!("{e}"))?;
     if !account_context_matches(account_generation, &session.token).await {
         return Err("account context changed".to_string());
     }
@@ -3706,58 +3723,56 @@ async fn account_auto_sync_inner(
     );
 
     let completed = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
-    let upload_outcomes: Vec<Result<(String, String, i64), String>> =
-        stream::iter(pending_uploads)
-            .map(|(session_id, bundle_json, hash)| {
-                let client = AccountClient::new();
-                let relay_url = relay_url.clone();
-                let acct_session = acct_session.clone();
-                let completed = completed.clone();
-                async move {
-                    if ensure_account_auto_sync_current(sync_operation_id).is_err() {
-                        return Err("account sync cancelled".to_string());
-                    }
-                    let result = match await_account_auto_sync(
-                        sync_operation_id,
-                        client.upload_session(&relay_url, &acct_session, &session_id, &bundle_json),
-                    )
-                    .await
-                    {
-                        Ok(result) => result,
-                        Err(e) => return Err(e),
-                    };
-                    match result {
-                        Ok(version) => {
-                            let done =
-                                completed.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
-                            let percent = if upload_total == 0 {
-                                95u8
-                            } else {
-                                20 + ((75 * done) / upload_total) as u8
-                            };
-                            if ensure_account_auto_sync_current(sync_operation_id).is_err() {
-                                return Err("account sync cancelled".to_string());
-                            }
-                            emit_sync_progress(
-                                sync_operation_id,
-                                "exporting_sessions",
-                                percent.min(95),
-                                Some(done),
-                                Some(upload_total),
-                                Some(session_id.as_str()),
-                            );
-                            Ok((session_id, hash, version))
-                        }
-                        Err(e) => {
-                            log::warn!("Auto-sync upload {session_id} failed: {e}");
-                            Err(format!("{session_id}: {e}"))
+    let upload_outcomes: Vec<Result<(String, String, i64), String>> = stream::iter(pending_uploads)
+        .map(|(session_id, bundle_json, hash)| {
+            let client = AccountClient::new();
+            let relay_url = relay_url.clone();
+            let acct_session = acct_session.clone();
+            let completed = completed.clone();
+            async move {
+                if ensure_account_auto_sync_current(sync_operation_id).is_err() {
+                    return Err("account sync cancelled".to_string());
+                }
+                let result = match await_account_auto_sync(
+                    sync_operation_id,
+                    client.upload_session(&relay_url, &acct_session, &session_id, &bundle_json),
+                )
+                .await
+                {
+                    Ok(result) => result,
+                    Err(e) => return Err(e),
+                };
+                match result {
+                    Ok(version) => {
+                        let done = completed.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
+                        let percent = if upload_total == 0 {
+                            95u8
+                        } else {
+                            20 + ((75 * done) / upload_total) as u8
+                        };
+                        if ensure_account_auto_sync_current(sync_operation_id).is_err() {
+                            return Err("account sync cancelled".to_string());
                         }
+                        emit_sync_progress(
+                            sync_operation_id,
+                            "exporting_sessions",
+                            percent.min(95),
+                            Some(done),
+                            Some(upload_total),
+                            Some(session_id.as_str()),
+                        );
+                        Ok((session_id, hash, version))
+                    }
+                    Err(e) => {
+                        log::warn!("Auto-sync upload {session_id} failed: {e}");
+                        Err(format!("{session_id}: {e}"))
                     }
                 }
-            })
-            .buffer_unordered(UPLOAD_CONCURRENCY)
-            .collect()
-            .await;
+            }
+        })
+        .buffer_unordered(UPLOAD_CONCURRENCY)
+        .collect()
+        .await;
 
     ensure_account_auto_sync_current(sync_operation_id)?;
 
@@ -4346,6 +4361,23 @@ mod sync_state_tests {
         );
     }
 
+    #[test]
+    fn device_rpc_timeout_is_bounded_for_peer_requests() {
+        assert_eq!(
+            account_device_rpc_timeout_ms(None),
+            ACCOUNT_DEVICE_RPC_DEFAULT_TIMEOUT_MS
+        );
+        assert_eq!(account_device_rpc_timeout_ms(Some(10_000)), 10_000);
+        assert_eq!(
+            account_device_rpc_timeout_ms(Some(100)),
+            ACCOUNT_DEVICE_RPC_MIN_TIMEOUT_MS
+        );
+        assert_eq!(
+            account_device_rpc_timeout_ms(Some(u64::MAX)),
+            ACCOUNT_DEVICE_RPC_DEFAULT_TIMEOUT_MS
+        );
+    }
+
     #[test]
     fn login_result_exposes_only_an_opaque_pending_owner() {
         let value = serde_json::to_value(AccountLoginResult {

src/web-ui/src/app/components/NavPanel/sections/sessions/SessionsSection.tsx

@@ -187,12 +187,14 @@ const SessionsSection: React.FC<SessionsSectionProps> = ({
     nextCursor?: string;
     hasMore: boolean;
     isLoading: boolean;
+    loadError: boolean;
   }>({
     totalTopLevelCount: null,
     syncedTopLevelCount: null,
     nextCursor: undefined,
     hasMore: false,
     isLoading: false,
+    loadError: false,
   });
   const [openMenuSessionId, setOpenMenuSessionId] = useState<string | null>(null);
   const [sessionMenuPosition, setSessionMenuPosition] = useState<{ top: number; left: number } | null>(null);
@@ -291,6 +293,7 @@ const SessionsSection: React.FC<SessionsSectionProps> = ({
       nextCursor: undefined,
       hasMore: false,
       isLoading: false,
+      loadError: false,
     });
   }, [workspaceId, workspacePath, remoteConnectionId, remoteSshHost]);
 
@@ -302,7 +305,11 @@ const SessionsSection: React.FC<SessionsSectionProps> = ({
 
       const requestId = metadataLoadRequestIdRef.current + 1;
       metadataLoadRequestIdRef.current = requestId;
-      setMetadataPageState(prev => ({ ...prev, isLoading: true }));
+      setMetadataPageState(prev => ({
+        ...prev,
+        isLoading: true,
+        loadError: false,
+      }));
 
       try {
         const page = await flowChatStore.loadSessionMetadataPage(
@@ -326,12 +333,17 @@ const SessionsSection: React.FC<SessionsSectionProps> = ({
             nextCursor: page.nextCursor,
             hasMore: page.hasMore,
             isLoading: false,
+            loadError: false,
           });
         }
         return page;
       } catch (error) {
         if (metadataLoadRequestIdRef.current === requestId) {
-          setMetadataPageState(prev => ({ ...prev, isLoading: false }));
+          setMetadataPageState(prev => ({
+            ...prev,
+            isLoading: false,
+            loadError: true,
+          }));
         }
         log.warn('Failed to load visible session metadata page', { error, workspacePath, cursor, limit });
         return null;
@@ -445,6 +457,7 @@ const SessionsSection: React.FC<SessionsSectionProps> = ({
         nextCursor: undefined,
         hasMore: false,
         isLoading: false,
+        loadError: false,
       });
       if (isVisible && workspacePath) {
         void loadMetadataPage(SESSIONS_LEVEL_0, undefined, 'sessions_nav_post_archive');
@@ -933,6 +946,21 @@ const SessionsSection: React.FC<SessionsSectionProps> = ({
         </div>
       );
     }
+    if (metadataPageState.loadError) {
+      return (
+        <div className="bitfun-nav-panel__inline-list">
+          <button
+            type="button"
+            className="bitfun-nav-panel__inline-action"
+            onClick={() => {
+              void loadInitialMetadataPage('sessions_nav_manual_retry');
+            }}
+          >
+            <span>{t('nav.sessions.loadFailedRetry')}</span>
+          </button>
+        </div>
+      );
+    }
     return null;
   }
 

src/web-ui/src/flow_chat/services/flow-chat-manager/EventHandlerModule.ts

@@ -82,6 +82,7 @@ import {
   scheduleModelResponseStatus,
 } from './RuntimeStatusModule';
 import { requestPeerSessionRefresh } from './PeerSessionRefreshModule';
+import { isPeerDeviceModeActive } from '@/infrastructure/peer-device/peerModeFlag';
 
 const log = createLogger('EventHandlerModule');
 const TURN_COMPLETION_QUIET_WINDOW_MS = 500;
@@ -1698,7 +1699,12 @@ export function processBatchedEvents(
           processNormalTextChunkInternal(context, sessionId, turnId, roundId, text, attemptId, attemptIndex);
         }
 
-        debouncedSaveDialogTurn(context, sessionId, turnId, 2000);
+        // The executing host owns turn persistence. A Peer controller receives
+        // the same chunks for rendering and must not echo a save RPC for every
+        // checkpoint, especially on a weak link.
+        if (!isPeerDeviceModeActive()) {
+          debouncedSaveDialogTurn(context, sessionId, turnId, 2000);
+        }
       } else if (eventType === 'tool:params') {
         const { sessionId, turnId, toolEvent } = payload;
         processToolParamsPartialInternal(sessionId, turnId, toolEvent);

src/web-ui/src/flow_chat/services/flow-chat-manager/PeerSessionRefreshModule.ts

@@ -120,6 +120,7 @@ export function installPeerSessionRefresh(context: FlowChatContext): () => void
   let inFlight = false;
   let queued = false;
   let immediateTimer: ReturnType<typeof setTimeout> | null = null;
+  let scheduleRefresh: RefreshRequester = () => {};
 
   const runRefresh = async (requestedSessionId?: string): Promise<void> => {
     if (disposed || inFlight || !isPeerDeviceModeActive()) {
@@ -207,7 +208,7 @@ export function installPeerSessionRefresh(context: FlowChatContext): () => void
     }
   };
 
-  const scheduleRefresh: RefreshRequester = (sessionId) => {
+  scheduleRefresh = (sessionId) => {
     if (disposed) {
       return;
     }

src/web-ui/src/flow_chat/services/flow-chat-manager/PersistenceModule.test.ts

@@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import type { DialogTurn, FlowTextItem, ModelRound } from '../../types/flow-chat';
 import {
   convertDialogTurnToBackendFormat,
+  debouncedSaveDialogTurn,
   immediateSaveDialogTurn,
   saveDialogTurnToDisk,
 } from './PersistenceModule';
@@ -291,6 +292,26 @@ describe('PersistenceModule', () => {
     expect(saveSessionTurn).toHaveBeenCalledTimes(1);
   });
 
+  it('checkpoints continuous streamed output without waiting for a quiet period', async () => {
+    const turn = createDialogTurn('processing');
+    const context = createContext(turn);
+
+    debouncedSaveDialogTurn(context, SESSION_ID, TURN_ID, 2000);
+    await vi.advanceTimersByTimeAsync(1000);
+    debouncedSaveDialogTurn(context, SESSION_ID, TURN_ID, 2000);
+    await vi.advanceTimersByTimeAsync(999);
+    expect(saveSessionTurn).not.toHaveBeenCalled();
+
+    await vi.advanceTimersByTimeAsync(1);
+    await flushMicrotasks();
+    expect(saveSessionTurn).toHaveBeenCalledTimes(1);
+
+    debouncedSaveDialogTurn(context, SESSION_ID, TURN_ID, 2000);
+    await vi.advanceTimersByTimeAsync(2000);
+    await flushMicrotasks();
+    expect(saveSessionTurn).toHaveBeenCalledTimes(2);
+  });
+
   it('flushes terminal turn saves immediately', async () => {
     const turn = createDialogTurn('completed');
     const context = createContext(turn);