fix(debugger): bound the Debug console by real lines and coalesce output events (#429)

Follow-up to #385. A single output event can carry thousands of lines, which became one console entry and escaped the line cap; output is now split at real line breaks (partial lines still merge across events) so the 50,000-line bound applies to physical lines. The host also coalesces debug output per session for up to 50 ms / 64 KB before it reaches desktop or remote IPC, flushing before any other session event so stops and exits never overtake earlier output.

Verified on macOS with a Java program printing ~480 lines/s for over a minute: the console stays responsive, follows the tail, and pause/step work.

Fixes #377

crates/ide-host/src/ops/debug.rs

@@ -1,3 +1,5 @@
+mod output;
+
 use crate::Host;
 use crate::rpc::Dispatcher;
 use ide_core::{DebugSessionId, IdeResult};
@@ -14,7 +16,8 @@ pub async fn debug_start(host: &Host, window: &str, config: LaunchConfig) -> Ide
         super::python::prepare_debug(&mut config, &super::python::paths(&project))?;
     }
     let label = window;
-    let owner = SessionOwner::new(host.owner_window(label), host.window_sink(label));
+    let (sink, flush_output) = output::coalescing_sink(host.window_sink(label));
+    let owner = SessionOwner::new(host.owner_window(label), sink);
     let settings = match host.project(label) {
         Ok(project) => project.settings.effective(),
         Err(_) => host.global_settings.effective(),
@@ -58,7 +61,9 @@ pub async fn debug_start(host: &Host, window: &str, config: LaunchConfig) -> Ide
         let path = ide_sdk::apply::tool_command(&root, &settings, command)?;
         Some(path.to_string_lossy().into_owned())
     };
-    host.debugger.start_with(config, owner, custom).await
+    let result = host.debugger.start_with(config, owner, custom).await;
+    flush_output();
+    result
 }
 
 /// A JVM launched by java-debug writes what java-debug decodes its output as

crates/ide-host/src/ops/debug/output.rs

