fix: add request timeouts, bounded queues, and tool output truncation

Author: Dimillian

src-tauri/src/backend/app_server.rs

@@ -52,6 +52,8 @@ fn build_initialize_params(client_version: &str) -> Value {
     })
 }
 
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(300);
+
 pub(crate) struct WorkspaceSession {
     pub(crate) entry: WorkspaceEntry,
     pub(crate) child: Mutex<Child>,
@@ -77,9 +79,24 @@ impl WorkspaceSession {
         let id = self.next_id.fetch_add(1, Ordering::SeqCst);
         let (tx, rx) = oneshot::channel();
         self.pending.lock().await.insert(id, tx);
-        self.write_message(json!({ "id": id, "method": method, "params": params }))
-            .await?;
-        rx.await.map_err(|_| "request canceled".to_string())
+        if let Err(error) = self
+            .write_message(json!({ "id": id, "method": method, "params": params }))
+            .await
+        {
+            self.pending.lock().await.remove(&id);
+            return Err(error);
+        }
+        match timeout(REQUEST_TIMEOUT, rx).await {
+            Ok(Ok(value)) => Ok(value),
+            Ok(Err(_)) => Err("request canceled".to_string()),
+            Err(_) => {
+                self.pending.lock().await.remove(&id);
+                Err(format!(
+                    "request timed out after {} seconds",
+                    REQUEST_TIMEOUT.as_secs()
+                ))
+            }
+        }
     }
 
     pub(crate) async fn send_notification(
@@ -406,6 +423,9 @@ pub(crate) async fn spawn_workspace_session<E: EventSink>(
                 }
             }
         }
+
+        // Ensure pending foreground requests cannot accumulate after process output ends.
+        session_clone.pending.lock().await.clear();
     });
 
     let workspace_id = entry.id.clone();

src-tauri/src/remote_backend/mod.rs

@@ -6,9 +6,11 @@ mod transport;
 use serde_json::{json, Value};
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 use tauri::AppHandle;
 use tokio::sync::Mutex;
+use tokio::time::timeout;
 
 use crate::state::AppState;
 use crate::types::{BackendMode, RemoteBackendProvider};
