Surface PTY StreamOutput broadcast lag to clients as a loss marker (closes #2985)

packages/cli/src/commands/session.rs

@@ -193,6 +193,9 @@ async fn stream_bridge(client: &mut SessionClient, session_id: String) -> Result
     // Drive the output stream to stdout.
     loop {
         match output_stream.message().await {
+            Ok(Some(chunk)) if chunk.dropped_chunks > 0 => {
+                write_stdout(dropped_notice(chunk.dropped_chunks).as_bytes())?;
+            }
             Ok(Some(chunk)) => {
                 write_stdout(&chunk.data)?;
             }
@@ -211,6 +214,19 @@ async fn stream_bridge(client: &mut SessionClient, session_id: String) -> Result
     Ok(())
 }
 
+/// Text written in place of output the daemon dropped because this stream fell
+/// behind a burst. Starts on a fresh line (raw mode needs the explicit `\r`)
+/// so it does not splice into a partially written line. Keep the wording in
+/// step with the desktop terminal's `formatDroppedNotice` (`pty-output.ts`).
+fn dropped_notice(dropped_chunks: u64) -> String {
+    let unit = if dropped_chunks == 1 {
+        "chunk"
+    } else {
+        "chunks"
+    };
+    format!("\r\n\x1b[33m[output truncated: {dropped_chunks} {unit} dropped]\x1b[0m\r\n")
+}
+
 fn detect_terminal_size(cols_override: Option<u32>, rows_override: Option<u32>) -> (u32, u32) {
     let (detected_cols, detected_rows) = terminal::size().unwrap_or((80, 24));
     let cols = cols_override.unwrap_or(detected_cols as u32);
@@ -227,3 +243,17 @@ fn format_unix_time(unix_secs: i64) -> String {
         None => "unknown".to_string(),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::dropped_notice;
+
+    #[test]
+    fn dropped_notice_starts_on_a_fresh_line_and_pluralises() {
+        assert_eq!(
+            dropped_notice(1),
+            "\r\n\x1b[33m[output truncated: 1 chunk dropped]\x1b[0m\r\n"
+        );
+        assert!(dropped_notice(42).contains("[output truncated: 42 chunks dropped]"));
+    }
+}

packages/daemon/src/services/agent_session_service.rs

@@ -264,28 +264,27 @@ impl AgentSessionService for AgentSessionHandler {
                         yield Ok(OutputChunk {
                             data: chunk.data,
                             timestamp_ms,
+                            dropped_chunks: 0,
                         });
                     }
                     Err(RecvError::Lagged(skipped)) => {
                         // Client (or this handler) fell behind the broadcast
-                        // buffer. Continue draining rather than tearing the
+                        // buffer. Keep draining rather than tearing the
                         // stream down — losing a render frame is preferable
-                        // to losing the whole session view.
-                        //
-                        // TODO(review): Lag is invisible to the client
-                        // today (only a server-side tracing::warn). Options
-                        // for surfacing it: a sentinel OutputChunk with an
-                        // in-band "[N bytes dropped]" notice, or a session-
-                        // level metric on ListSessions. Deferred — current
-                        // behavior matches what a real terminal does when
-                        // the kernel buffer overflows (silent drop).
+                        // to losing the whole session view — but tell the
+                        // client about the gap with a loss-marker chunk so it
+                        // can show that output was truncated.
                         debug_assert!(skipped > 0, "RecvError::Lagged with zero skipped chunks");
                         tracing::warn!(
                             session_id = %id,
                             skipped,
                             "StreamOutput subscriber lagged; some chunks dropped"
                         );
-                        continue;
+                        yield Ok(OutputChunk {
+                            data: Vec::new(),
+                            timestamp_ms: Utc::now().timestamp_millis(),
+                            dropped_chunks: skipped,
+                        });
                     }
                     Err(RecvError::Closed) => break,
                 }

packages/daemon/tests/agent_session_grpc.rs

@@ -413,3 +413,61 @@ async fn terminate_unknown_session_reports_was_not_running() {
 
     let _ = shutdown.send(());
 }
+
+#[tokio::test]
+async fn stream_output_reports_dropped_chunks_when_subscriber_lags() {
+    let (mut client, manager, shutdown, _tempdir) = spawn_test_daemon().await;
+
+    // Flood far more output than the broadcast buffer (256 chunks of at most
+    // 4 KiB) plus the HTTP/2 receive window can hold, while the client
+    // deliberately does not read. The daemon's subscriber must then lag, and
+    // the gap has to reach the client as a loss marker rather than vanish.
+    // The leading sleep lets the stream subscribe before the flood starts.
+    let session = PtySession::launch_for_test(
+        "sh",
+        vec!["-c".into(), "sleep 0.2 && yes | head -c 33554432".into()],
+    )
+    .expect("launch flood session");
+    let id = manager.insert(session).await;
+
+    let mut stream = client
+        .stream_output(StreamOutputRequest {
+            session_id: id.to_string(),
+        })
+        .await
+        .expect("stream_output rpc accepted")
+        .into_inner();
+
+    // Not reading for a while is what creates the backpressure: the HTTP/2
+    // window fills, the daemon stops polling its receiver, and the flood
+    // overruns the broadcast buffer. Removing this sleep removes the lag.
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    let marker = timeout(Duration::from_secs(10), async {
+        while let Some(Ok(chunk)) = stream.next().await {
+            if chunk.dropped_chunks > 0 {
+                return Some(chunk);
+            }
+        }
+        None
+    })
+    .await
+    .ok()
+    .flatten()
+    .expect("a lagging subscriber should receive a dropped-chunks marker");
+
+    assert!(
+        marker.data.is_empty(),
+        "loss marker must not carry output bytes"
+    );
+    assert!(marker.timestamp_ms > 0);
+
+    client
+        .terminate_session(TerminateSessionRequest {
+            session_id: id.to_string(),
+        })
+        .await
+        .expect("terminate rpc");
+
+    let _ = shutdown.send(());
+}

packages/desktop-app/src-tauri/src/commands/agent_session.rs

@@ -72,6 +72,10 @@ pub struct LaunchSessionResult {
 pub struct PtyOutputPayload {
     pub data: Vec<u8>,
     pub timestamp_ms: i64,
+    /// Non-zero when the daemon dropped this many chunks because the stream
+    /// fell behind; `data` is then empty and the frontend shows a truncation
+    /// notice in place of the missing output.
+    pub dropped_chunks: u64,
 }
 
 #[derive(Debug, Serialize)]
@@ -219,6 +223,7 @@ pub async fn launch_session(
                             let payload = PtyOutputPayload {
                                 data: chunk.data.to_vec(),
                                 timestamp_ms: chunk.timestamp_ms,
+                                dropped_chunks: chunk.dropped_chunks,
                             };
                             if let Err(e) = app.emit(&event_name, payload) {
                                 tracing::warn!(

packages/desktop-app/src/lib/components/agent/pty-output.ts

@@ -0,0 +1,11 @@
+/**
+ * Terminal text written in place of PTY output the daemon dropped because the
+ * stream fell behind a burst. Starts on a fresh line so it does not splice
+ * into a partially rendered line, and is coloured like the session-closed
+ * notice so it reads as a notice rather than program output. Keep the wording
+ * in step with the CLI's `dropped_notice` (`packages/cli/src/commands/session.rs`).
+ */
+export function formatDroppedNotice(droppedChunks: number): string {
+  const unit = droppedChunks === 1 ? 'chunk' : 'chunks';
+  return `\r\n\x1b[33m[output truncated: ${droppedChunks} ${unit} dropped]\x1b[0m\r\n`;
+}

packages/desktop-app/src/lib/components/agent/pty-terminal.svelte

@@ -5,6 +5,7 @@
   import { listen, type UnlistenFn } from '@tauri-apps/api/event';
   import { ptyWriteInput, ptyResizeTerminal } from '$lib/services/tauri-commands';
   import { createLogger } from '$lib/utils/logger';
+  import { formatDroppedNotice } from './pty-output';
 
   const log = createLogger('PtyTerminal');
 
@@ -45,11 +46,18 @@
     });
 
     // Subscribe to output events from the Tauri backend
-    unlistenOutput = await listen<{ data: number[]; timestampMs: number }>(
+    unlistenOutput = await listen<{ data: number[]; timestampMs: number; droppedChunks: number }>(
       `pty-output-${sessionId}`,
       (event) => {
-        const bytes = new Uint8Array(event.payload.data);
-        terminal.write(bytes);
+        const { data, droppedChunks } = event.payload;
+        if (droppedChunks > 0) {
+          // The daemon discarded output because this stream fell behind a
+          // burst; mark the gap so the user knows the rendering is incomplete.
+          log.warn('PTY output dropped', { sessionId, droppedChunks });
+          terminal.write(formatDroppedNotice(droppedChunks));
+          return;
+        }
+        terminal.write(new Uint8Array(data));
       }
     );
 

packages/desktop-app/src/tests/components/pty-terminal.test.ts

@@ -0,0 +1,95 @@
+/**
+ * pty-terminal — rendering of daemon loss markers.
+ *
+ * When a PTY output stream falls behind a burst, the daemon drops chunks and
+ * sends a marker (`droppedChunks > 0`, empty `data`) in their place. The
+ * terminal must show a visible truncation notice for it instead of silently
+ * rendering a desynced stream.
+ */
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
+import { render, cleanup } from '@testing-library/svelte';
+
+vi.mock('$lib/utils/logger', () => ({
+  createLogger: () => ({ debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() })
+}));
+
+vi.mock('$lib/services/tauri-commands', () => ({
+  ptyWriteInput: vi.fn().mockResolvedValue(undefined),
+  ptyResizeTerminal: vi.fn().mockResolvedValue(undefined)
+}));
+
+const terminalWrite = vi.fn();
+vi.mock('@xterm/xterm', () => ({
+  Terminal: class {
+    cols = 80;
+    rows = 24;
+    write = terminalWrite;
+    writeln = vi.fn();
+    loadAddon = vi.fn();
+    open = vi.fn();
+    onData = vi.fn();
+    dispose = vi.fn();
+  }
+}));
+
+vi.mock('@xterm/addon-fit', () => ({
+  FitAddon: class {
+    fit = vi.fn();
+  }
+}));
+
+type OutputPayload = { data: number[]; timestampMs: number; droppedChunks: number };
+const listeners = new Map<string, (event: { payload: OutputPayload }) => void>();
+vi.mock('@tauri-apps/api/event', () => ({
+  listen: vi.fn((name: string, handler: (event: { payload: OutputPayload }) => void) => {
+    listeners.set(name, handler);
+    return Promise.resolve(() => {});
+  })
+}));
+
+import PtyTerminal from '$lib/components/agent/pty-terminal.svelte';
+
+async function mountAndGetOutputHandler(sessionId: string) {
+  render(PtyTerminal, { props: { sessionId } });
+  await vi.waitFor(() => expect(listeners.has(`pty-output-${sessionId}`)).toBe(true));
+  return listeners.get(`pty-output-${sessionId}`)!;
+}
+
+describe('PtyTerminal', () => {
+  beforeEach(() => {
+    terminalWrite.mockReset();
+    listeners.clear();
+    vi.stubGlobal(
+      'ResizeObserver',
+      class {
+        observe = vi.fn();
+        disconnect = vi.fn();
+      }
+    );
+  });
+
+  afterEach(() => {
+    cleanup();
+    vi.unstubAllGlobals();
+  });
+
+  it('writes ordinary output bytes straight to the terminal', async () => {
+    const onOutput = await mountAndGetOutputHandler('s1');
+
+    onOutput({ payload: { data: [104, 105], timestampMs: 1, droppedChunks: 0 } });
+
+    expect(terminalWrite).toHaveBeenCalledTimes(1);
+    expect(terminalWrite.mock.calls[0][0]).toEqual(new Uint8Array([104, 105]));
+  });
+
+  it('shows a truncation notice for a loss marker', async () => {
+    const onOutput = await mountAndGetOutputHandler('s2');
+
+    onOutput({ payload: { data: [], timestampMs: 1, droppedChunks: 42 } });
+
+    expect(terminalWrite).toHaveBeenCalledTimes(1);
+    const written = terminalWrite.mock.calls[0][0];
+    expect(typeof written).toBe('string');
+    expect(written).toContain('[output truncated: 42 chunks dropped]');
+  });
+});

packages/proto/proto/agent_session_service.proto

@@ -67,11 +67,20 @@ message StreamOutputRequest {
 // distinguish them at this layer. The bytes may contain partial UTF-8 sequences
 // and terminal escape codes; consumers are expected to render them the way a
 // terminal would.
+//
+// A chunk with a non-zero `dropped_chunks` is a loss marker rather than
+// output: this subscriber fell behind the daemon's output buffer and that many
+// chunks were discarded before it could read them. Its `data` is empty.
+// Consumers should surface the gap (e.g. an "output truncated" notice) instead
+// of rendering a silently desynced stream.
 message OutputChunk {
   // Raw bytes read from the PTY (may include ANSI escape sequences).
   bytes data = 1;
   // Unix timestamp in milliseconds when this chunk was produced.
   int64 timestamp_ms = 2;
+  // Number of chunks dropped immediately before this point in the stream.
+  // Zero for ordinary output chunks.
+  uint64 dropped_chunks = 3;
 }
 
 // ---------------------------------------------------------------------------