@@ -0,0 +1,182 @@
+use ide_core::event::DebugEvent;
+use ide_core::{DebugSessionId, IdeEvent, SharedSink};
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+const WINDOW: Duration = Duration::from_millis(50);
+const MAX_BYTES: usize = 64 * 1024;
+
+#[derive(Default)]
+struct Pending {
+    sessions: HashMap<DebugSessionId, (String, String)>,
+    scheduled: bool,
+}
+
+fn flush(pending: &mut Pending, sink: &SharedSink, session: DebugSessionId) {
+    if let Some((category, output)) = pending.sessions.remove(&session) {
+        sink.emit(IdeEvent::Debug { session, event: DebugEvent::Output { category, output } });
+    }
+}
+
+/// All debug output sources (including build output and child sessions) share this
+/// bounded batch before reaching desktop or remote IPC. Control events flush
+/// their session first, so a stop or exit cannot overtake preceding output.
+pub(super) fn coalescing_sink(sink: SharedSink) -> (SharedSink, impl Fn()) {
+    let pending = Arc::new(Mutex::new(Pending::default()));
+    let flush_pending = {
+        let pending = pending.clone();
+        let sink = sink.clone();
+        move || {
+            let mut state = pending.lock();
+            for (session, (category, output)) in state.sessions.drain() {
+                sink.emit(IdeEvent::Debug { session, event: DebugEvent::Output { category, output } });
+            }
+        }
+    };
+    let runtime = tokio::runtime::Handle::current();
+    let output: SharedSink = Arc::new(move |event| {
+        let mut state = pending.lock();
+        match event {
+            IdeEvent::Debug { session, event: DebugEvent::Output { category, output } } => {
+                if output.is_empty() {
+                    return;
+                }
+                if state.sessions.get(&session).is_some_and(|(c, _)| *c != category) {
+                    flush(&mut state, &sink, session);
+                }
+                let mut rest = output.as_str();
+                while !rest.is_empty() {
+                    let (_, text) = state.sessions.entry(session).or_insert_with(|| (category.clone(), String::new()));
+                    let mut end = rest.len().min(MAX_BYTES - text.len());
+                    while !rest.is_char_boundary(end) {
+                        end -= 1;
+                    }
+                    text.push_str(&rest[..end]);
+                    rest = &rest[end..];
+                    if text.len() == MAX_BYTES || !rest.is_empty() {
+                        flush(&mut state, &sink, session);
+                    }
+                }
+                if !state.scheduled && !state.sessions.is_empty() {
+                    state.scheduled = true;
+                    let pending = pending.clone();
+                    let sink = sink.clone();
+                    runtime.spawn(async move {
+                        tokio::time::sleep(WINDOW).await;
+                        let mut state = pending.lock();
+                        for (session, (category, output)) in state.sessions.drain() {
+                            sink.emit(IdeEvent::Debug { session, event: DebugEvent::Output { category, output } });
+                        }
+                        state.scheduled = false;
+                    });
+                }
+            }
+            IdeEvent::Debug { session, event } => {
+                flush(&mut state, &sink, session);
+                sink.emit(IdeEvent::Debug { session, event });
+            }
+            event => sink.emit(event),
+        }
+    });
+    (output, flush_pending)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn setup() -> (SharedSink, Arc<Mutex<Vec<IdeEvent>>>) {
+        let events = Arc::new(Mutex::new(Vec::new()));
+        let saved = events.clone();
+        (coalescing_sink(Arc::new(move |event| saved.lock().push(event))).0, events)
+    }
+
+    fn output(sink: &SharedSink, session: u32, category: &str, output: &str) {
+        sink.emit(IdeEvent::Debug {
+            session: DebugSessionId(session),
+            event: DebugEvent::Output { category: category.into(), output: output.into() },
+        });
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn explicit_start_completion_flush_does_not_duplicate_output() {
+        let events = Arc::new(Mutex::new(Vec::new()));
+        let saved = events.clone();
+        let (sink, flush) = coalescing_sink(Arc::new(move |event| saved.lock().push(event)));
+        output(&sink, 1, "console", "build failed\n");
+        flush();
+        assert_eq!(events.lock().len(), 1);
+        tokio::task::yield_now().await;
+        tokio::time::advance(WINDOW).await;
+        tokio::task::yield_now().await;
+        assert_eq!(events.lock().len(), 1);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn batches_per_session_and_flushes_idle_output() {
+        let (sink, events) = setup();
+        for _ in 0..500 {
+            output(&sink, 1, "stdout", "line\n");
+        }
+        output(&sink, 2, "stdout", "other\n");
+        assert!(events.lock().is_empty());
+        tokio::task::yield_now().await;
+        tokio::time::advance(WINDOW).await;
+        tokio::task::yield_now().await;
+        let events = events.lock();
+        assert_eq!(events.len(), 2);
+        assert!(events.iter().any(|event| matches!(event,
+            IdeEvent::Debug { session: DebugSessionId(1), event: DebugEvent::Output { output, .. } }
+            if output == &"line\n".repeat(500))));
+    }
+
+    #[tokio::test]
+    async fn categories_and_control_events_keep_order() {
+        let (sink, events) = setup();
+        output(&sink, 1, "stdout", "out");
+        output(&sink, 1, "stderr", "err");
+        sink.emit(IdeEvent::Debug { session: DebugSessionId(1), event: DebugEvent::Terminated });
+        let events = events.lock();
+        assert_eq!(events.len(), 3);
+        assert!(matches!(&events[0], IdeEvent::Debug { event: DebugEvent::Output { category, output }, .. }
+            if category == "stdout" && output == "out"));
+        assert!(matches!(&events[1], IdeEvent::Debug { event: DebugEvent::Output { category, output }, .. }
+            if category == "stderr" && output == "err"));
+        assert!(matches!(&events[2], IdeEvent::Debug { event: DebugEvent::Terminated, .. }));
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn terminating_one_session_preserves_other_pending_output() {
+        let (sink, events) = setup();
+        output(&sink, 1, "stdout", "first");
+        output(&sink, 2, "stdout", "second");
+        sink.emit(IdeEvent::Debug { session: DebugSessionId(1), event: DebugEvent::Terminated });
+        assert_eq!(events.lock().len(), 2);
+        tokio::task::yield_now().await;
+        tokio::time::advance(WINDOW).await;
+        tokio::task::yield_now().await;
+        let events = events.lock();
+        assert_eq!(events.len(), 3);
+        assert!(matches!(&events[2],
+            IdeEvent::Debug { session: DebugSessionId(2), event: DebugEvent::Output { output, .. } }
+            if output == "second"));
+    }
+
+    #[tokio::test]
+    async fn large_unicode_events_are_bounded_without_loss() {
+        let (sink, events) = setup();
+        let text = "日志\n".repeat(100_000);
+        output(&sink, 1, "stdout", &text);
+        sink.emit(IdeEvent::Debug { session: DebugSessionId(1), event: DebugEvent::Terminated });
+        let mut combined = String::new();
+        for event in events.lock().iter() {
+            if let IdeEvent::Debug { event: DebugEvent::Output { output, .. }, .. } = event {
+                assert!(output.len() <= MAX_BYTES);
+                combined.push_str(output);
+            }
+        }
+        assert_eq!(combined, text);
+    }
+}

src/features/debugger/__tests__/console.test.ts

@@ -1,9 +1,15 @@
-import { afterEach, describe, expect, it } from "vitest";
-import { appendConsole, clearConsole, useDebug } from "../state";
+import * as settings from "@/platform/workflowSettings";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { adoptSession, appendConsole, appendConsoleQueued, clearConsole, useDebug } from "../state";
 
 const lines = (session: number) => useDebug.getState().console[session] ?? [];
 
-afterEach(() => useDebug.setState({ console: {} }));
+afterEach(() => {
+  vi.useRealTimers();
+  vi.restoreAllMocks();
+  for (const id of Object.keys(useDebug.getState().console)) clearConsole(+id);
+  useDebug.setState({ console: {}, sessions: {}, active: null });
+});
 
 describe("debug console", () => {
   it("renders ANSI colours instead of showing the escapes", () => {
@@ -44,3 +50,49 @@ describe("debug console", () => {
     expect(lines(6)[0]!.spans).toEqual([{ text: "after clear\n" }]);
   });
 });
+
+describe("multiline output", () => {
+  it("splits styled chunks into physical lines and keeps partial-line identity", () => {
+    appendConsole(10, "stdout", "prefix ");
+    const first = lines(10)[0]!;
+    appendConsole(10, "stdout", "\x1b[31mred\n\nlast\x1b[0m");
+    expect(lines(10).map((line) => line.text)).toEqual(["prefix red\n", "\n", "last"]);
+    expect(lines(10)[0]!.id).toBe(first.id);
+    expect(first.text).toBe("prefix ");
+    expect(lines(10)[1]!.spans).toEqual([{ text: "\n", fg: "var(--ansi-red)" }]);
+    appendConsole(10, "stderr", "error\n");
+    appendConsole(10, "stdout", "next\n");
+    expect(lines(10).map((line) => line.category)).toEqual(["stdout", "stdout", "stdout", "stderr", "stdout"]);
+  });
+
+  it("keeps per-frame batching while bounding one event containing 60k lines", () => {
+    vi.useFakeTimers();
+    adoptSession(11);
+    const listener = vi.fn();
+    const unsubscribe = useDebug.subscribe(listener);
+    appendConsoleQueued(11, "stdout", "line\n".repeat(60_000));
+    appendConsoleQueued(11, "stdout", "tail\n");
+    expect(listener).not.toHaveBeenCalled();
+    vi.advanceTimersToNextFrame();
+    expect(listener).toHaveBeenCalledTimes(1);
+    unsubscribe();
+    expect(lines(11).length).toBeLessThanOrEqual(50_000);
+    expect(lines(11).length).toBeGreaterThan(40_000);
+    expect(lines(11).at(-1)!.text).toBe("tail\n");
+    expect(lines(11).every((line) => line.text === "line\n" || line.text === "tail\n")).toBe(true);
+    expect(new Set(lines(11).map((line) => line.id)).size).toBe(lines(11).length);
+  });
+
+  it("trims a multiline event by the configured character buffer", () => {
+    const current = settings.workflowSettings();
+    vi.spyOn(settings, "workflowSettings").mockReturnValue({
+      ...current,
+      console: { ...current.console, overrideBuffer: true, bufferKb: 8 },
+    });
+    appendConsole(12, "stdout", ("x".repeat(4000) + "\n").repeat(6000));
+    const output = lines(12);
+    expect(output.length).toBeGreaterThan(0);
+    expect(output.reduce((n, line) => n + line.text.length + 1, 0)).toBeLessThanOrEqual(8 * 1024);
+    expect(output.every((line) => line.text.length === 4001)).toBe(true);
+  });
+});

src/features/debugger/state.ts

@@ -95,7 +95,7 @@ interface ExecutionPoint {
 
 export const useExecutionPoint = create<ExecutionPoint>(() => ({ path: null, line: 0, top: true, values: {} }));
 
-/** Console buffer: at most this many entries (merged output runs) per session. */
+/** Console buffer: at most this many physical lines per session. */
 const MAX_CONSOLE_LINES = 50_000;
 let lineSeq = 0;
 /** ANSI state per session and output category (a colour can span several output events). */
@@ -269,21 +269,15 @@ export function appendConsole(session: DebugSessionId, category: string, raw: st
   useDebug.setState((s) => {
     const prev = s.console[session] ?? [];
     const next = [...prev];
-    const last = next[next.length - 1];
-    // Merge partial lines of the same category.
-    if (last && last.category === category && !last.text.endsWith("\n")) {
-      next[next.length - 1] = { ...last, text: last.text + text, spans: appendSpans(last.spans, spans) };
-    } else {
-      next.push({ id: ++lineSeq, category, text, spans });
-    }
+    appendConsoleLines(next, category, spans);
     trimConsole(next);
     return { console: { ...s.console, [session]: next } };
   });
 }
 
 /**
- * Queues streamed debuggee output: ANSI parsing happens now, the console store
- * is updated once per animation frame with all pending chunks merged per
+ * Queues streamed debuggee output: ANSI parsing and the console store update happen
+ * once per animation frame with all pending chunks merged per
  * (session, category), so a flood of `output` events can't saturate React.
  */
 export function appendConsoleQueued(session: DebugSessionId, category: string, raw: string) {
@@ -314,13 +308,7 @@ function flushConsoleOutput() {
       if (!text) continue;
       const prev = consoleBySession[session] ?? [];
       const next = [...prev];
-      const last = next[next.length - 1];
-      // Merge partial lines of the same category.
-      if (last && last.category === category && !last.text.endsWith("\n")) {
-        next[next.length - 1] = { ...last, text: last.text + text, spans: appendSpans(last.spans, spans) };
-      } else {
-        next.push({ id: ++lineSeq, category, text, spans });
-      }
+      appendConsoleLines(next, category, spans);
       trimConsole(next);
       consoleBySession[session] = next;
     }
@@ -328,6 +316,30 @@ function flushConsoleOutput() {
   });
 }
 
+/** Split styled output at real line breaks, retaining partial lines across events. */
+function appendConsoleLines(lines: ConsoleLine[], category: string, spans: Span[]) {
+  const last = lines.at(-1);
+  let line =
+    last && last.category === category && !last.text.endsWith("\n") ? { ...last, spans: [...last.spans] } : undefined;
+  if (line) lines[lines.length - 1] = line;
+  for (const span of spans) {
+    let start = 0;
+    while (start < span.text.length) {
+      const newline = span.text.indexOf("\n", start);
+      const end = newline < 0 ? span.text.length : newline + 1;
+      const text = span.text.slice(start, end);
+      if (!line) {
+        line = { id: ++lineSeq, category, text: "", spans: [] };
+        lines.push(line);
+      }
+      line.text += text;
+      line.spans = appendSpans(line.spans, [{ ...span, text }]);
+      if (newline >= 0) line = undefined;
+      start = end;
+    }
+  }
+}
+
 /** Drops the oldest lines once past the line or configured character cap. */
 function trimConsole(lines: ConsoleLine[]) {
   let drop = 0;