fix(terminal): stop stale multiplex stream handles from swallowing input (#16325)

* test(terminal): pin park/reveal re-subscribe on a shared multiplexer

Investigating STA-5098. Parking a mirrored remote tab closes its stream
while a sibling tab keeps the multiplexer alive, so the reveal
re-subscribes on an instance that already retired a stream.

This came back green, which exonerates the multiplexer as the cause of
STA-5098 — the wedge is above it. Kept as a contract guard; the header
says explicitly that it is not coverage for that ticket.

* fix(terminal): stop stale multiplex stream handles from swallowing input

A stream handle whose record was dropped (park close, or a reconnect that
clears the stream table) kept reporting success: sendFrame gates only on
socket readiness, never on stream membership. The host drops those frames
for an unknown stream id, so a revealed cold-parked remote pane looked
connected while the PTY never saw a byte and never painted (STA-5098).

Reporting success also defeated the transport's own recovery — it re-sends
input over terminal.send when the stream refuses it, which never ran.

Guard the three public senders on stream membership, the check close() and
setOutputPaused already use, and drain input queued behind a viewport claim
on every stream install rather than only a still-pending claim.

Withdraws the parked-reveal re-subscribe test: its fake host answered
Subscribe with an immediate snapshot, so it could not fail.

* chore(terminal): tighten the stale-stream comments

src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts

@@ -377,6 +377,20 @@ export function createRemoteRuntimePtyTransport(
       pendingClaimQueryReplyCount += 1
     }
   }
+  // Why: clearing the claim flag without draining strands the queued bytes.
+  const flushPendingClaimInput = (stream: RemoteRuntimeMultiplexedTerminal): void => {
+    const queued = pendingClaimInput
+    pendingViewportClaim = false
+    pendingClaimInput = []
+    pendingClaimQueryReplyCount = 0
+    for (const segment of queued) {
+      stream.sendInput(segment.text)
+    }
+    for (const resolve of viewportClaimReadyWaiters) {
+      resolve(true)
+    }
+    viewportClaimReadyWaiters.clear()
+  }
   // Why: tab/leaf ids are shared by paired viewers; the instance suffix keeps one viewer's refresh off peer records.
   const clientId = `desktop:${tabId ?? 'tab'}:${leafId ?? 'leaf'}:${createBrowserUuid()}`
   const terminalCreateMutationId = createBrowserUuid()
