fix(remote-terminal): stop the stream watchdog from tearing down a terminal whose output is still flowing (#33971)

* fix(remote-terminal): stop the stream watchdog from tearing down a terminal whose output is still flowing

The delivery deadline anchored at the first in-flight byte and only cleared when no bytes were in flight. Under continuous output that never happens, so a healthy stream was torn down and resubscribed about every 30s (152 recoveries in one crash bundle). Anchor the deadline to the oldest unsettled or unacknowledged chunk instead.

* fix(remote-terminal): ignore deliveries after the stream watchdog is disposed

* perf(remote-terminal): drain watchdog ACK queue in O(1) and arm the deadline once

Replace per-chunk Array.shift() with a head index plus amortized compaction so
one ACK covering many tiny frames no longer blocks the renderer (200k one-byte
chunks: ~17s before, a few ms now). Arm the delivery timer once and recompute
the oldest-pending anchor when it fires instead of re-arming per frame.
Add tests for ACK stall under continuous settled output and the drain cost.

---------

Co-authored-by: m4air <m4air@Mac.localdomain>
Co-authored-by: Neil <neil@stably.ai>

src/renderer/src/runtime/remote-terminal-stream-watchdog.test.ts

@@ -0,0 +1,127 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS,
+  createRemoteTerminalStreamWatchdog
+} from './remote-terminal-stream-watchdog'
+
+describe('remote terminal stream watchdog delivery deadline', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('anchors the deadline to the oldest unsettled delivery instead of the last settled sibling', () => {
+    const onStall = vi.fn()
+    const watchdog = createRemoteTerminalStreamWatchdog(onStall)
+
+    watchdog.beginOutputDelivery(100)
+    for (let tick = 0; tick < 3; tick += 1) {
+      vi.advanceTimersByTime(9_000)
+      const settle = watchdog.beginOutputDelivery(10)
+      settle()
+    }
+    expect(onStall).not.toHaveBeenCalled()
+
+    vi.advanceTimersByTime(REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS - 27_000)
+
+    expect(onStall).toHaveBeenCalledTimes(1)
+    expect(onStall.mock.calls[0]?.[0]).toMatchObject({ reason: 'delivery-credit-timeout' })
+  })
+
+  it('stays quiet while continuous output keeps settling and acknowledging', () => {
+    const onStall = vi.fn()
+    const watchdog = createRemoteTerminalStreamWatchdog(onStall)
+
+    // Each chunk parses 50ms and is acked 100ms after it lands, so bytes are always in flight.
+    for (
+      let elapsedMs = 0;
+      elapsedMs < REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS * 4;
+      elapsedMs += 20
+    ) {
+      const settle = watchdog.beginOutputDelivery(1_000)
+      setTimeout(settle, 50)
+      setTimeout(() => watchdog.recordOutputAcknowledged(1_000), 100)
+      vi.advanceTimersByTime(20)
+    }
+
+    expect(onStall).not.toHaveBeenCalled()
+  })
+
+  it('trips 30s after the oldest unacknowledged chunk once ACKs stop, even while output keeps settling', () => {
+    const onStall = vi.fn()
+    const watchdog = createRemoteTerminalStreamWatchdog(onStall)
+
+    // Healthy start: deliveries settle and are ACKed.
+    for (let tick = 0; tick < 50; tick += 1) {
+      watchdog.beginOutputDelivery(1_000)()
+      watchdog.recordOutputAcknowledged(1_000)
+      vi.advanceTimersByTime(20)
+    }
+    const ackStoppedAtMs = Date.now()
+    // ACKs stop (only part of the oldest chunk is ever acknowledged) while output keeps settling.
+    let partialAckSent = false
+    let stalledAfterMs: number | null = null
+    for (let tick = 0; tick < 2_000 && stalledAfterMs === null; tick += 1) {
+      watchdog.beginOutputDelivery(1_000)()
+      if (!partialAckSent) {
+        watchdog.recordOutputAcknowledged(500)
+        partialAckSent = true
+      }
+      vi.advanceTimersByTime(20)
+      if (onStall.mock.calls.length > 0) {
+        stalledAfterMs = Date.now() - ackStoppedAtMs
+      }
+    }
+
+    expect(onStall).toHaveBeenCalledTimes(1)
+    expect(onStall.mock.calls[0]?.[0]).toMatchObject({ reason: 'delivery-credit-timeout' })
+    expect(stalledAfterMs).toBeGreaterThanOrEqual(REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS)
+    expect(stalledAfterMs).toBeLessThanOrEqual(REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS + 20)
+  })
+
+  it('drains a single ACK covering many tiny chunks without quadratic cost', () => {
+    vi.useRealTimers()
+    const onStall = vi.fn()
+    const watchdog = createRemoteTerminalStreamWatchdog(onStall)
+    const chunkCount = 200_000
+    for (let index = 0; index < chunkCount; index += 1) {
+      watchdog.beginOutputDelivery(1)()
+    }
+
+    const startedAt = performance.now()
+    watchdog.recordOutputAcknowledged(chunkCount)
+    const elapsedMs = performance.now() - startedAt
+    watchdog.dispose()
+
+    // Repeated Array.shift() takes seconds here; a head index takes a few ms.
+    expect(elapsedMs).toBeLessThan(100)
+  })
+
+  it('stays armed while parsed bytes remain unacknowledged to the host', () => {
+    const onStall = vi.fn()
+    const watchdog = createRemoteTerminalStreamWatchdog(onStall)
+
+    watchdog.beginOutputDelivery(100)()
+    vi.advanceTimersByTime(REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS)
+
+    expect(onStall).toHaveBeenCalledTimes(1)
+    expect(onStall.mock.calls[0]?.[0]).toMatchObject({
+      outstandingDeliveryBytes: 0,
+      reason: 'delivery-credit-timeout'
+    })
+  })
+
+  it('disarms once the acknowledgement reaches the transport', () => {
+    const onStall = vi.fn()
+    const watchdog = createRemoteTerminalStreamWatchdog(onStall)
+
+    watchdog.beginOutputDelivery(100)()
+    watchdog.recordOutputAcknowledged(100)
+    vi.advanceTimersByTime(REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS * 2)
+
+    expect(onStall).not.toHaveBeenCalled()
+  })
+})