@@ -18,6 +20,9 @@ use self::protocol::{build_request_line, DEFAULT_REMOTE_HOST, DISCONNECTED_MESSA
 use self::tcp_transport::TcpTransport;
 use self::transport::{PendingMap, RemoteTransport, RemoteTransportConfig, RemoteTransportKind};
 
+const REMOTE_REQUEST_TIMEOUT: Duration = Duration::from_secs(300);
+const REMOTE_SEND_TIMEOUT: Duration = Duration::from_secs(15);
+
 pub(crate) fn normalize_path_for_remote(path: String) -> String {
     let trimmed = path.trim();
     if trimmed.is_empty() {
@@ -58,7 +63,7 @@ pub(crate) struct RemoteBackend {
 }
 
 struct RemoteBackendInner {
-    out_tx: tokio::sync::mpsc::UnboundedSender<String>,
+    out_tx: tokio::sync::mpsc::Sender<String>,
     pending: Arc<Mutex<PendingMap>>,
     next_id: AtomicU64,
     connected: Arc<std::sync::atomic::AtomicBool>,
@@ -75,12 +80,32 @@ impl RemoteBackend {
         self.inner.pending.lock().await.insert(id, tx);
 
         let message = build_request_line(id, method, params)?;
-        if self.inner.out_tx.send(message).is_err() {
-            self.inner.pending.lock().await.remove(&id);
-            return Err(DISCONNECTED_MESSAGE.to_string());
+        match timeout(REMOTE_SEND_TIMEOUT, self.inner.out_tx.send(message)).await {
+            Ok(Ok(())) => {}
+            Ok(Err(_)) => {
+                self.inner.pending.lock().await.remove(&id);
+                return Err(DISCONNECTED_MESSAGE.to_string());
+            }
+            Err(_) => {
+                self.inner.pending.lock().await.remove(&id);
+                return Err(format!(
+                    "remote backend request dispatch timed out after {} seconds",
+                    REMOTE_SEND_TIMEOUT.as_secs()
+                ));
+            }
         }
 
-        rx.await.map_err(|_| DISCONNECTED_MESSAGE.to_string())?
+        match timeout(REMOTE_REQUEST_TIMEOUT, rx).await {
+            Ok(Ok(result)) => result,
+            Ok(Err(_)) => Err(DISCONNECTED_MESSAGE.to_string()),
+            Err(_) => {
+                self.inner.pending.lock().await.remove(&id);
+                Err(format!(
+                    "remote backend request timed out after {} seconds",
+                    REMOTE_REQUEST_TIMEOUT.as_secs()
+                ))
+            }
+        }
     }
 }
 

src-tauri/src/remote_backend/orbit_ws_transport.rs

@@ -15,6 +15,7 @@ use super::transport::{
 };
 
 pub(crate) struct OrbitWsTransport;
+const OUTBOUND_QUEUE_CAPACITY: usize = 512;
 
 impl RemoteTransport for OrbitWsTransport {
     fn connect(&self, app: AppHandle, config: RemoteTransportConfig) -> TransportFuture {
@@ -29,7 +30,7 @@ impl RemoteTransport for OrbitWsTransport {
                 .map_err(|err| format!("Failed to connect to Orbit relay at {ws_url}: {err}"))?;
             let (mut writer, mut reader) = stream.split();
 
-            let (out_tx, mut out_rx) = mpsc::unbounded_channel::<String>();
+            let (out_tx, mut out_rx) = mpsc::channel::<String>(OUTBOUND_QUEUE_CAPACITY);
             let pending = Arc::new(Mutex::new(PendingMap::new()));
             let pending_for_writer = Arc::clone(&pending);
             let pending_for_reader = Arc::clone(&pending);

src-tauri/src/remote_backend/transport.rs

@@ -12,6 +12,7 @@ use tokio::sync::{mpsc, oneshot, Mutex};
 use super::protocol::{parse_incoming_line, IncomingMessage, DISCONNECTED_MESSAGE};
 
 pub(crate) type PendingMap = HashMap<u64, oneshot::Sender<Result<Value, String>>>;
+const OUTBOUND_QUEUE_CAPACITY: usize = 512;
 
 #[derive(Clone, Debug)]
 pub(crate) enum RemoteTransportConfig {
@@ -48,7 +49,7 @@ impl RemoteTransportConfig {
 }
 
 pub(crate) struct TransportConnection {
-    pub(crate) out_tx: mpsc::UnboundedSender<String>,
+    pub(crate) out_tx: mpsc::Sender<String>,
     pub(crate) pending: Arc<Mutex<PendingMap>>,
     pub(crate) connected: Arc<AtomicBool>,
 }
@@ -69,7 +70,7 @@ where
     R: AsyncRead + Unpin + Send + 'static,
     W: AsyncWrite + Unpin + Send + 'static,
 {
-    let (out_tx, mut out_rx) = mpsc::unbounded_channel::<String>();
+    let (out_tx, mut out_rx) = mpsc::channel::<String>(OUTBOUND_QUEUE_CAPACITY);
     let pending = Arc::new(Mutex::new(PendingMap::new()));
     let pending_for_writer = Arc::clone(&pending);
     let pending_for_reader = Arc::clone(&pending);

src-tauri/src/shared/codex_aux_core.rs

@@ -352,7 +352,10 @@ where
 
     let mut response_text = String::new();
     let collect_result = timeout(Duration::from_secs(60), async {
-        while let Some(event) = rx.recv().await {
+        loop {
+            let Some(event) = rx.recv().await else {
+                return Err("Background response stream closed before completion".to_string());
+            };
             let method = event.get("method").and_then(|m| m.as_str()).unwrap_or("");
             match method {
                 "item/agentMessage/delta" => {

src-tauri/src/terminal.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 
 use portable_pty::{native_pty_system, CommandBuilder, PtySize};
 use serde::Serialize;
-use tauri::{AppHandle, State};
+use tauri::{AppHandle, Manager, State};
 use tokio::sync::Mutex;
 
 use crate::backend::events::{EventSink, TerminalExit, TerminalOutput};
@@ -66,6 +66,8 @@ fn resolve_locale() -> String {
 
 fn spawn_terminal_reader(
     event_sink: impl EventSink,
+    app: AppHandle,
+    session: Arc<TerminalSession>,
     workspace_id: String,
     terminal_id: String,
     mut reader: Box<dyn Read + Send>,
@@ -125,10 +127,24 @@ fn spawn_terminal_reader(
                 Err(_) => break,
             }
         }
+        let cleanup_workspace_id = workspace_id.clone();
+        let cleanup_terminal_id = terminal_id.clone();
+        let cleanup_session = Arc::clone(&session);
         event_sink.emit_terminal_exit(TerminalExit {
             workspace_id,
             terminal_id,
         });
+        tauri::async_runtime::spawn(async move {
+            let state = app.state::<AppState>();
+            let mut sessions = state.terminal_sessions.lock().await;
+            let key = terminal_key(&cleanup_workspace_id, &cleanup_terminal_id);
+            let should_remove = sessions
+                .get(&key)
+                .is_some_and(|current| Arc::ptr_eq(current, &cleanup_session));
+            if should_remove {
+                sessions.remove(&key);
+            }
+        });
     });
 }
 
@@ -219,10 +235,17 @@ pub(crate) async fn terminal_open(
             .await;
             return Ok(TerminalSessionInfo { id });
         }
-        sessions.insert(key, session);
+        sessions.insert(key, Arc::clone(&session));
     }
-    let event_sink = TauriEventSink::new(app);
-    spawn_terminal_reader(event_sink, workspace_id, terminal_id, reader);
+    let event_sink = TauriEventSink::new(app.clone());
+    spawn_terminal_reader(
+        event_sink,
+        app,
+        Arc::clone(&session),
+        workspace_id,
+        terminal_id,
+        reader,
+    );
 
     Ok(TerminalSessionInfo { id: session_id })
 }

src/utils/threadItems.test.ts

@@ -29,8 +29,8 @@ describe("threadItems", () => {
     }
   });
 
-  it("preserves tool output for fileChange and commandExecution", () => {
-    const output = "x".repeat(21000);
+  it("truncates extremely large tool output for fileChange and commandExecution", () => {
+    const output = "x".repeat(250000);
     const item: ConversationItem = {
       id: "tool-1",
       kind: "tool",
@@ -42,7 +42,9 @@ describe("threadItems", () => {
     const normalized = normalizeItem(item);
     expect(normalized.kind).toBe("tool");
     if (normalized.kind === "tool") {
-      expect(normalized.output).toBe(output);
+      expect(normalized.output).not.toBe(output);
+      expect(normalized.output?.endsWith("...")).toBe(true);
+      expect((normalized.output ?? "").length).toBeLessThan(output.length);
     }
   });
 

src/utils/threadItems.ts

@@ -2,8 +2,9 @@ import type { ConversationItem } from "../types";
 
 const MAX_ITEMS_PER_THREAD = 200;
 const MAX_ITEM_TEXT = 20000;
+const MAX_LARGE_TOOL_TEXT = 200000;
 const TOOL_OUTPUT_RECENT_ITEMS = 40;
-const NO_TRUNCATE_TOOL_TYPES = new Set(["fileChange", "commandExecution"]);
+const LARGE_TOOL_TYPES = new Set(["fileChange", "commandExecution"]);
 const READ_COMMANDS = new Set(["cat", "sed", "head", "tail", "less", "more", "nl"]);
 const LIST_COMMANDS = new Set(["ls", "tree", "find", "fd"]);
 const SEARCH_COMMANDS = new Set(["rg", "grep", "ripgrep", "findstr"]);
@@ -50,6 +51,13 @@ function truncateText(text: string, maxLength = MAX_ITEM_TEXT) {
   return `${text.slice(0, sliceLength)}...`;
 }
 
+function truncateToolText(toolType: string, text: string) {
+  const maxLength = LARGE_TOOL_TYPES.has(toolType)
+    ? MAX_LARGE_TOOL_TEXT
+    : MAX_ITEM_TEXT;
+  return truncateText(text, maxLength);
+}
+
 function normalizeStringList(value: unknown) {
   if (Array.isArray(value)) {
     return value.map((entry) => asString(entry)).filter(Boolean);
@@ -94,23 +102,19 @@ export function normalizeItem(item: ConversationItem): ConversationItem {
     return { ...item, diff: truncateText(item.diff) };
   }
   if (item.kind === "tool") {
-    const isNoTruncateTool = NO_TRUNCATE_TOOL_TYPES.has(item.toolType);
     return {
       ...item,
       title: truncateText(item.title, 200),
       detail: truncateText(item.detail, 2000),
-      output: isNoTruncateTool
-        ? item.output
-        : item.output
-          ? truncateText(item.output)
-          : item.output,
+      output: item.output
+        ? truncateToolText(item.toolType, item.output)
+        : item.output,
       changes: item.changes
         ? item.changes.map((change) => ({
             ...change,
-            diff:
-              isNoTruncateTool || !change.diff
-                ? change.diff
-                : truncateText(change.diff),
+            diff: change.diff
+              ? truncateToolText(item.toolType, change.diff)
+              : change.diff,
           }))
         : item.changes,
     };