fix(exec): cap buffered output for exec and assistant streams

code-rs/core/src/codex.rs

@@ -553,6 +553,7 @@ use crate::exec::ExecToolCallOutput;
 use crate::exec::SandboxType;
 use crate::exec::StdoutStream;
 use crate::exec::StreamOutput;
+use crate::exec::EXEC_CAPTURE_MAX_BYTES;
 use crate::exec::process_exec_tool_call;
 use crate::review_format::format_review_findings_block;
 use crate::exec_env::create_env;
@@ -9626,6 +9627,14 @@ fn format_exec_output_str(exec_output: &ExecToolCallOutput) -> String {
     // Always use the aggregated (stdout + stderr interleaved) stream so the
     // model sees the full build log regardless of which stream a tool used.
     let mut formatted_output = aggregated_output.text.clone();
+    if let Some(truncated_before_bytes) = aggregated_output.truncated_before_bytes {
+        let note = format!(
+            "… clipped {} from the start of command output (showing last {}).\n\n",
+            format_bytes(truncated_before_bytes),
+            format_bytes(EXEC_CAPTURE_MAX_BYTES),
+        );
+        formatted_output = format!("{note}{formatted_output}");
+    }
     if let Some(truncated_after_lines) = aggregated_output.truncated_after_lines {
         formatted_output.push_str(&format!(
             "\n\n[Output truncated after {truncated_after_lines} lines: too many lines or bytes.]",
@@ -9699,6 +9708,22 @@ fn format_exec_output_with_limit(
     serde_json::to_string(&payload).expect("serialize ExecOutput")
 }
 
+fn format_bytes(bytes: usize) -> String {
+    const KIB: f64 = 1024.0;
+    const MIB: f64 = KIB * 1024.0;
+    const GIB: f64 = MIB * 1024.0;
+    let bytes_f = bytes as f64;
+    if bytes >= GIB as usize {
+        format!("{:.1} GiB", bytes_f / GIB)
+    } else if bytes >= MIB as usize {
+        format!("{:.1} MiB", bytes_f / MIB)
+    } else if bytes >= KIB as usize {
+        format!("{:.1} KiB", bytes_f / KIB)
+    } else {
+        format!("{bytes} B")
+    }
+}
+
 fn get_last_assistant_message_from_turn(responses: &[ResponseItem]) -> Option<String> {
     responses.iter().rev().find_map(|item| {
         if let ResponseItem::Message { role, content, .. } = item {

code-rs/core/src/exec.rs

@@ -48,6 +48,7 @@ const EXEC_TIMEOUT_EXIT_CODE: i32 = 124; // conventional timeout exit code
 // I/O buffer sizing
 const READ_CHUNK_SIZE: usize = 8192; // bytes per read
 const AGGREGATE_BUFFER_INITIAL_CAPACITY: usize = 8 * 1024; // 8 KiB
+pub(crate) const EXEC_CAPTURE_MAX_BYTES: usize = 32 * 1024 * 1024; // 32 MiB tail per stream
 
 /// Limit the number of ExecCommandOutputDelta events emitted per exec call.
 /// Aggregation still collects full output; only the live event stream is capped.
@@ -230,6 +231,7 @@ fn is_likely_sandbox_denied(sandbox_type: SandboxType, exit_code: i32) -> bool {
 pub struct StreamOutput<T> {
     pub text: T,
     pub truncated_after_lines: Option<u32>,
+    pub truncated_before_bytes: Option<usize>,
 }
 #[derive(Debug)]
 struct RawExecToolCallOutput {
@@ -245,6 +247,7 @@ impl StreamOutput<String> {
         Self {
             text,
             truncated_after_lines: None,
+            truncated_before_bytes: None,
         }
     }
 }
@@ -254,13 +257,74 @@ impl StreamOutput<Vec<u8>> {
         StreamOutput {
             text: String::from_utf8_lossy(&self.text).to_string(),
             truncated_after_lines: self.truncated_after_lines,
+            truncated_before_bytes: self.truncated_before_bytes,
         }
     }
 }
 
-#[inline]
-fn append_all(dst: &mut Vec<u8>, src: &[u8]) {
-    dst.extend_from_slice(src);
+#[derive(Debug)]
+struct TruncatingBuffer {
+    buf: Vec<u8>,
+    max_bytes: usize,
+    truncated_before_bytes: usize,
+}
+
+impl TruncatingBuffer {
+    fn with_capacity(max_bytes: usize, capacity_hint: usize) -> Self {
+        let capped_hint = max_bytes.min(capacity_hint);
+        Self {
+            buf: Vec::with_capacity(capped_hint),
+            max_bytes,
+            truncated_before_bytes: 0,
+        }
+    }
+
+    fn push(&mut self, chunk: &[u8]) {
+        if chunk.is_empty() {
+            return;
+        }
+
+        if self.max_bytes == 0 {
+            self.truncated_before_bytes = self
+                .truncated_before_bytes
+                .saturating_add(chunk.len());
+            return;
+        }
+
+        if chunk.len() >= self.max_bytes {
+            let drop_existing = self.buf.len();
+            let keep_start = chunk.len().saturating_sub(self.max_bytes);
+            self.truncated_before_bytes = self
+                .truncated_before_bytes
+                .saturating_add(drop_existing + keep_start);
+            self.buf.clear();
+            self.buf
+                .extend_from_slice(&chunk[chunk.len() - self.max_bytes..]);
+            return;
+        }
+
+        let total = self.buf.len().saturating_add(chunk.len());
+        if total <= self.max_bytes {
+            self.buf.extend_from_slice(chunk);
+            return;
+        }
+
+        let overflow = total - self.max_bytes;
+        self.buf.drain(..overflow);
+        self.truncated_before_bytes = self
+            .truncated_before_bytes
+            .saturating_add(overflow);
+        self.buf.extend_from_slice(chunk);
+    }
+
+    fn into_parts(self) -> (Vec<u8>, Option<usize>) {
+        let truncated = if self.truncated_before_bytes > 0 {
+            Some(self.truncated_before_bytes)
+        } else {
+            None
+        };
+        (self.buf, truncated)
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -398,8 +462,16 @@ async fn consume_truncated_output(
         stdout_handle.abort();
         stderr_handle.abort();
         (
-            StreamOutput { text: Vec::new(), truncated_after_lines: None },
-            StreamOutput { text: Vec::new(), truncated_after_lines: None },
+            StreamOutput {
+                text: Vec::new(),
+                truncated_after_lines: None,
+                truncated_before_bytes: None,
+            },
+            StreamOutput {
+                text: Vec::new(),
+                truncated_after_lines: None,
+                truncated_before_bytes: None,
+            },
         )
     } else {
         (stdout_handle.await??, stderr_handle.await??)
@@ -407,13 +479,18 @@ async fn consume_truncated_output(
 
     drop(agg_tx);
 
-    let mut combined_buf = Vec::with_capacity(AGGREGATE_BUFFER_INITIAL_CAPACITY);
+    let mut combined = TruncatingBuffer::with_capacity(
+        EXEC_CAPTURE_MAX_BYTES,
+        AGGREGATE_BUFFER_INITIAL_CAPACITY,
+    );
     while let Ok(chunk) = agg_rx.recv().await {
-        append_all(&mut combined_buf, &chunk);
+        combined.push(&chunk);
     }
+    let (combined_buf, truncated_before_bytes) = combined.into_parts();
     let aggregated_output = StreamOutput {
         text: combined_buf,
         truncated_after_lines: None,
+        truncated_before_bytes,
     };
 
     Ok(RawExecToolCallOutput {
@@ -431,12 +508,13 @@ async fn read_capped<R: AsyncRead + Unpin + Send + 'static>(
     is_stderr: bool,
     aggregate_tx: Option<Sender<Vec<u8>>>,
 ) -> io::Result<StreamOutput<Vec<u8>>> {
-    let mut buf = Vec::with_capacity(AGGREGATE_BUFFER_INITIAL_CAPACITY);
+    let mut buf = TruncatingBuffer::with_capacity(
+        EXEC_CAPTURE_MAX_BYTES,
+        AGGREGATE_BUFFER_INITIAL_CAPACITY,
+    );
     let mut tmp = [0u8; READ_CHUNK_SIZE];
     let mut emitted_deltas: usize = 0;
 
-    // No caps: append all bytes
-
     loop {
         let n = reader.read(&mut tmp).await?;
         if n == 0 {
@@ -481,13 +559,15 @@ async fn read_capped<R: AsyncRead + Unpin + Send + 'static>(
             let _ = tx.send(tmp[..n].to_vec()).await;
         }
 
-        append_all(&mut buf, &tmp[..n]);
+        buf.push(&tmp[..n]);
         // Continue reading to EOF to avoid back-pressure
     }
 
+    let (buf, truncated_before_bytes) = buf.into_parts();
     Ok(StreamOutput {
         text: buf,
         truncated_after_lines: None,
+        truncated_before_bytes,
     })
 }
 

code-rs/core/src/history/state.rs

@@ -748,18 +748,63 @@ fn prune_exec_stream(chunks: &mut Vec<ExecStreamChunk>, max_bytes: usize) {
     }
 }
 
-fn append_assistant_delta(deltas: &mut Vec<AssistantStreamDelta>, delta: AssistantStreamDelta) {
+fn append_assistant_delta(deltas: &mut Vec<AssistantStreamDelta>, delta: AssistantStreamDelta) -> usize {
     if let Some(last) = deltas.last_mut() {
         if delta.sequence.is_some() && delta.sequence == last.sequence {
             last.delta.push_str(&delta.delta);
-            return;
+            return prune_assistant_stream(deltas, MAX_ASSISTANT_STREAM_RETAINED_BYTES);
         }
         if delta.sequence.is_none() && last.sequence.is_none() {
             last.delta.push_str(&delta.delta);
-            return;
+            return prune_assistant_stream(deltas, MAX_ASSISTANT_STREAM_RETAINED_BYTES);
         }
     }
     deltas.push(delta);
+    prune_assistant_stream(deltas, MAX_ASSISTANT_STREAM_RETAINED_BYTES)
+}
+
+fn prune_assistant_stream(deltas: &mut Vec<AssistantStreamDelta>, max_bytes: usize) -> usize {
+    if max_bytes == 0 {
+        let dropped = deltas.iter().map(|d| d.delta.len()).sum();
+        deltas.clear();
+        return dropped;
+    }
+
+    let mut total: usize = deltas.iter().map(|d| d.delta.len()).sum();
+    if total <= max_bytes {
+        return 0;
+    }
+
+    let mut truncated = 0usize;
+    while total > max_bytes && !deltas.is_empty() {
+        if let Some(first) = deltas.first_mut() {
+            let first_len = first.delta.len();
+            if total.saturating_sub(first_len) >= max_bytes {
+                let removed = deltas.remove(0);
+                truncated = truncated.saturating_add(removed.delta.len());
+                total = total.saturating_sub(first_len);
+                continue;
+            }
+
+            let mut drain = total - max_bytes;
+            drain = drain.min(first_len);
+            while drain < first_len && !first.delta.is_char_boundary(drain) {
+                drain += 1;
+            }
+
+            if drain >= first_len {
+                let removed = deltas.remove(0);
+                truncated = truncated.saturating_add(removed.delta.len());
+                total = total.saturating_sub(first_len);
+                continue;
+            }
+
+            first.delta.drain(..drain);
+            truncated = truncated.saturating_add(drain);
+            total = total.saturating_sub(drain);
+        }
+    }
+    truncated
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -779,6 +824,8 @@ pub struct AssistantStreamState {
     pub metadata: Option<MessageMetadata>,
     pub in_progress: bool,
     pub last_updated_at: SystemTime,
+    #[serde(default)]
+    pub truncated_prefix_bytes: usize,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -947,6 +994,9 @@ const EXEC_STREAM_BYTE_STEP: usize = 2 * 1024 * 1024;
 /// Older bytes are truncated from the front once this threshold is exceeded.
 pub const MAX_EXEC_STREAM_RETAINED_BYTES: usize = 32 * 1024 * 1024; // 32 MiB
 
+/// Maximum bytes retained in memory for assistant streaming output.
+pub const MAX_ASSISTANT_STREAM_RETAINED_BYTES: usize = 6 * 1024 * 1024; // 6 MiB
+
 const ASSISTANT_STREAM_CHUNK_THRESHOLD: usize = 2048;
 const ASSISTANT_STREAM_CHUNK_STEP: usize = 256;
 const ASSISTANT_STREAM_BYTE_THRESHOLD: usize = 6 * 1024 * 1024;
@@ -1746,7 +1796,12 @@ impl HistoryState {
                         if let Some(delta_clone) = delta.clone() {
                             self.usage_tracker
                                 .add_assistant_delta(updated.id, delta_clone.delta.len());
-                            append_assistant_delta(&mut updated.deltas, delta_clone);
+                            let truncated = append_assistant_delta(&mut updated.deltas, delta_clone);
+                            if truncated > 0 {
+                                updated.truncated_prefix_bytes = updated
+                                    .truncated_prefix_bytes
+                                    .saturating_add(truncated);
+                            }
                         }
                         updated.preview_markdown = preview_markdown.clone();
                         if let Some(meta) = metadata.clone() {
@@ -1768,6 +1823,7 @@ impl HistoryState {
                 }
 
                 let mut deltas = Vec::new();
+                let mut truncated_prefix_bytes = 0usize;
                 if let Some(delta_value) = delta {
                     if let Some(existing_id) = self.records
                         .iter()
@@ -1781,7 +1837,7 @@ impl HistoryState {
                         self.usage_tracker
                             .add_assistant_delta(existing_id, delta_value.delta.len());
                     }
-                    deltas.push(delta_value);
+                    truncated_prefix_bytes = append_assistant_delta(&mut deltas, delta_value);
                 }
                 let citations = metadata
                     .as_ref()
@@ -1796,6 +1852,7 @@ impl HistoryState {
                     metadata,
                     in_progress: true,
                     last_updated_at: now,
+                    truncated_prefix_bytes,
                 };
                 let record = HistoryRecord::AssistantStream(assistant_state);
                 self.apply_event(HistoryEvent::Insert {
@@ -2593,6 +2650,7 @@ mod tests {
             metadata: None,
             in_progress: true,
             last_updated_at: SystemTime::UNIX_EPOCH,
+            truncated_prefix_bytes: 0,
         }));
 
         let snapshot = state.snapshot();
@@ -2762,6 +2820,7 @@ mod tests {
             metadata: Some(metadata.clone()),
             in_progress: true,
             last_updated_at: now,
+            truncated_prefix_bytes: 0,
         }));
 
         records.push(HistoryRecord::AssistantMessage(AssistantMessageState {

code-rs/tui/src/chatwidget.rs

@@ -18268,6 +18268,7 @@ use crate::chatwidget::message::UserMessage;
             metadata: None,
             in_progress,
             last_updated_at: SystemTime::now(),
+            truncated_prefix_bytes: 0,
         }
     }
 
@@ -23303,6 +23304,7 @@ mod tests {
             metadata: None,
             in_progress: true,
             last_updated_at: SystemTime::now(),
+            truncated_prefix_bytes: 0,
         };
         let stream_cell = history_cell::new_streaming_content(stream_state, &chat.config);
 

code-rs/tui/src/history_cell/stream.rs

@@ -1,5 +1,6 @@
 use super::*;
 use crate::history::{AssistantMessageState, AssistantStreamState};
+use code_core::history::state::MAX_ASSISTANT_STREAM_RETAINED_BYTES;
 use code_core::config_types::UriBasedFileOpener;
 use ratatui::style::Style;
 use ratatui::text::Line;
@@ -95,10 +96,20 @@ pub(crate) fn stream_lines_from_state_with_context(
     cwd: &Path,
     show_ellipsis: bool,
 ) -> Vec<Line<'static>> {
+    let mut markdown = state.preview_markdown.clone();
+    if state.truncated_prefix_bytes > 0 {
+        let note = format!(
+            "… clipped {} from the start of assistant response (showing last {}).\n\n",
+            format_bytes(state.truncated_prefix_bytes),
+            format_bytes(MAX_ASSISTANT_STREAM_RETAINED_BYTES),
+        );
+        markdown = format!("{note}{markdown}");
+    }
+
     let message_state = AssistantMessageState {
         id: state.id,
         stream_id: Some(state.stream_id.clone()),
-        markdown: state.preview_markdown.clone(),
+        markdown,
         citations: state.citations.clone(),
         metadata: state.metadata.clone(),
         token_usage: state
@@ -144,3 +155,19 @@ fn ellipsis_line() -> Line<'static> {
         Style::default().fg(crate::colors::text_dim()),
     )
 }
+
+fn format_bytes(bytes: usize) -> String {
+    const KIB: f64 = 1024.0;
+    const MIB: f64 = KIB * 1024.0;
+    const GIB: f64 = MIB * 1024.0;
+    let bytes_f = bytes as f64;
+    if bytes >= GIB as usize {
+        format!("{:.1} GiB", bytes_f / GIB)
+    } else if bytes >= MIB as usize {
+        format!("{:.1} MiB", bytes_f / MIB)
+    } else if bytes >= KIB as usize {
+        format!("{:.1} KiB", bytes_f / KIB)
+    } else {
+        format!("{bytes} B")
+    }
+}