diff --git a/packages/cli/src/commands/session.rs b/packages/cli/src/commands/session.rs index 923319bdb..cd0120410 100644 --- a/packages/cli/src/commands/session.rs +++ b/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, rows_override: Option) -> (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]")); + } +} diff --git a/packages/daemon/src/services/agent_session_service.rs b/packages/daemon/src/services/agent_session_service.rs index 5de11a91e..16fc537a9 100644 --- a/packages/daemon/src/services/agent_session_service.rs +++ b/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, } diff --git a/packages/daemon/tests/agent_session_grpc.rs b/packages/daemon/tests/agent_session_grpc.rs index 668a238f2..2cea4308c 100644 --- a/packages/daemon/tests/agent_session_grpc.rs +++ b/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(()); +} diff --git a/packages/desktop-app/src-tauri/src/commands/agent_session.rs b/packages/desktop-app/src-tauri/src/commands/agent_session.rs index de8b641bd..134441688 100644 --- a/packages/desktop-app/src-tauri/src/commands/agent_session.rs +++ b/packages/desktop-app/src-tauri/src/commands/agent_session.rs @@ -72,6 +72,10 @@ pub struct LaunchSessionResult { pub struct PtyOutputPayload { pub data: Vec, 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!( diff --git a/packages/desktop-app/src/lib/components/agent/pty-output.ts b/packages/desktop-app/src/lib/components/agent/pty-output.ts new file mode 100644 index 000000000..37cc95f60 --- /dev/null +++ b/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`; +} diff --git a/packages/desktop-app/src/lib/components/agent/pty-terminal.svelte b/packages/desktop-app/src/lib/components/agent/pty-terminal.svelte index 3b3b9639a..c80a8cd98 100644 --- a/packages/desktop-app/src/lib/components/agent/pty-terminal.svelte +++ b/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)); } ); diff --git a/packages/desktop-app/src/tests/components/pty-terminal.test.ts b/packages/desktop-app/src/tests/components/pty-terminal.test.ts new file mode 100644 index 000000000..1644453fb --- /dev/null +++ b/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 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]'); + }); +}); diff --git a/packages/proto/proto/agent_session_service.proto b/packages/proto/proto/agent_session_service.proto index 51d0761e8..b528f9878 100644 --- a/packages/proto/proto/agent_session_service.proto +++ b/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; } // ---------------------------------------------------------------------------