src/renderer/src/runtime/remote-terminal-stream-watchdog.ts

@@ -24,9 +24,12 @@ export function createRemoteTerminalStreamWatchdog(
   let responseTimer: ReturnType<typeof setTimeout> | null = null
   let deliveryTimer: ReturnType<typeof setTimeout> | null = null
   let outstandingDeliveryBytes = 0
+  // Unsettled deliveries in arrival order; the oldest one's start is the stall anchor.
+  const unsettledDeliveries = new Set<{ startedAtMs: number }>()
   // Host credit reopens on ACK frames; parsed-but-unacked bytes remain credit whose loss can stall output.
-  let unacknowledgedBytes = 0
-  let deliveryPendingSinceMs: number | null = null
+  // Why a head index: one ACK can cover thousands of tiny frames, and repeated shift() is O(n) each.
+  let unacknowledgedChunks: { bytes: number; startedAtMs: number }[] = []
+  let unacknowledgedHead = 0
   let lastInboundAtMs = Date.now()
   let commandResponseProbePending = false
   let disposed = false
@@ -60,29 +63,55 @@ export function createRemoteTerminalStreamWatchdog(
       reason
     })
   }
-  // Anchor deadline: sibling settles must not postpone it forever, and clearing at zero parse credit would require the very inbound output a stall blocks.
-  const syncDeliveryTimer = (): void => {
-    if (disposed || outstandingDeliveryBytes + unacknowledgedBytes <= 0) {
-      clearDeliveryTimer()
-      deliveryPendingSinceMs = null
-      return
-    }
-    deliveryPendingSinceMs ??= Date.now()
-    if (deliveryTimer) {
-      return
-    }
+  const oldestPendingAtMs = (): number => {
+    const oldestDelivery = unsettledDeliveries.values().next().value
+    return Math.min(
+      oldestDelivery?.startedAtMs ?? Infinity,
+      unacknowledgedChunks[unacknowledgedHead]?.startedAtMs ?? Infinity
+    )
+  }
+  // Why oldest pending: flowing output always has bytes in flight; only the oldest byte not moving is a stall.
+  // Why arm once and recheck on fire: the anchor only moves later, so re-arming per frame would churn timers on the output hot path.
+  const armDeliveryTimer = (anchorMs: number): void => {
     const remainingMs = Math.max(
       0,
-      REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS - (Date.now() - deliveryPendingSinceMs)
+      REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS - (Date.now() - anchorMs)
     )
-    deliveryTimer = setTimeout(() => trip('delivery-credit-timeout'), remainingMs)
+    deliveryTimer = setTimeout(() => {
+      deliveryTimer = null
+      const currentAnchorMs = oldestPendingAtMs()
+      if (disposed || currentAnchorMs === Infinity) {
+        return
+      }
+      if (Date.now() - currentAnchorMs >= REMOTE_TERMINAL_DELIVERY_STALL_TIMEOUT_MS) {
+        trip('delivery-credit-timeout')
+        return
+      }
+      armDeliveryTimer(currentAnchorMs)
+    }, remainingMs)
+  }
+  const ensureDeliveryTimer = (): void => {
+    if (disposed || deliveryTimer) {
+      return
+    }
+    const anchorMs = oldestPendingAtMs()
+    if (anchorMs !== Infinity) {
+      armDeliveryTimer(anchorMs)
+    }
   }
 
   return {
     beginOutputDelivery(bytes) {
+      if (disposed) {
+        return () => {}
+      }
+      const delivery = { startedAtMs: Date.now() }
       outstandingDeliveryBytes += bytes
-      unacknowledgedBytes += bytes
-      syncDeliveryTimer()
+      unsettledDeliveries.add(delivery)
+      if (bytes > 0) {
+        unacknowledgedChunks.push({ bytes, startedAtMs: delivery.startedAtMs })
+      }
+      ensureDeliveryTimer()
       let settled = false
       return () => {
         if (settled || disposed) {
@@ -90,15 +119,34 @@ export function createRemoteTerminalStreamWatchdog(
         }
         settled = true
         outstandingDeliveryBytes = Math.max(0, outstandingDeliveryBytes - bytes)
-        syncDeliveryTimer()
+        unsettledDeliveries.delete(delivery)
       }
     },
     recordOutputAcknowledged(bytes) {
       if (disposed) {
         return
       }
-      unacknowledgedBytes = Math.max(0, unacknowledgedBytes - bytes)
-      syncDeliveryTimer()
+      let remaining = bytes
+      while (remaining > 0 && unacknowledgedHead < unacknowledgedChunks.length) {
+        const oldest = unacknowledgedChunks[unacknowledgedHead]
+        const consumed = Math.min(oldest.bytes, remaining)
+        oldest.bytes -= consumed
+        remaining -= consumed
+        if (oldest.bytes === 0) {
+          unacknowledgedHead += 1
+        }
+      }
+      // Why compact only past half: keeps slice cost amortized O(1) per consumed chunk.
+      if (unacknowledgedHead === unacknowledgedChunks.length) {
+        unacknowledgedChunks = []
+        unacknowledgedHead = 0
+      } else if (
+        unacknowledgedHead > 1024 &&
+        unacknowledgedHead * 2 > unacknowledgedChunks.length
+      ) {
+        unacknowledgedChunks = unacknowledgedChunks.slice(unacknowledgedHead)
+        unacknowledgedHead = 0
+      }
     },
     completeCommandResponseProbe() {
       commandResponseProbePending = false
@@ -124,8 +172,9 @@ export function createRemoteTerminalStreamWatchdog(
       clearResponseTimer()
       clearDeliveryTimer()
       outstandingDeliveryBytes = 0
-      unacknowledgedBytes = 0
-      deliveryPendingSinceMs = null
+      unsettledDeliveries.clear()
+      unacknowledgedChunks = []
+      unacknowledgedHead = 0
     }
   }
 }