@@ -1338,8 +1352,8 @@ export function createRemoteRuntimePtyTransport(
     }
     const stream = getCurrentMultiplexedStream(targetHandle)
     if (claim ? stream?.claimViewport(cols, rows) : stream?.resize(cols, rows)) {
-      if (claim) {
-        pendingViewportClaim = false
+      if (claim && stream) {
+        flushPendingClaimInput(stream)
       }
       return
     }
@@ -1953,17 +1967,6 @@ export function createRemoteRuntimePtyTransport(
     // Why: a viewport change during the subscribe round-trip hit the no-op one-shot fallback; replay the latest viewport so the PTY isn't stuck at subscribe-time size.
     if (pendingViewportClaim && desiredViewport) {
       nextStream.claimViewport(desiredViewport.cols, desiredViewport.rows)
-      pendingViewportClaim = false
-      const queuedInput = pendingClaimInput
-      pendingClaimInput = []
-      pendingClaimQueryReplyCount = 0
-      for (const segment of queuedInput) {
-        nextStream.sendInput(segment.text)
-      }
-      for (const resolve of viewportClaimReadyWaiters) {
-        resolve(true)
-      }
-      viewportClaimReadyWaiters.clear()
     } else if (
       desiredViewport &&
       (desiredViewport.cols !== subscribedViewport?.cols ||
@@ -1971,6 +1974,8 @@ export function createRemoteRuntimePtyTransport(
     ) {
       nextStream.resize(desiredViewport.cols, desiredViewport.rows)
     }
+    // Why: a live claim may already have cleared the flag, so drain on every install.
+    flushPendingClaimInput(nextStream)
   }
 
   const transport: PtyTransport = {

src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts

@@ -464,14 +464,18 @@ class RemoteRuntimeTerminalMultiplexer {
 
     const stream: RemoteRuntimeMultiplexedTerminal = {
       streamId,
-      sendInput: (text) => this.sendInput(state, text),
+      sendInput: (text) => this.isRegisteredStream(state) && this.sendInput(state, text),
       resize: (cols, rows) =>
+        this.isRegisteredStream(state) &&
         this.sendFrame(
           streamId,
           TerminalStreamOpcode.Resize,
           encodeTerminalStreamJson({ cols, rows })
         ),
       claimViewport: (cols, rows) => {
+        if (!this.isRegisteredStream(state)) {
+          return false
+        }
         const claimed = this.sendFrame(
           streamId,
           TerminalStreamOpcode.ClaimViewport,
@@ -1181,6 +1185,11 @@ class RemoteRuntimeTerminalMultiplexer {
     )
   }
 
+  // Why: sendFrame gates on readiness alone; a dropped handle would still report success.
+  private isRegisteredStream(stream: RemoteRuntimeMultiplexedTerminalState): boolean {
+    return this.streams.get(stream.streamId) === stream
+  }
+
   private sendInput(stream: RemoteRuntimeMultiplexedTerminalState, text: string): boolean {
     const sent = this.sendFrame(
       stream.streamId,

src/renderer/src/runtime/remote-runtime-terminal-stale-stream-frames.test.ts

@@ -0,0 +1,87 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  TerminalStreamOpcode,
+  decodeTerminalStreamFrame,
+  decodeTerminalStreamText
+} from '../../../shared/terminal-stream-protocol'
+import {
+  getRemoteRuntimeTerminalMultiplexer,
+  resetRemoteRuntimeTerminalMultiplexersForTests,
+  type RemoteRuntimeMultiplexedTerminal
+} from './remote-runtime-terminal-multiplexer'
+import { replaceRuntimeEnvironmentRevisions } from './runtime-environment-revision'
+
+// Why: sendFrame gates on socket readiness alone, so a dropped stream handle reported success
+// while the host discarded the frames for an unknown stream id.
+
+type SubscribeCallbacks = {
+  onResponse: (response: unknown) => void
+  onBinary?: (bytes: Uint8Array<ArrayBufferLike>) => void
+  onError?: (error: { message: string }) => void
+  onClose?: () => void
+}
+
+describe('remote terminal stale stream frames', () => {
+  let sent: Uint8Array<ArrayBufferLike>[]
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetRemoteRuntimeTerminalMultiplexersForTests()
+    replaceRuntimeEnvironmentRevisions([])
+    sent = []
+
+    const subscribe = vi.fn(async (_args: unknown, callbacks: SubscribeCallbacks) => {
+      queueMicrotask(() => callbacks.onResponse({ ok: true, result: { type: 'ready' } }))
+      return {
+        unsubscribe: vi.fn(),
+        sendBinary: (bytes: Uint8Array<ArrayBufferLike>) => {
+          sent.push(bytes)
+        }
+      }
+    })
+    vi.stubGlobal('window', { api: { runtimeEnvironments: { subscribe } } })
+  })
+
+  afterEach(() => {
+    vi.unstubAllGlobals()
+  })
+
+  async function subscribeStream(terminal: string): Promise<RemoteRuntimeMultiplexedTerminal> {
+    const stream = await getRemoteRuntimeTerminalMultiplexer('env-1').subscribeTerminal({
+      terminal,
+      client: { id: 'desktop-1', type: 'desktop' },
+      callbacks: { onData: () => {}, onSnapshot: () => {} }
+    })
+    await Promise.resolve()
+    return stream
+  }
+
+  function inputTextOnWire(): string {
+    return sent
+      .map((bytes) => decodeTerminalStreamFrame(bytes))
+      .filter((frame) => frame?.opcode === TerminalStreamOpcode.Input)
+      .map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : ''))
+      .join('')
+  }
+
+  function opcodeCount(opcode: TerminalStreamOpcode): number {
+    return sent.filter((bytes) => decodeTerminalStreamFrame(bytes)?.opcode === opcode).length
+  }
+
+  it('refuses input and viewport frames from a closed stream while a sibling keeps the socket live', async () => {
+    const parked = await subscribeStream('terminal-1')
+    // A sibling stream keeps the multiplexer connected, so `ready` stays true after the close.
+    await subscribeStream('terminal-2')
+
+    parked.close()
+    sent = []
+
+    expect(parked.sendInput('never-delivered\r')).toBe(false)
+    expect(parked.claimViewport(80, 24)).toBe(false)
+    expect(parked.resize(80, 24)).toBe(false)
+
+    expect(inputTextOnWire()).toBe('')
+    expect(opcodeCount(TerminalStreamOpcode.ClaimViewport)).toBe(0)
+    expect(opcodeCount(TerminalStreamOpcode.Resize)).toBe(0)
+  })
+})