fix: agent-terminal-escaping

src-tauri/src/commands/remote_pty_commands.rs

@@ -7,6 +7,7 @@
 use crate::core::feature_preview::PreviewFeature;
 use crate::core::remote_control_plane::SshEndpoint;
 use crate::core::remote_pty::{PtyLaunchSpec, RemotePtyBinding, RemotePtyError, RemotePtyManager};
+use crate::pty::Utf8StreamDecoder;
 use crate::AppState;
 use serde::Serialize;
 use tauri::{Emitter, State};
@@ -96,9 +97,14 @@ pub async fn remote_pty_create(
   let data_event = data_event_name(&session_id);
   let exit_event = exit_event_name(&session_id);
   let app_for_data = app.clone();
+  let app_for_decoder_exit = app.clone();
   let sid_for_data = session_id.clone();
   let app_for_exit = app;
   let sid_for_exit = session_id.clone();
+  let decoder = std::sync::Arc::new(std::sync::Mutex::new(Utf8StreamDecoder::new()));
+  let data_decoder = decoder.clone();
+  let exit_decoder = decoder;
+  let data_event_for_exit = data_event.clone();
 
   manager
     .create(
@@ -111,18 +117,23 @@ pub async fn remote_pty_create(
         // Never log raw terminal data (PRD "never log raw terminal output ...
         // by default"); only the event name/session id are logged, in
         // `remote_pty_create`'s entry line above.
-        if let Err(error) =
-          app_for_data.emit(&data_event, String::from_utf8_lossy(&chunk).into_owned())
-        {
-          log::warn!(
-            "remote pty emit failed: session_id={}, event={}, error={}",
-            sid_for_data,
-            data_event,
-            error
-          );
+        let decoded = data_decoder.lock().unwrap().push(&chunk);
+        if !decoded.is_empty() {
+          if let Err(error) = app_for_data.emit(&data_event, decoded) {
+            log::warn!(
+              "remote pty emit failed: session_id={}, event={}, error={}",
+              sid_for_data,
+              data_event,
+              error
+            );
+          }
         }
       },
       move |exit_status| {
+        let trailing = exit_decoder.lock().unwrap().finish();
+        if !trailing.is_empty() {
+          let _ = app_for_decoder_exit.emit(&data_event_for_exit, trailing);
+        }
         if let Err(error) = app_for_exit.emit(&exit_event, RemotePtyExitPayload { exit_status }) {
           log::warn!(
             "remote pty exit emit failed: session_id={}, event={}, error={}",

src-tauri/src/pty.rs

@@ -14,32 +14,51 @@ use std::thread;
 /// - `new_bytes`: the new bytes read from the PTY
 ///
 /// Returns a valid UTF-8 String, potentially leaving trailing incomplete bytes in `pending`.
-fn process_utf8_chunk(pending: &mut Vec<u8>, new_bytes: &[u8]) -> String {
-  // Combine pending bytes with new bytes
-  let mut combined = std::mem::take(pending);
-  combined.extend_from_slice(new_bytes);
-
-  match std::str::from_utf8(&combined) {
-    Ok(valid_str) => {
-      // All bytes are valid UTF-8
-      valid_str.to_string()
-    }
-    Err(error) => {
-      let valid_up_to = error.valid_up_to();
-
-      // Check if this is an incomplete sequence at the end (not a real error)
-      if error.error_len().is_none() {
-        // Incomplete sequence at end - buffer the trailing bytes
-        let (valid, trailing) = combined.split_at(valid_up_to);
-        *pending = trailing.to_vec();
-
-        // Return the valid portion (should always be valid UTF-8)
-        String::from_utf8(valid.to_vec()).unwrap_or_default()
-      } else {
-        // Invalid UTF-8 mid-stream: use lossy output (rare for a real PTY)
-        String::from_utf8_lossy(&combined).to_string()
+#[derive(Default)]
+pub(crate) struct Utf8StreamDecoder {
+  pending: Vec<u8>,
+}
+
+impl Utf8StreamDecoder {
+  pub(crate) fn new() -> Self {
+    Self::default()
+  }
+
+  pub(crate) fn push(&mut self, new_bytes: &[u8]) -> String {
+    self.pending.extend_from_slice(new_bytes);
+    let mut output = String::new();
+
+    loop {
+      match std::str::from_utf8(&self.pending) {
+        Ok(valid) => {
+          output.push_str(valid);
+          self.pending.clear();
+          break;
+        }
+        Err(error) => {
+          let valid_up_to = error.valid_up_to();
+          output.push_str(std::str::from_utf8(&self.pending[..valid_up_to]).unwrap_or_default());
+          match error.error_len() {
+            None => {
+              self.pending.drain(..valid_up_to);
+              break;
+            }
+            Some(invalid_len) => {
+              output.push('�');
+              self.pending.drain(..valid_up_to + invalid_len);
+            }
+          }
+        }
       }
     }
+
+    output
+  }
+
+  pub(crate) fn finish(&mut self) -> String {
+    let output = String::from_utf8_lossy(&self.pending).into_owned();
+    self.pending.clear();
+    output
   }
 }
 
@@ -409,7 +428,7 @@ impl PtyManager {
     let reader_session_id = session_id.clone();
     thread::spawn(move || {
       let mut buffer = [0u8; 8192];
-      let mut pending_bytes: Vec<u8> = Vec::with_capacity(4);
+      let mut utf8_decoder = Utf8StreamDecoder::new();
       let mut line_buffer = String::new();
       let mut suppressed_tail = String::new();
       let mut non_matching_lines_emitted: usize = 0;
@@ -422,8 +441,8 @@ impl PtyManager {
         match reader.read(&mut buffer) {
           Ok(0) => {
             // EOF: flush any pending bytes
-            if !pending_bytes.is_empty() {
-              let data = String::from_utf8_lossy(&pending_bytes).to_string();
+            {
+              let data = utf8_decoder.finish();
               if !data.is_empty() {
                 line_buffer.push_str(&data);
               }
@@ -435,7 +454,7 @@ impl PtyManager {
             break;
           }
           Ok(n) => {
-            let mut data = process_utf8_chunk(&mut pending_bytes, &buffer[..n]);
+            let mut data = utf8_decoder.push(&buffer[..n]);
             if data.is_empty() {
               continue;
             }
@@ -690,6 +709,25 @@ mod tests {
     }
   }
 
+  #[test]
+  fn utf8_stream_decoder_preserves_characters_split_across_chunks() {
+    let mut decoder = Utf8StreamDecoder::new();
+    let separator = "⎯".as_bytes();
+
+    assert_eq!(decoder.push(&separator[..1]), "");
+    assert_eq!(decoder.push(&separator[1..2]), "");
+    assert_eq!(decoder.push(&separator[2..]), "⎯");
+    assert_eq!(decoder.finish(), "");
+  }
+
+  #[test]
+  fn utf8_stream_decoder_recovers_after_invalid_bytes() {
+    let mut decoder = Utf8StreamDecoder::new();
+
+    assert_eq!(decoder.push(&[b'a', 0xff, b'b']), "a�b");
+    assert_eq!(decoder.push("⎯".as_bytes()), "⎯");
+  }
+
   #[test]
   fn rejects_duplicate_live_session_id() {
     let manager = PtyManager::new();

src/components/terminal/AgentTerminalPanel.tsx

@@ -6,7 +6,10 @@ import React, {
   useState,
 } from "react";
 import { useAgentMessageQueue } from "../../hooks/useAgentMessageQueue";
-import { looksLikeAgentUserQuestion } from "../../lib/agentMessageQueue";
+import {
+  looksLikeAgentUserQuestion,
+  looksLikeShellPrompt,
+} from "../../lib/agentMessageQueue";
 import {
   ptyClose,
   ptyWrite,
@@ -194,10 +197,10 @@ export const AgentTerminalPanel = ({
   };
 
   const handleTerminalIdle = () => {
+    const outputWindow = terminalQuestionWindow(processOutputTailRef.current);
     markIdle({
-      awaitingQuestion: looksLikeAgentUserQuestion(
-        terminalQuestionWindow(processOutputTailRef.current),
-      ),
+      awaitingQuestion: looksLikeAgentUserQuestion(outputWindow),
+      shellPrompt: looksLikeShellPrompt(outputWindow),
     });
     chatRecorderRef.current?.idle();
     onTerminalIdle?.();

src/hooks/useAgentMessageQueue.test.ts

@@ -154,6 +154,21 @@ describe("useAgentMessageQueue", () => {
     expect(write).not.toHaveBeenCalled();
   });
 
+  it("does not send queued text after the agent has returned to a shell", async () => {
+    const { result } = renderHook(() =>
+      useAgentMessageQueue({ ptySessionId: "pty-1", write }),
+    );
+
+    act(() => {
+      result.current.enqueue("text with ' quotes and $(substitutions)");
+      result.current.markIdle({ shellPrompt: true });
+    });
+
+    await waitFor(() => expect(result.current.isBusy).toBe(false));
+    expect(write).not.toHaveBeenCalled();
+    expect(result.current.messages).toHaveLength(1);
+  });
+
   it("sends after a later idle that is not a user question", async () => {
     const { result } = renderHook(() =>
       useAgentMessageQueue({ ptySessionId: "pty-1", write }),

src/hooks/useAgentMessageQueue.ts

@@ -24,7 +24,10 @@ export interface UseAgentMessageQueueResult {
   /** Mark the agent as producing output (busy). */
   markBusy: () => void;
   /** Mark the agent idle and flush the oldest queued message if any. */
-  markIdle: (options?: { awaitingQuestion?: boolean }) => void;
+  markIdle: (options?: {
+    awaitingQuestion?: boolean;
+    shellPrompt?: boolean;
+  }) => void;
   clear: () => void;
 }
 
@@ -130,8 +133,12 @@ export function useAgentMessageQueue({
     isBusyRef.current = true;
   };
 
-  const markIdle = (options?: { awaitingQuestion?: boolean }) => {
-    awaitingQuestionRef.current = options?.awaitingQuestion === true;
+  const markIdle = (options?: {
+    awaitingQuestion?: boolean;
+    shellPrompt?: boolean;
+  }) => {
+    awaitingQuestionRef.current =
+      options?.awaitingQuestion === true || options?.shellPrompt === true;
     setIsBusy(false);
     isBusyRef.current = false;
     void flushOldest();

src/lib/agentMessageQueue.test.ts

@@ -5,6 +5,7 @@ import {
   enqueueAgentMessage,
   formatAgentMessageForPty,
   looksLikeAgentUserQuestion,
+  looksLikeShellPrompt,
   removeAgentMessage,
   updateAgentMessage,
 } from "./agentMessageQueue";
@@ -66,6 +67,22 @@ describe("agentMessageQueue", () => {
   });
 });
 
+describe("looksLikeShellPrompt", () => {
+  it("detects the zsh prompt shown after an agent process exits", () => {
+    expect(looksLikeShellPrompt("projects/treq [main] » ")).toBe(true);
+  });
+
+  it("detects an unterminated zsh parser continuation prompt", () => {
+    expect(looksLikeShellPrompt("subsh quote> ")).toBe(true);
+  });
+
+  it("does not mistake ordinary agent output for a shell prompt", () => {
+    expect(looksLikeShellPrompt("Tests 1 failed | 321 passed (322)\n")).toBe(
+      false,
+    );
+  });
+});
+
 describe("looksLikeAgentUserQuestion", () => {
   it("detects a permission prompt that ends with a question mark", () => {
     const output = [

src/lib/agentMessageQueue.ts

@@ -95,3 +95,23 @@ export function looksLikeAgentUserQuestion(output: string): boolean {
   if (lines.length === 0) return false;
   return lines.slice(-QUESTION_TAIL_LINES).some((line) => line.endsWith("?"));
 }
+
+/** True when the agent process has returned control to an interactive shell. */
+export function looksLikeShellPrompt(output: string): boolean {
+  const visible = stripAnsiEscapes(output).replace(/\r/g, "");
+  const lastLine = visible
+    .split("\n")
+    .map((line) => line.trimEnd())
+    .filter((line) => line.trim().length > 0)
+    .at(-1);
+  if (!lastLine) return false;
+
+  if (
+    /^(?:(?:subsh|cmdsubst|mathsubst) )*(?:dquote|quote|cmdsubst|mathsubst|subsh)>\s*$/.test(
+      lastLine,
+    )
+  ) {
+    return true;
+  }
+  return /(?:^|\s)[$#%❯»]\s*$/.test(lastLine);
+}