fix(core): throttle shell tool live text updates

packages/core/src/tools/shell.test.ts

@@ -736,7 +736,7 @@ describe('ShellTool', () => {
     describe('Streaming to `updateOutput`', () => {
       let updateOutputMock: Mock;
       beforeEach(() => {
-        vi.useFakeTimers({ toFake: ['Date'] });
+        vi.useFakeTimers({ toFake: ['Date', 'setTimeout', 'clearTimeout'] });
         updateOutputMock = vi.fn();
       });
       afterEach(() => {
@@ -789,6 +789,364 @@ describe('ShellTool', () => {
         });
         await promise;
       });
+
+      it('should throttle live text updates while preserving the latest output', async () => {
+        const invocation = shellTool.build({
+          command: 'npm test',
+          is_background: false,
+        });
+        const promise = invocation.execute(mockAbortSignal, updateOutputMock);
+
+        // Leading-edge fires immediately
+        mockShellOutputCallback({ type: 'data', chunk: 'line 1' });
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+        expect(updateOutputMock).toHaveBeenLastCalledWith('line 1');
+
+        // Suppressed: trailing flush scheduled
+        mockShellOutputCallback({ type: 'data', chunk: 'line 2' });
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+
+        // Advance time: trailing flush fires, emitting 'line 2'
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS + 1);
+        expect(updateOutputMock).toHaveBeenCalledTimes(2);
+        expect(updateOutputMock).toHaveBeenLastCalledWith('line 2');
+
+        // Advance time past the interval window again so next chunk fires immediately
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS + 1);
+
+        mockShellOutputCallback({ type: 'data', chunk: 'line 3' });
+        expect(updateOutputMock).toHaveBeenCalledTimes(3);
+        expect(updateOutputMock).toHaveBeenLastCalledWith('line 3');
+
+        resolveExecutionPromise({
+          rawOutput: Buffer.from('line 1\nline 2\nline 3'),
+          output: 'line 1\nline 2\nline 3',
+          exitCode: 0,
+          signal: null,
+          error: null,
+          aborted: false,
+          pid: 12345,
+          executionMethod: 'child_process',
+        });
+        await promise;
+      });
+
+      it('should flush the last suppressed text chunk when the command goes quiet', async () => {
+        const invocation = shellTool.build({
+          command: 'long-running-cmd',
+          is_background: false,
+        });
+        const promise = invocation.execute(mockAbortSignal, updateOutputMock);
+
+        // Leading-edge update
+        mockShellOutputCallback({ type: 'data', chunk: 'progress: 0%' });
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+
+        // Suppressed: within the throttle window
+        mockShellOutputCallback({ type: 'data', chunk: 'progress: 50%' });
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+
+        // Advance time to trigger the trailing flush timer
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS + 1);
+
+        // The trailing flush must have fired with the latest suppressed chunk
+        expect(updateOutputMock).toHaveBeenCalledTimes(2);
+        expect(updateOutputMock).toHaveBeenLastCalledWith('progress: 50%');
+
+        resolveExecutionPromise({
+          rawOutput: Buffer.from('progress: 50%'),
+          output: 'progress: 50%',
+          exitCode: 0,
+          signal: null,
+          error: null,
+          aborted: false,
+          pid: 12345,
+          executionMethod: 'child_process',
+        });
+        await promise;
+      });
+
+      it('should coalesce 3+ rapid text chunks within a window into a single trailing flush', async () => {
+        // Regression: in one throttle window, the leading-edge chunk fires
+        // immediately, and any subsequent chunks (regardless of count) are
+        // collapsed into ONE trailing flush carrying the latest text. The
+        // timer must not be repeatedly rescheduled per chunk — that would
+        // be wasteful and (depending on the math) could push the flush
+        // beyond the original window.
+        const invocation = shellTool.build({
+          command: 'streaming-cmd',
+          is_background: false,
+        });
+        const promise = invocation.execute(mockAbortSignal, updateOutputMock);
+
+        // Leading edge: fires immediately at t=0
+        mockShellOutputCallback({ type: 'data', chunk: 'chunk 1' });
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+        expect(updateOutputMock).toHaveBeenLastCalledWith('chunk 1');
+
+        // Three rapid suppressed chunks within the same window. None of
+        // these should fire updateOutput synchronously, and the trailing
+        // flush should not have run yet.
+        await vi.advanceTimersByTimeAsync(50);
+        mockShellOutputCallback({ type: 'data', chunk: 'chunk 2' });
+        await vi.advanceTimersByTimeAsync(50);
+        mockShellOutputCallback({ type: 'data', chunk: 'chunk 3' });
+        await vi.advanceTimersByTimeAsync(50);
+        mockShellOutputCallback({ type: 'data', chunk: 'chunk 4' });
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+
+        // Drain the throttle window. The single trailing flush should
+        // fire exactly once and carry the LATEST suppressed chunk.
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS + 1);
+        expect(updateOutputMock).toHaveBeenCalledTimes(2);
+        expect(updateOutputMock).toHaveBeenLastCalledWith('chunk 4');
+
+        resolveExecutionPromise({
+          rawOutput: Buffer.from('chunk 1chunk 2chunk 3chunk 4'),
+          output: 'chunk 1chunk 2chunk 3chunk 4',
+          exitCode: 0,
+          signal: null,
+          error: null,
+          aborted: false,
+          pid: 12345,
+          executionMethod: 'child_process',
+        });
+        await promise;
+      });
+
+      it('should cancel a pending trailing flush when the command completes', async () => {
+        // Lifecycle invariant: if the command resolves while a trailing
+        // flush timer is pending, the timer MUST be cancelled. Otherwise
+        // the timer would fire after `execute()` returns and trigger a
+        // phantom updateOutput call against stale `cumulativeOutput`,
+        // racing against the consumer that has already moved on.
+        const invocation = shellTool.build({
+          command: 'quick-cmd',
+          is_background: false,
+        });
+        const promise = invocation.execute(mockAbortSignal, updateOutputMock);
+
+        // Leading-edge update + suppressed chunk (timer pending)
+        mockShellOutputCallback({ type: 'data', chunk: 'first' });
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+        mockShellOutputCallback({ type: 'data', chunk: 'second' });
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+
+        // Resolve BEFORE the throttle window elapses. No further chunks.
+        resolveExecutionPromise({
+          rawOutput: Buffer.from('first\nsecond'),
+          output: 'first\nsecond',
+          exitCode: 0,
+          signal: null,
+          error: null,
+          aborted: false,
+          pid: 12345,
+          executionMethod: 'child_process',
+        });
+        await promise;
+
+        // Advancing time past the original window must not produce a
+        // late updateOutput call — the timer was cancelled on settle.
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS * 2);
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+      });
+
+      it('should not fire a duplicate trailing flush after a leading-edge update', async () => {
+        // After a trailing flush emits in window N, the next chunk in
+        // window N+1 takes the leading-edge path. `doUpdate()` is the
+        // single point that cancels any pending trailing-flush timer,
+        // so even if a stale timer were somehow still scheduled when a
+        // leading-edge update fires, no duplicate updateOutput call can
+        // escape. This test asserts the end-to-end invariant: suppress
+        // → trailing flush → leading-edge → suppress → trailing flush
+        // produces exactly the expected sequence with no duplicates.
+        const invocation = shellTool.build({
+          command: 'multi-window-cmd',
+          is_background: false,
+        });
+        const promise = invocation.execute(mockAbortSignal, updateOutputMock);
+
+        // Window 1: leading-edge 'a' at t=0
+        mockShellOutputCallback({ type: 'data', chunk: 'a' });
+        expect(updateOutputMock).toHaveBeenCalledTimes(1);
+        expect(updateOutputMock).toHaveBeenLastCalledWith('a');
+
+        // Window 1: suppressed 'b' schedules trailing flush
+        await vi.advanceTimersByTimeAsync(100);
+        mockShellOutputCallback({ type: 'data', chunk: 'b' });
+        expect(updateOutputMock).toHaveBeenCalledTimes(1);
+
+        // Trailing flush fires at the window boundary with 'b'
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS);
+        expect(updateOutputMock).toHaveBeenCalledTimes(2);
+        expect(updateOutputMock).toHaveBeenLastCalledWith('b');
+
+        // Window 2: advance past the interval, next chunk takes the
+        // leading-edge path. If `doUpdate()` failed to cancel the (now
+        // already-fired) timer, no harm; if doUpdate fails to cancel a
+        // *future* timer scheduled later, we'd see duplicates below.
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS + 1);
+        mockShellOutputCallback({ type: 'data', chunk: 'c' });
+        expect(updateOutputMock).toHaveBeenCalledTimes(3);
+        expect(updateOutputMock).toHaveBeenLastCalledWith('c');
+
+        // Window 2: suppressed 'd' schedules another trailing flush
+        await vi.advanceTimersByTimeAsync(50);
+        mockShellOutputCallback({ type: 'data', chunk: 'd' });
+        expect(updateOutputMock).toHaveBeenCalledTimes(3);
+
+        // The trailing flush fires exactly once with 'd'.
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS);
+        expect(updateOutputMock).toHaveBeenCalledTimes(4);
+        expect(updateOutputMock).toHaveBeenLastCalledWith('d');
+
+        // Drain a long quiet period — no spurious late updates from
+        // any zombie timers.
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS * 5);
+        expect(updateOutputMock).toHaveBeenCalledTimes(4);
+
+        resolveExecutionPromise({
+          rawOutput: Buffer.from('abcd'),
+          output: 'abcd',
+          exitCode: 0,
+          signal: null,
+          error: null,
+          aborted: false,
+          pid: 12345,
+          executionMethod: 'child_process',
+        });
+        await promise;
+      });
+
+      it('should cancel a pending trailing flush when the abort signal fires', async () => {
+        // If the user cancels (or the timeout fires) while a trailing
+        // flush is pending, the abort listener must cancel the timer.
+        // Otherwise we'd flash a stale frame between the abort and the
+        // result promise settling with `aborted: true`.
+        const ac = new AbortController();
+        const invocation = shellTool.build({
+          command: 'sleep 1',
+          is_background: false,
+        });
+        const promise = invocation.execute(ac.signal, updateOutputMock);
+
+        // Leading-edge + suppressed (timer pending)
+        mockShellOutputCallback({ type: 'data', chunk: 'partial' });
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+        mockShellOutputCallback({ type: 'data', chunk: 'more partial' });
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+
+        // Abort. The timer must be cancelled synchronously.
+        ac.abort();
+
+        // Drain the would-be window. updateOutput must NOT be called.
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS * 2);
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+
+        // Settle the execution as aborted so the test cleanly exits.
+        resolveExecutionPromise({
+          rawOutput: Buffer.from('partial'),
+          output: 'partial',
+          exitCode: null,
+          signal: 15,
+          error: null,
+          aborted: true,
+          pid: 12345,
+          executionMethod: 'child_process',
+        });
+        await promise;
+
+        // Even after settle + further time, no late update.
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS * 2);
+        expect(updateOutputMock).toHaveBeenCalledOnce();
+      });
+
+      it('should clean up a pending trailing flush if execute() rejects', async () => {
+        // ShellExecutionService.execute() can throw before resolving
+        // (e.g. PTY dynamic import failure). The tool must propagate the
+        // error AND ensure no scheduled timer survives to fire a late
+        // updateOutput call after the caller has already seen the error.
+        // (No chunks can arrive before execute() resolves, so the timer
+        // is never actually scheduled in this path. The contract we
+        // verify here is that the abort listener is torn down — which we
+        // observe indirectly via "no late update on subsequent abort".)
+        const ac = new AbortController();
+        mockShellExecutionService.mockImplementationOnce(() => {
+          throw new Error('pty-import-failed');
+        });
+
+        const invocation = shellTool.build({
+          command: 'pty-cmd',
+          is_background: false,
+        });
+
+        await expect(
+          invocation.execute(ac.signal, updateOutputMock),
+        ).rejects.toThrow('pty-import-failed');
+
+        // After rejection, aborting must not crash and must not produce
+        // any updateOutput calls (no listener leak).
+        ac.abort();
+        await vi.advanceTimersByTimeAsync(OUTPUT_UPDATE_INTERVAL_MS * 2);
+        expect(updateOutputMock).not.toHaveBeenCalled();
+      });
+
+      it('should pass ANSI chunks through immediately without throttling', async () => {
+        const invocation = shellTool.build({
+          command: 'interactive-cmd',
+          is_background: false,
+        });
+        const promise = invocation.execute(mockAbortSignal, updateOutputMock);
+
+        const ansiChunk1: import('../utils/terminalSerializer.js').AnsiOutput =
+          [
+            [
+              {
+                text: 'Hello',
+                bold: false,
+                italic: false,
+                dim: false,
+                underline: false,
+                inverse: false,
+                fg: '',
+                bg: '',
+              },
+            ],
+          ];
+        const ansiChunk2: import('../utils/terminalSerializer.js').AnsiOutput =
+          [
+            [
+              {
+                text: 'World',
+                bold: false,
+                italic: false,
+                dim: false,
+                underline: false,
+                inverse: false,
+                fg: '',
+                bg: '',
+              },
+            ],
+          ];
+
+        // Both ANSI chunks should fire updateOutput immediately, back-to-back
+        mockShellOutputCallback({ type: 'data', chunk: ansiChunk1 });
+        mockShellOutputCallback({ type: 'data', chunk: ansiChunk2 });
+
+        expect(updateOutputMock).toHaveBeenCalledTimes(2);
+
+        resolveExecutionPromise({
+          rawOutput: Buffer.from(''),
+          output: '',
+          exitCode: 0,
+          signal: null,
+          error: null,
+          aborted: false,
+          pid: 12345,
+          executionMethod: 'child_process',
+        });
+        await promise;
+      });
     });
 
     describe('long-running foreground hint', () => {