fix: prevent TUI freeze on massive bash output and fix spinner rendering (#500)
- Sync OutputSink.push(): eliminate promise chain per chunk, buffer management and onChunk run inline, file writes deferred via queue - 64KB native read buffer (was 4KB): reduces chunk count ~16x - chunkThrottleMs in OutputSink: gate onChunk to every 50ms - BashExecutionComponent streaming throttle: gate + 100-line cap - Remove requestRender from chunk callbacks: spinner drives renders - Remove double sanitization in appendOutput (already done by OutputSink) - Inline SEGMENT_RESET in TUI doRender buffer writes: eliminates O(N) string allocations per frame from #applyLineResets - Cache header Text in BashExecutionComponent (created once, reused) - Gate sixel mask computation behind protocol + passthrough check - Fix spinner: #spinnerFrame made optional, interval calls #updateDisplay - Remove pendingChunks promise chains from bash-executor and bash-interactive
This commit is contained in:
@@ -278,8 +278,8 @@ fn run_pty_sync(
|
||||
let (reader_tx, reader_rx) = mpsc::channel::<ReaderEvent>();
|
||||
let reader_thread = std::thread::spawn(move || {
|
||||
const REPLACEMENT: &str = "\u{FFFD}";
|
||||
const BUF: usize = 4096;
|
||||
let mut buf = [0u8; BUF + 4];
|
||||
const BUF: usize = 65536;
|
||||
let mut buf = vec![0u8; BUF + 4];
|
||||
let mut it = 0;
|
||||
loop {
|
||||
match reader.read(&mut buf[it..BUF]) {
|
||||
|
||||
@@ -582,7 +582,7 @@ async fn run_shell_command(
|
||||
let mut reader_handle = tokio::spawn({
|
||||
let reader_cancel = reader_cancel.clone();
|
||||
async move {
|
||||
read_output(reader_file, on_chunk, reader_cancel, activity_tx).await;
|
||||
Box::pin(read_output(reader_file, on_chunk, reader_cancel, activity_tx)).await;
|
||||
Result::<()>::Ok(())
|
||||
}
|
||||
});
|
||||
@@ -790,8 +790,8 @@ async fn read_output(
|
||||
activity: mpsc::Sender<()>,
|
||||
) {
|
||||
const REPLACEMENT: &str = "\u{FFFD}";
|
||||
const BUF: usize = 4096;
|
||||
let mut buf = [0u8; BUF + 4]; // +4 for max UTF-8 char
|
||||
const BUF: usize = 65536;
|
||||
let mut buf = vec![0u8; BUF + 4]; // +4 for max UTF-8 char
|
||||
let mut it = 0;
|
||||
|
||||
let reader = tokio::fs::File::from_std(reader);
|
||||
|
||||
@@ -69,11 +69,16 @@ export async function executeBash(command: string, options?: BashExecutorOptions
|
||||
onChunk: options?.onChunk,
|
||||
artifactPath: options?.artifactPath,
|
||||
artifactId: options?.artifactId,
|
||||
// Throttle the streaming preview callback to avoid saturating the
|
||||
// event loop when commands produce massive output (e.g. seq 1 50M).
|
||||
chunkThrottleMs: options?.onChunk ? 50 : 0,
|
||||
});
|
||||
|
||||
let pendingChunks = Promise.resolve();
|
||||
// sink.push() is synchronous — buffer management, counters, and onChunk
|
||||
// all run inline. File writes (artifact path) are handled asynchronously
|
||||
// inside the sink. No promise chain needed.
|
||||
const enqueueChunk = (chunk: string) => {
|
||||
pendingChunks = pendingChunks.then(() => sink.push(chunk)).catch(() => {});
|
||||
sink.push(chunk);
|
||||
};
|
||||
|
||||
if (options?.signal?.aborted) {
|
||||
@@ -160,8 +165,6 @@ export async function executeBash(command: string, options?: BashExecutorOptions
|
||||
hardTimeoutDeferred.promise.then(() => ({ kind: "hard-timeout" as const })),
|
||||
]);
|
||||
|
||||
await pendingChunks;
|
||||
|
||||
if (winner.kind === "hard-timeout") {
|
||||
if (shellSession) {
|
||||
resetSession = true;
|
||||
@@ -215,7 +218,6 @@ export async function executeBash(command: string, options?: BashExecutorOptions
|
||||
if (userSignal) {
|
||||
userSignal.removeEventListener("abort", abortHandler);
|
||||
}
|
||||
await pendingChunks;
|
||||
if (resetSession) {
|
||||
shellSessions.delete(sessionKey);
|
||||
}
|
||||
|
||||
@@ -6,13 +6,17 @@ import { sanitizeText } from "@oh-my-pi/pi-natives";
|
||||
import { Container, ImageProtocol, Loader, Spacer, TERMINAL, Text, type TUI } from "@oh-my-pi/pi-tui";
|
||||
import { getSymbolTheme, theme } from "../../modes/theme/theme";
|
||||
import { formatTruncationMetaNotice, type TruncationMeta } from "../../tools/output-meta";
|
||||
import { getSixelLineMask, sanitizeWithOptionalSixelPassthrough } from "../../utils/sixel";
|
||||
import { getSixelLineMask, isSixelPassthroughEnabled, sanitizeWithOptionalSixelPassthrough } from "../../utils/sixel";
|
||||
import { DynamicBorder } from "./dynamic-border";
|
||||
import { truncateToVisualLines } from "./visual-truncate";
|
||||
|
||||
// Preview line limit when not expanded (matches tool execution behavior)
|
||||
const PREVIEW_LINES = 20;
|
||||
const STREAMING_LINE_CAP = PREVIEW_LINES * 5;
|
||||
const MAX_DISPLAY_LINE_CHARS = 4000;
|
||||
// Minimum interval between processing incoming chunks for display (ms).
|
||||
// Chunks arriving faster than this are accumulated and processed in one batch.
|
||||
const CHUNK_THROTTLE_MS = 50;
|
||||
|
||||
export class BashExecutionComponent extends Container {
|
||||
#outputLines: string[] = [];
|
||||
@@ -21,7 +25,10 @@ export class BashExecutionComponent extends Container {
|
||||
#loader: Loader;
|
||||
#truncation?: TruncationMeta;
|
||||
#expanded = false;
|
||||
#displayDirty = false;
|
||||
#chunkGate = false;
|
||||
#contentContainer: Container;
|
||||
#headerText: Text;
|
||||
|
||||
constructor(
|
||||
private readonly command: string,
|
||||
@@ -45,8 +52,8 @@ export class BashExecutionComponent extends Container {
|
||||
this.addChild(this.#contentContainer);
|
||||
|
||||
// Command header
|
||||
const header = new Text(theme.fg(colorKey, theme.bold(`$ ${command}`)), 1, 0);
|
||||
this.#contentContainer.addChild(header);
|
||||
this.#headerText = new Text(theme.fg(colorKey, theme.bold(`$ ${command}`)), 1, 0);
|
||||
this.#contentContainer.addChild(this.#headerText);
|
||||
|
||||
// Loader
|
||||
this.#loader = new Loader(
|
||||
@@ -72,14 +79,22 @@ export class BashExecutionComponent extends Container {
|
||||
|
||||
override invalidate(): void {
|
||||
super.invalidate();
|
||||
this.#displayDirty = false;
|
||||
this.#updateDisplay();
|
||||
}
|
||||
|
||||
appendOutput(chunk: string): void {
|
||||
const clean = sanitizeWithOptionalSixelPassthrough(chunk, sanitizeText);
|
||||
// During high-throughput output (e.g. seq 1 500M), processing every
|
||||
// chunk would saturate the event loop. Instead, accept one chunk per
|
||||
// throttle window and drop the rest — the OutputSink captures everything
|
||||
// for the artifact, and setComplete() replaces with the final output.
|
||||
if (this.#chunkGate) return;
|
||||
this.#chunkGate = true;
|
||||
setTimeout(() => {
|
||||
this.#chunkGate = false;
|
||||
}, CHUNK_THROTTLE_MS);
|
||||
|
||||
// Append to output lines
|
||||
const incomingLines = clean.split("\n");
|
||||
const incomingLines = chunk.split("\n");
|
||||
if (this.#outputLines.length > 0 && incomingLines.length > 0) {
|
||||
const lastIndex = this.#outputLines.length - 1;
|
||||
const mergedLines = [`${this.#outputLines[lastIndex]}${incomingLines[0]}`, ...incomingLines.slice(1)];
|
||||
@@ -90,7 +105,12 @@ export class BashExecutionComponent extends Container {
|
||||
this.#outputLines.push(...this.#clampLinesPreservingSixel(incomingLines));
|
||||
}
|
||||
|
||||
this.#updateDisplay();
|
||||
// Cap stored lines during streaming to avoid unbounded memory growth
|
||||
if (this.#outputLines.length > STREAMING_LINE_CAP) {
|
||||
this.#outputLines = this.#outputLines.slice(-STREAMING_LINE_CAP);
|
||||
}
|
||||
|
||||
this.#displayDirty = true;
|
||||
}
|
||||
|
||||
setComplete(
|
||||
@@ -115,6 +135,14 @@ export class BashExecutionComponent extends Container {
|
||||
this.#updateDisplay();
|
||||
}
|
||||
|
||||
override render(width: number): string[] {
|
||||
if (this.#displayDirty) {
|
||||
this.#displayDirty = false;
|
||||
this.#updateDisplay();
|
||||
}
|
||||
return super.render(width);
|
||||
}
|
||||
|
||||
#updateDisplay(): void {
|
||||
const availableLines = this.#outputLines;
|
||||
|
||||
@@ -122,15 +150,16 @@ export class BashExecutionComponent extends Container {
|
||||
const previewLogicalLines = availableLines.slice(-PREVIEW_LINES);
|
||||
const hiddenLineCount = availableLines.length - previewLogicalLines.length;
|
||||
const sixelLineMask =
|
||||
TERMINAL.imageProtocol === ImageProtocol.Sixel ? getSixelLineMask(availableLines) : undefined;
|
||||
TERMINAL.imageProtocol === ImageProtocol.Sixel && isSixelPassthroughEnabled()
|
||||
? getSixelLineMask(availableLines)
|
||||
: undefined;
|
||||
const hasSixelOutput = sixelLineMask?.some(Boolean) ?? false;
|
||||
|
||||
// Rebuild content container
|
||||
this.#contentContainer.clear();
|
||||
|
||||
// Command header
|
||||
const header = new Text(theme.fg("bashMode", theme.bold(`$ ${this.command}`)), 1, 0);
|
||||
this.#contentContainer.addChild(header);
|
||||
this.#contentContainer.addChild(this.#headerText);
|
||||
|
||||
// Output
|
||||
if (availableLines.length > 0) {
|
||||
|
||||
@@ -72,9 +72,8 @@ export class PythonExecutionComponent extends Container {
|
||||
}
|
||||
|
||||
appendOutput(chunk: string): void {
|
||||
const clean = sanitizeText(chunk);
|
||||
|
||||
const newLines = clean.split("\n").map(line => this.#clampDisplayLine(line));
|
||||
// Chunk is pre-sanitized by OutputSink.push() — no need to sanitize again.
|
||||
const newLines = chunk.split("\n").map(line => this.#clampDisplayLine(line));
|
||||
if (this.#outputLines.length > 0 && newLines.length > 0) {
|
||||
this.#outputLines[this.#outputLines.length - 1] = this.#clampDisplayLine(
|
||||
`${this.#outputLines[this.#outputLines.length - 1]}${newLines[0]}`,
|
||||
|
||||
@@ -105,17 +105,16 @@ export class ToolExecutionComponent extends Container {
|
||||
// Cached converted images for Kitty protocol (which requires PNG), keyed by index
|
||||
#convertedImages: Map<number, { data: string; mimeType: string }> = new Map();
|
||||
// Spinner animation for partial task results
|
||||
#spinnerFrame = 0;
|
||||
#spinnerFrame?: number;
|
||||
#spinnerInterval?: NodeJS.Timeout;
|
||||
// Track if args are still being streamed (for edit/write spinner)
|
||||
#argsComplete = false;
|
||||
#renderState: {
|
||||
spinnerFrame: number;
|
||||
spinnerFrame?: number;
|
||||
expanded: boolean;
|
||||
isPartial: boolean;
|
||||
renderContext?: Record<string, unknown>;
|
||||
} = {
|
||||
spinnerFrame: 0,
|
||||
expanded: false,
|
||||
isPartial: true,
|
||||
};
|
||||
@@ -328,10 +327,9 @@ export class ToolExecutionComponent extends Container {
|
||||
this.#spinnerInterval = setInterval(() => {
|
||||
const frameCount = theme.spinnerFrames.length;
|
||||
if (frameCount === 0) return;
|
||||
this.#spinnerFrame = (this.#spinnerFrame + 1) % frameCount;
|
||||
this.#spinnerFrame = ((this.#spinnerFrame ?? -1) + 1) % frameCount;
|
||||
this.#renderState.spinnerFrame = this.#spinnerFrame;
|
||||
this.#ui.requestRender();
|
||||
// NO updateDisplay() — existing component closures read from renderState
|
||||
}, 80);
|
||||
} else if (!needsSpinner && this.#spinnerInterval) {
|
||||
clearInterval(this.#spinnerInterval);
|
||||
@@ -346,6 +344,7 @@ export class ToolExecutionComponent extends Container {
|
||||
if (this.#spinnerInterval) {
|
||||
clearInterval(this.#spinnerInterval);
|
||||
this.#spinnerInterval = undefined;
|
||||
this.#spinnerFrame = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -690,7 +690,6 @@ export class CommandController {
|
||||
chunk => {
|
||||
if (this.ctx.bashComponent) {
|
||||
this.ctx.bashComponent.appendOutput(chunk);
|
||||
this.ctx.ui.requestRender();
|
||||
}
|
||||
},
|
||||
{ excludeFromContext },
|
||||
@@ -732,7 +731,6 @@ export class CommandController {
|
||||
chunk => {
|
||||
if (this.ctx.pythonComponent) {
|
||||
this.ctx.pythonComponent.appendOutput(chunk);
|
||||
this.ctx.ui.requestRender();
|
||||
}
|
||||
},
|
||||
{ excludeFromContext },
|
||||
|
||||
@@ -32,6 +32,8 @@ export interface OutputSinkOptions {
|
||||
artifactId?: string;
|
||||
spillThreshold?: number;
|
||||
onChunk?: (chunk: string) => void;
|
||||
/** Minimum ms between onChunk calls. 0 = every chunk (default). */
|
||||
chunkThrottleMs?: number;
|
||||
}
|
||||
|
||||
export interface TruncationResult {
|
||||
@@ -521,6 +523,7 @@ export class OutputSink {
|
||||
#totalBytes = 0;
|
||||
#sawData = false;
|
||||
#truncated = false;
|
||||
#lastChunkTime = 0;
|
||||
|
||||
#file?: {
|
||||
path: string;
|
||||
@@ -528,22 +531,46 @@ export class OutputSink {
|
||||
sink: Bun.FileSink;
|
||||
};
|
||||
|
||||
// Queue of chunks waiting for the file sink to be created.
|
||||
#pendingFileWrites?: string[];
|
||||
#fileReady = false;
|
||||
|
||||
readonly #artifactPath?: string;
|
||||
readonly #artifactId?: string;
|
||||
readonly #spillThreshold: number;
|
||||
readonly #onChunk?: (chunk: string) => void;
|
||||
readonly #chunkThrottleMs: number;
|
||||
|
||||
constructor(options?: OutputSinkOptions) {
|
||||
const { artifactPath, artifactId, spillThreshold = DEFAULT_MAX_BYTES, onChunk } = options ?? {};
|
||||
const {
|
||||
artifactPath,
|
||||
artifactId,
|
||||
spillThreshold = DEFAULT_MAX_BYTES,
|
||||
onChunk,
|
||||
chunkThrottleMs = 0,
|
||||
} = options ?? {};
|
||||
this.#artifactPath = artifactPath;
|
||||
this.#artifactId = artifactId;
|
||||
this.#spillThreshold = spillThreshold;
|
||||
this.#onChunk = onChunk;
|
||||
this.#chunkThrottleMs = chunkThrottleMs;
|
||||
}
|
||||
|
||||
async push(chunk: string): Promise<void> {
|
||||
/**
|
||||
* Push a chunk of output. The buffer management and onChunk callback run
|
||||
* synchronously. File sink writes are deferred and serialized internally.
|
||||
*/
|
||||
push(chunk: string): void {
|
||||
chunk = sanitizeWithOptionalSixelPassthrough(chunk, sanitizeText);
|
||||
this.#onChunk?.(chunk);
|
||||
|
||||
// Throttled onChunk: only call the callback when enough time has passed.
|
||||
if (this.#onChunk) {
|
||||
const now = Date.now();
|
||||
if (now - this.#lastChunkTime >= this.#chunkThrottleMs) {
|
||||
this.#lastChunkTime = now;
|
||||
this.#onChunk(chunk);
|
||||
}
|
||||
}
|
||||
|
||||
const dataBytes = Buffer.byteLength(chunk, "utf-8");
|
||||
this.#totalBytes += dataBytes;
|
||||
@@ -556,10 +583,9 @@ export class OutputSink {
|
||||
const threshold = this.#spillThreshold;
|
||||
const willOverflow = this.#bufferBytes + dataBytes > threshold;
|
||||
|
||||
// Write to file if already spilling or about to overflow
|
||||
if (this.#file != null || willOverflow) {
|
||||
const sink = await this.#ensureFileSink();
|
||||
await sink?.write(chunk);
|
||||
// Write to artifact file if configured and past the threshold
|
||||
if (this.#artifactPath && (this.#file != null || willOverflow)) {
|
||||
this.#writeToFile(chunk);
|
||||
}
|
||||
|
||||
if (!willOverflow) {
|
||||
@@ -589,14 +615,64 @@ export class OutputSink {
|
||||
if (this.#file) this.#truncated = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Write a chunk to the artifact file. Handles the async file sink creation
|
||||
* by queuing writes until the sink is ready, then draining synchronously.
|
||||
*/
|
||||
#writeToFile(chunk: string): void {
|
||||
if (this.#fileReady && this.#file) {
|
||||
// Fast path: file sink exists, write synchronously
|
||||
this.#file.sink.write(chunk);
|
||||
return;
|
||||
}
|
||||
// File sink not yet created — queue this chunk and kick off creation
|
||||
if (!this.#pendingFileWrites) {
|
||||
this.#pendingFileWrites = [chunk];
|
||||
void this.#createFileSink();
|
||||
} else {
|
||||
this.#pendingFileWrites.push(chunk);
|
||||
}
|
||||
}
|
||||
|
||||
async #createFileSink(): Promise<void> {
|
||||
if (!this.#artifactPath || this.#fileReady) return;
|
||||
try {
|
||||
const sink = Bun.file(this.#artifactPath).writer();
|
||||
this.#file = { path: this.#artifactPath, artifactId: this.#artifactId, sink };
|
||||
|
||||
// Flush existing buffer to file BEFORE it gets trimmed further.
|
||||
if (this.#buffer.length > 0) {
|
||||
sink.write(this.#buffer);
|
||||
}
|
||||
|
||||
// Drain any chunks that arrived while the sink was being created
|
||||
if (this.#pendingFileWrites) {
|
||||
for (const pending of this.#pendingFileWrites) {
|
||||
sink.write(pending);
|
||||
}
|
||||
this.#pendingFileWrites = undefined;
|
||||
}
|
||||
|
||||
this.#fileReady = true;
|
||||
} catch {
|
||||
try {
|
||||
await this.#file?.sink?.end();
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
this.#file = undefined;
|
||||
this.#pendingFileWrites = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
createInput(): WritableStream<Uint8Array | string> {
|
||||
const dec = new TextDecoder("utf-8", { ignoreBOM: true });
|
||||
const finalize = async () => {
|
||||
await this.push(dec.decode());
|
||||
const finalize = () => {
|
||||
this.push(dec.decode());
|
||||
};
|
||||
return new WritableStream({
|
||||
write: async chunk => {
|
||||
await this.push(typeof chunk === "string" ? chunk : dec.decode(chunk, { stream: true }));
|
||||
write: chunk => {
|
||||
this.push(typeof chunk === "string" ? chunk : dec.decode(chunk, { stream: true }));
|
||||
},
|
||||
close: finalize,
|
||||
abort: finalize,
|
||||
@@ -620,32 +696,6 @@ export class OutputSink {
|
||||
artifactId: this.#file?.artifactId,
|
||||
};
|
||||
}
|
||||
|
||||
// -- private ---------------------------------------------------------------
|
||||
|
||||
async #ensureFileSink(): Promise<Bun.FileSink | null> {
|
||||
if (!this.#artifactPath) return null;
|
||||
if (this.#file) return this.#file.sink;
|
||||
|
||||
try {
|
||||
const sink = Bun.file(this.#artifactPath).writer();
|
||||
this.#file = { path: this.#artifactPath, artifactId: this.#artifactId, sink };
|
||||
|
||||
// Flush existing buffer to file BEFORE it gets trimmed further.
|
||||
if (this.#buffer.length > 0) {
|
||||
await sink.write(this.#buffer);
|
||||
}
|
||||
return sink;
|
||||
} catch {
|
||||
try {
|
||||
await this.#file?.sink?.end();
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
this.#file = undefined;
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
|
||||
@@ -295,7 +295,6 @@ export async function runInteractiveBashPty(
|
||||
},
|
||||
): Promise<BashInteractiveResult> {
|
||||
const sink = new OutputSink({ artifactPath: options.artifactPath, artifactId: options.artifactId });
|
||||
let pendingChunks = Promise.resolve();
|
||||
const result = await ui.custom<BashInteractiveResult>(
|
||||
(tui, uiTheme, _keybindings, done) => {
|
||||
const session = new PtySession();
|
||||
@@ -309,7 +308,6 @@ export async function runInteractiveBashPty(
|
||||
tui.requestRender();
|
||||
void (async () => {
|
||||
await component.flushOutput();
|
||||
await pendingChunks;
|
||||
const summary = await sink.dump();
|
||||
done({
|
||||
exitCode: run.exitCode,
|
||||
@@ -362,15 +360,13 @@ export async function runInteractiveBashPty(
|
||||
if (finished || err || !chunk) return;
|
||||
component.appendOutput(chunk);
|
||||
const normalizedChunk = normalizeCaptureChunk(chunk);
|
||||
pendingChunks = pendingChunks.then(() => sink.push(normalizedChunk)).catch(() => {});
|
||||
sink.push(normalizedChunk);
|
||||
tui.requestRender();
|
||||
},
|
||||
)
|
||||
.then(finalize)
|
||||
.catch(error => {
|
||||
pendingChunks = pendingChunks
|
||||
.then(() => sink.push(`PTY error: ${error instanceof Error ? error.message : String(error)}\n`))
|
||||
.catch(() => {});
|
||||
sink.push(`PTY error: ${error instanceof Error ? error.message : String(error)}\n`);
|
||||
finalize({ exitCode: undefined, cancelled: false, timedOut: false });
|
||||
});
|
||||
return component;
|
||||
|
||||
@@ -289,8 +289,8 @@ export class PythonTool implements AgentTool<typeof pythonSchema> {
|
||||
const executorOptions: PythonExecutorOptions = {
|
||||
...baseExecutorOptions,
|
||||
reset: isFirstCell ? reset : false,
|
||||
onChunk: async chunk => {
|
||||
await outputSink!.push(chunk);
|
||||
onChunk: chunk => {
|
||||
outputSink!.push(chunk);
|
||||
},
|
||||
};
|
||||
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
|
||||
import { BashExecutionComponent } from "@oh-my-pi/pi-coding-agent/modes/components/bash-execution";
|
||||
import { getThemeByName, setThemeInstance } from "@oh-my-pi/pi-coding-agent/modes/theme/theme";
|
||||
import { sanitizeWithOptionalSixelPassthrough } from "@oh-my-pi/pi-coding-agent/utils/sixel";
|
||||
import { sanitizeText } from "@oh-my-pi/pi-natives";
|
||||
import type { TUI } from "@oh-my-pi/pi-tui";
|
||||
|
||||
const SIXEL = "\x1bPqabc\x1b\\";
|
||||
@@ -66,11 +68,76 @@ describe("BashExecutionComponent SIXEL sanitization", () => {
|
||||
delete Bun.env.PI_FORCE_IMAGE_PROTOCOL;
|
||||
delete Bun.env.PI_ALLOW_SIXEL_PASSTHROUGH;
|
||||
|
||||
const component = new BashExecutionComponent("echo sixel", ui, false);
|
||||
component.appendOutput(SIXEL);
|
||||
// appendOutput receives pre-sanitized chunks from OutputSink.
|
||||
// Simulate that: sanitize before passing to the component.
|
||||
const sanitized = sanitizeWithOptionalSixelPassthrough(SIXEL, sanitizeText);
|
||||
const component = new BashExecutionComponent("test sixel", ui, false);
|
||||
component.appendOutput(sanitized);
|
||||
component.setComplete(0, false);
|
||||
|
||||
expect(component.getOutput()).not.toContain("\x1bPq");
|
||||
expect(component.getOutput()).toBe("");
|
||||
});
|
||||
});
|
||||
|
||||
describe("BashExecutionComponent streaming throttle", () => {
|
||||
const ui = { requestRender: () => {} } as unknown as TUI;
|
||||
|
||||
beforeEach(async () => {
|
||||
const theme = await getThemeByName("dark");
|
||||
expect(theme).toBeDefined();
|
||||
setThemeInstance(theme!);
|
||||
});
|
||||
|
||||
it("caps stored lines during streaming", () => {
|
||||
const component = new BashExecutionComponent("test", ui, false);
|
||||
|
||||
// Flood with 500 lines in one chunk (exceeds STREAMING_LINE_CAP of 100)
|
||||
const lines = Array.from({ length: 500 }, (_, i) => `line${i}`).join("\n");
|
||||
component.appendOutput(lines);
|
||||
|
||||
// Internal lines should be capped (we can't read #outputLines directly,
|
||||
// but getOutput() returns the joined lines — it should have at most ~100 lines)
|
||||
const output = component.getOutput();
|
||||
const outputLineCount = output.split("\n").length;
|
||||
expect(outputLineCount).toBeLessThanOrEqual(101); // 100 cap + possible partial
|
||||
// Should retain the tail, not the head
|
||||
expect(output).toContain("line499");
|
||||
expect(output).not.toContain("line0\n");
|
||||
});
|
||||
|
||||
it("gate drops rapid chunks", async () => {
|
||||
const component = new BashExecutionComponent("test", ui, false);
|
||||
|
||||
// Send 100 chunks rapidly (all in same tick, before setTimeout fires)
|
||||
for (let i = 0; i < 100; i++) {
|
||||
component.appendOutput(`chunk${i}\n`);
|
||||
}
|
||||
|
||||
// Only the first chunk should have been processed (gate blocks the rest)
|
||||
const output = component.getOutput();
|
||||
expect(output).toContain("chunk0");
|
||||
expect(output).not.toContain("chunk99");
|
||||
|
||||
// After the gate timer expires, the next chunk is accepted
|
||||
await Bun.sleep(60); // CHUNK_THROTTLE_MS is 50
|
||||
component.appendOutput("after_gate\n");
|
||||
expect(component.getOutput()).toContain("after_gate");
|
||||
});
|
||||
|
||||
it("setComplete replaces streaming output with final output", () => {
|
||||
const component = new BashExecutionComponent("test", ui, false);
|
||||
|
||||
// Stream some partial output
|
||||
component.appendOutput("streaming_line\n");
|
||||
|
||||
// Complete with different final output
|
||||
component.setComplete(0, false, { output: "final_line_1\nfinal_line_2" });
|
||||
|
||||
const output = component.getOutput();
|
||||
expect(output).toContain("final_line_1");
|
||||
expect(output).toContain("final_line_2");
|
||||
// Streaming output is replaced, not appended
|
||||
expect(output).not.toContain("streaming_line");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -194,10 +194,11 @@ describe("executeBash", () => {
|
||||
chunks.push(chunk);
|
||||
},
|
||||
});
|
||||
const combined = chunks.join("");
|
||||
// At least one chunk should have been delivered to onChunk
|
||||
expect(chunks.length).toBeGreaterThan(0);
|
||||
const combined = chunks.join("");
|
||||
expect(combined).toContain("line1");
|
||||
expect(combined).toContain("line20");
|
||||
// Final result always has the complete output regardless of chunk throttle
|
||||
expect(result.output).toContain("line1");
|
||||
expect(result.output).toContain("line20");
|
||||
});
|
||||
@@ -206,23 +207,61 @@ describe("executeBash", () => {
|
||||
if (process.platform === "win32") {
|
||||
return;
|
||||
}
|
||||
let totalBytes = 0;
|
||||
let sawChunk = false;
|
||||
const result = await executeBash("awk 'BEGIN { for (i = 0; i < 100000; i++) printf \"a\" }'", {
|
||||
cwd: tempDir,
|
||||
timeout: 5000,
|
||||
onChunk: chunk => {
|
||||
onChunk: () => {
|
||||
sawChunk = true;
|
||||
totalBytes += Buffer.byteLength(chunk, "utf-8");
|
||||
},
|
||||
});
|
||||
expect(sawChunk).toBe(true);
|
||||
expect(totalBytes).toBe(100000);
|
||||
expect(result.totalBytes).toBe(100000);
|
||||
expect(result.outputBytes).toBeLessThanOrEqual(DEFAULT_MAX_BYTES);
|
||||
expect(result.output).toContain("a");
|
||||
});
|
||||
|
||||
it("handles multi-million line output without freeze or OOM", async () => {
|
||||
if (process.platform === "win32") return;
|
||||
|
||||
// 5 million lines ~= 40MB of output. Before the 64KB read buffer and
|
||||
// direct-push fixes, this would freeze or OOM the process.
|
||||
const lineCount = 5_000_000;
|
||||
let chunkCount = 0;
|
||||
const start = Date.now();
|
||||
const result = await executeBash(`seq 1 ${lineCount}`, {
|
||||
cwd: tempDir,
|
||||
timeout: 30_000,
|
||||
onChunk: () => {
|
||||
chunkCount++;
|
||||
},
|
||||
});
|
||||
const elapsed = Date.now() - start;
|
||||
|
||||
// Should complete, not hang or OOM
|
||||
expect(result.exitCode).toBe(0);
|
||||
expect(result.cancelled).toBe(false);
|
||||
|
||||
// Output summary should reflect all lines
|
||||
expect(result.totalLines).toBeGreaterThanOrEqual(lineCount);
|
||||
|
||||
// Truncated output should be within the spill threshold
|
||||
expect(result.outputBytes).toBeLessThanOrEqual(DEFAULT_MAX_BYTES);
|
||||
|
||||
// The tail of the output should contain numbers near the end of the range.
|
||||
// The exact last number may be split across a truncation boundary, so
|
||||
// check for a number within the last 1000 lines.
|
||||
expect(result.output).toContain(String(lineCount - 500));
|
||||
|
||||
// With 64KB read buffer, ~40MB should produce ~600 chunks, not 5M.
|
||||
// Allow generous headroom but ensure it's orders of magnitude below lineCount.
|
||||
expect(chunkCount).toBeLessThan(lineCount / 100);
|
||||
|
||||
// Should complete in reasonable time (not frozen). On a modern machine
|
||||
// seq 1 5000000 itself takes ~0.5s; with JS overhead allow 20s.
|
||||
expect(elapsed).toBeLessThan(20_000);
|
||||
}, 35_000);
|
||||
|
||||
it("sources snapshot env vars across session commands", async () => {
|
||||
if (process.platform === "win32") {
|
||||
return;
|
||||
|
||||
+5
-16
@@ -890,17 +890,6 @@ export class TUI extends Container {
|
||||
return result;
|
||||
}
|
||||
|
||||
#applyLineResets(lines: string[]): string[] {
|
||||
const reset = SEGMENT_RESET;
|
||||
for (let i = 0; i < lines.length; i++) {
|
||||
const line = lines[i];
|
||||
if (!TERMINAL.isImageLine(line)) {
|
||||
lines[i] = line + reset;
|
||||
}
|
||||
}
|
||||
return lines;
|
||||
}
|
||||
|
||||
/** Splice overlay content into a base line at a specific column. Single-pass optimized. */
|
||||
#compositeLineAt(
|
||||
baseLine: string,
|
||||
@@ -1001,11 +990,9 @@ export class TUI extends Container {
|
||||
newLines = this.#compositeOverlays(newLines, width, height);
|
||||
}
|
||||
|
||||
// Extract cursor position before applying line resets (marker must be found first)
|
||||
// Extract cursor position (marker must be found before diff comparison)
|
||||
const cursorPos = this.#extractCursorPosition(newLines, height);
|
||||
|
||||
newLines = this.#applyLineResets(newLines);
|
||||
|
||||
// Width changed - need full re-render (line wrapping changes)
|
||||
const widthChanged = this.#previousWidth !== 0 && this.#previousWidth !== width;
|
||||
|
||||
@@ -1014,9 +1001,11 @@ export class TUI extends Container {
|
||||
this.#fullRedrawCount += 1;
|
||||
let buffer = "\x1b[?2026h"; // Begin synchronized output
|
||||
if (clear) buffer += "\x1b[3J\x1b[2J\x1b[H"; // Clear scrollback, screen, and home
|
||||
const reset = SEGMENT_RESET;
|
||||
for (let i = 0; i < newLines.length; i++) {
|
||||
if (i > 0) buffer += "\r\n";
|
||||
buffer += newLines[i];
|
||||
const line = newLines[i];
|
||||
buffer += TERMINAL.isImageLine(line) ? line : line + reset;
|
||||
}
|
||||
buffer += "\x1b[?2026l"; // End synchronized output
|
||||
this.terminal.write(buffer);
|
||||
@@ -1211,7 +1200,7 @@ export class TUI extends Container {
|
||||
].join("\n");
|
||||
throw new Error(errorMsg);
|
||||
}
|
||||
buffer += line;
|
||||
buffer += isImage ? line : line + SEGMENT_RESET;
|
||||
}
|
||||
|
||||
// Track where cursor ended up after rendering
|
||||
|
||||
Reference in New Issue
Block a user