fix(ssh): bound PTY model admission

src/main/ipc/pty-pending-data-drain-contract.ts

@@ -9,6 +9,7 @@ export type PendingPtyData = {
   droppedOutput?: true
   droppedMode2031Data?: string
   droppedMode2031ScanState?: Mode2031ReplyScanState
+  projectionAdmissionIds?: readonly string[]
 }
 
 export type PtyPendingDataDrainDisposition = 'active' | 'background' | 'blocked'

src/main/ipc/pty.ts

@@ -158,6 +158,9 @@ import {
   unmarkHiddenRendererPty
 } from './pty-hidden-delivery-gate'
 import { PtyPendingDataDrainQueue, type PendingPtyData } from './pty-pending-data-drain-queue'
+import { SshPtyOutputIntake } from './ssh-pty-output-intake'
+import { installSshPtyOutputIntake } from './ssh-pty-output-intake-registry'
+import type { LegacySshProjectionSemantics } from './ssh-pty-legacy-projection'
 import {
   clearNativeWindowsConptyPty,
   isNativeWindowsLocalPtySpawn,
@@ -1467,6 +1470,7 @@ let rendererDidStartLoadingHandler: (() => void) | null = null
 
 // Why: Restart daemon must re-bind provider→renderer listeners after replaceDaemonProvider swaps localProvider, else subscribers stay bound to the disposed adapter and new PTY data silently drops.
 let rebindProviderListeners: (() => void) | null = null
+let sshOutputIntakeCleanup: (() => void) | null = null
 
 export function rebindLocalProviderListeners(): void {
   rebindProviderListeners?.()
@@ -1835,6 +1839,7 @@ export function registerPtyHandlers(
     },
     () => isHiddenPtyDeliveryGateEnabled(getSettings?.())
   )
+  let sshOutputIntake: SshPtyOutputIntake | null = null
   // Why: resuming a paused producer during exit can synchronously emit; those bytes must not queue behind pty:exit.
   const rendererExitingPtyIds = new Set<string>()
   const rendererDeliveryRestoreNeededPtys = new Set<string>()
@@ -1995,6 +2000,14 @@ export function registerPtyHandlers(
   }
 
   function clearPendingPtyData(): void {
+    for (const pending of pendingData.values()) {
+      if (pending.projectionAdmissionIds) {
+        sshOutputIntake?.transferProjections(
+          pending.projectionAdmissionIds,
+          'renderer-lifecycle-reset'
+        )
+      }
+    }
     pendingData.clear()
   }
 
@@ -2260,6 +2273,9 @@ export function registerPtyHandlers(
       accounting.lastAckAtMs = Date.now()
     }
     rendererInFlightTotalChars = Math.max(0, rendererInFlightTotalChars - acknowledged)
+    if (acknowledged > 0) {
+      sshOutputIntake?.settleProjectionPrefix(id, acknowledged)
+    }
     return acknowledged
   }
 
@@ -2331,6 +2347,12 @@ export function registerPtyHandlers(
       // Why drop pending: everything at/before markerSeq comes from the snapshot, so flushing pre-marker bytes would double-paint the restore.
       const pending = pendingData.get(id)
       if (pending) {
+        if (pending.projectionAdmissionIds) {
+          sshOutputIntake?.transferProjections(
+            pending.projectionAdmissionIds,
+            'renderer-delivery-writeoff'
+          )
+        }
         pendingDroppedChars += pending.data.length
         deletePendingPtyData(id)
         pendingOverflowMarkedPtys.delete(id)
@@ -2360,7 +2382,11 @@ export function registerPtyHandlers(
     return writtenOff
   }
 
-  function sendPtyDataToRenderer(id: string, payload: PtyDataPayload): boolean {
+  function sendPtyDataToRenderer(
+    id: string,
+    payload: PtyDataPayload,
+    projectionAdmissionIds?: readonly string[]
+  ): boolean {
     const charCount = getPtyPayloadCharCount(payload)
     const accounting = rendererDeliveryAccountingByPty.get(id)
     const hadAccounting = accounting !== undefined
@@ -2395,6 +2421,9 @@ export function registerPtyHandlers(
         }
       }
       rendererDeliveryRestoreNeededPtys.add(id)
+      if (projectionAdmissionIds) {
+        sshOutputIntake?.transferProjections(projectionAdmissionIds, 'renderer-send-failed')
+      }
       mainDeliveryBreadcrumbs.record('pty-data-send-failed', {
         id: redactPtyIdForDiagnostics(id),
         chars: charCount
@@ -2402,6 +2431,17 @@ export function registerPtyHandlers(
       console.error('[pty] renderer data send failed; payload will not be retried', error)
       return false
     }
+    if (projectionAdmissionIds) {
+      try {
+        sshOutputIntake?.publishProjectionPrefix(
+          projectionAdmissionIds,
+          payload.data.length,
+          charCount
+        )
+      } catch {
+        sshOutputIntake?.transferProjections(projectionAdmissionIds, 'projection-publish-failed')
+      }
+    }
     if (rendererDeliveryRestoreNeededPtys.has(id)) {
       try {
         sendModelRestoreNeededMarker(id, 'delivery-heal', runtime?.getPtyOutputSequence(id))
@@ -2516,6 +2556,9 @@ export function registerPtyHandlers(
       pendingOverflowMarkedPtys.add(id)
     }
     pendingDroppedChars += pending.data.length
+    if (pending.projectionAdmissionIds) {
+      sshOutputIntake?.transferProjections(pending.projectionAdmissionIds, 'pending-cap')
+    }
     const mode2031 = scanDroppedMode2031Data(pending.data, INITIAL_MODE_2031_REPLY_SCAN_STATE)
     // Why no trimmed content tail: a mid-stream gap would corrupt the pane; the droppedOutput sentinel repaints from the snapshot and realigns by sequence (only query bytes ride along).
     return {
@@ -2534,10 +2577,14 @@ export function registerPtyHandlers(
     preservesSeq: boolean,
     containsBackgroundOutput: boolean,
     rawLength = data.length,
-    transformed = false
+    transformed = false,
+    projectionSemanticsId?: string
   ): PendingPtyData {
     // Why stay dropped at O(1): once over the cap the restore sentinel supersedes interim bytes; queries still get carved out (bounded) so replies survive the whole episode.
     if (existing?.droppedOutput === true) {
+      if (projectionSemanticsId) {
+        sshOutputIntake?.transferProjections([projectionSemanticsId], 'pending-cap')
+      }
       const mode2031 = scanDroppedMode2031Data(
         data,
         existing.droppedMode2031ScanState ?? INITIAL_MODE_2031_REPLY_SCAN_STATE
@@ -2562,7 +2609,8 @@ export function registerPtyHandlers(
         ...(typeof startSeq === 'number' ? { startSeq } : {}),
         ...(rawLength !== data.length ? { rawLength } : {}),
         ...(transformed ? { transformed: true } : {}),
-        ...(nextContainsBackgroundOutput ? { containsBackgroundOutput: true } : {})
+        ...(nextContainsBackgroundOutput ? { containsBackgroundOutput: true } : {}),
+        ...(projectionSemanticsId ? { projectionAdmissionIds: [projectionSemanticsId] } : {})
       })
     }
     const existingRawLength = existing.rawLength ?? existing.data.length
@@ -2571,7 +2619,15 @@ export function registerPtyHandlers(
       ...(!preservesSeq || existing.transformed || transformed
         ? { rawLength: existingRawLength + rawLength, transformed: true as const }
         : {}),
-      ...(nextContainsBackgroundOutput ? { containsBackgroundOutput: true } : {})
+      ...(nextContainsBackgroundOutput ? { containsBackgroundOutput: true } : {}),
+      ...((existing.projectionAdmissionIds?.length ?? 0) > 0 || projectionSemanticsId
+        ? {
+            projectionAdmissionIds: [
+              ...(existing.projectionAdmissionIds ?? []),
+              ...(projectionSemanticsId ? [projectionSemanticsId] : [])
+            ]
+          }
+        : {})
     }
     if (typeof existing.startSeq === 'number') {
       next.startSeq = existing.startSeq
@@ -2656,6 +2712,9 @@ export function registerPtyHandlers(
           pendingOverflowMarkedPtys.delete(id)
           updateProducerFlowControl(id)
           const drop = recordHiddenRendererPtyDataDrop(id, pending.data.length)
+          if (pending.projectionAdmissionIds) {
+            sshOutputIntake?.transferProjections(pending.projectionAdmissionIds, 'hidden-drop')
+          }
           warnIfDroppingHiddenBytesForVisiblePty(id, pending.data.length)
           if (drop.shouldEmitRestoreMarker) {
             sendModelRestoreNeededMarker(id, 'hidden-drop', runtime?.getPtyOutputSequence(id))
@@ -2671,11 +2730,15 @@ export function registerPtyHandlers(
           updateProducerFlowControl(id)
           // Why droppedOutput sentinel: pending-cap drop means the pane must repaint from the snapshot, not continue a gapped stream (data = carved query bytes only).
           if (
-            !sendPtyDataToRenderer(id, {
+            !sendPtyDataToRenderer(
               id,
-              data: pending.data + getDroppedMode2031RendererData(pending),
-              droppedOutput: true
-            })
+              {
+                id,
+                data: pending.data + getDroppedMode2031RendererData(pending),
+                droppedOutput: true
+              },
+              pending.projectionAdmissionIds
+            )
           ) {
             sendFailed = true
             break
@@ -2695,6 +2758,9 @@ export function registerPtyHandlers(
           if (pending.containsBackgroundOutput === true) {
             nextPending.containsBackgroundOutput = true
           }
+          if (pending.projectionAdmissionIds) {
+            nextPending.projectionAdmissionIds = pending.projectionAdmissionIds
+          }
           pendingData.replaceWithRemainder(selection, nextPending)
         } else {
           pendingData.remove(selection)
@@ -2711,7 +2777,8 @@ export function registerPtyHandlers(
               pending.containsBackgroundOutput,
               pending.rawLength,
               pending.transformed
-            )
+            ),
+            pending.projectionAdmissionIds
           )
         ) {
           sendFailed = true
@@ -2792,11 +2859,15 @@ export function registerPtyHandlers(
       if (remaining) {
         if (remaining.droppedOutput === true) {
           // Sentinel entry: only salvaged query bytes remain; keep the flag so the renderer knows the span was dropped.
-          sendPtyDataToRenderer(payload.id, {
-            id: payload.id,
-            data: remaining.data,
-            droppedOutput: true
-          })
+          sendPtyDataToRenderer(
+            payload.id,
+            {
+              id: payload.id,
+              data: remaining.data,
+              droppedOutput: true
+            },
+            remaining.projectionAdmissionIds
+          )
         } else {
           sendPtyDataToRenderer(
             payload.id,
@@ -2807,10 +2878,12 @@ export function registerPtyHandlers(
               remaining.containsBackgroundOutput,
               remaining.rawLength,
               remaining.transformed
-            )
+            ),
+            remaining.projectionAdmissionIds
           )
         }
       }
+      sshOutputIntake?.transferPtyProjections(payload.id, 'pty-exit')
       // Why resume a dead PTY (no-op): avoid leaving a stale paused mark behind for a reused id.
       producerFlowControl.release(payload.id)
       pendingOverflowMarkedPtys.delete(payload.id)
@@ -2845,6 +2918,212 @@ export function registerPtyHandlers(
     }
   }
 
+  function acceptPtyDataForRenderer(
+    payload: {
+      id: string
+      data: string
+      sequenceChars?: number
+      transformed?: boolean
+    },
+    outputSeq: number | undefined,
+    projection?: LegacySshProjectionSemantics
+  ): void {
+    const rawLength = payload.sequenceChars ?? payload.data.length
+    const preservesSeq = !payload.transformed && rawLength === payload.data.length
+    const startSeq = typeof outputSeq === 'number' ? Math.max(0, outputSeq - rawLength) : undefined
+    const projectionId = projection?.identity.projectionSemanticsId
+    if (mainWindow.isDestroyed()) {
+      if (projectionId) {
+        sshOutputIntake?.transferProjections([projectionId], 'renderer-destroyed')
+      }
+      if (flushTimer) {
+        clearTimeout(flushTimer)
+        flushTimer = null
+      }
+      producerFlowControl.releaseAll()
+      clearDeliveryResyncProbe()
+      clearPendingPtyData()
+      pendingOverflowMarkedPtys.clear()
+      rendererDeliveryAccountingByPty.clear()
+      rendererInFlightTotalChars = 0
+      clearDispatcherReadyWatchdog()
+      return
+    }
+    if (rendererExitingPtyIds.has(payload.id)) {
+      if (projectionId) {
+        sshOutputIntake?.transferProjections([projectionId], 'pty-exiting')
+      }
+      return
+    }
+    if (shouldDropHiddenRendererPtyData(payload.id, getSettings?.())) {
+      if (projectionId) {
+        sshOutputIntake?.transferProjections([projectionId], 'hidden-drop')
+      }
+      const droppedChars = projection ? rawLength : payload.data.length
+      const drop = recordHiddenRendererPtyDataDrop(payload.id, droppedChars)
+      warnIfDroppingHiddenBytesForVisiblePty(payload.id, droppedChars)
+      if (drop.shouldEmitRestoreMarker) {
+        sendModelRestoreNeededMarker(payload.id, 'hidden-drop', outputSeq)
+      }
+      return
+    }
+    if (payload.data.length === 0 && !payload.transformed) {
+      if (projectionId) {
+        sshOutputIntake?.transferProjections([projectionId], 'empty-projection')
+      }
+      return
+    }
+    const containsBackgroundOutput =
+      rendererPtyIsKnownHidden(payload.id) || ptyHasHiddenRendererResizeOutput(payload.id)
+    if (containsBackgroundOutput) {
+      markHiddenRendererResizeOutputDelivered(payload.id)
+    }
+    const overflowMarkedBeforeAppend = pendingOverflowMarkedPtys.has(payload.id)
+    const pending = appendPendingPtyData(
+      payload.id,
+      pendingData.get(payload.id),
+      payload.data,
+      startSeq,
+      preservesSeq,
+      containsBackgroundOutput,
+      rawLength,
+      payload.transformed === true,
+      projectionId
+    )
+    const shouldEmitPendingCapRestoreMarker =
+      pending.droppedOutput === true &&
+      !overflowMarkedBeforeAppend &&
+      pendingOverflowMarkedPtys.has(payload.id)
+    const nextData = pending.data + getDroppedMode2031RendererData(pending)
+    const isInteractiveOutput = shouldSendInteractiveOutputNow(
+      payload.id,
+      nextData,
+      performance.now()
+    )
+    if (isInteractiveOutput && rendererPtyDispatcherReady) {
+      if (!canSendPtyDataToRenderer(payload.id, { interactive: true })) {
+        setPendingPtyData(payload.id, pending)
+        if (shouldEmitPendingCapRestoreMarker) {
+          sendModelRestoreNeededMarker(payload.id, 'pending-cap', outputSeq)
+        }
+        updateProducerFlowControl(payload.id)
+        requestDeliveryResyncForGatedPty()
+        return
+      }
+      deletePendingPtyData(payload.id)
+      clearFlushTimerIfIdle()
+      if (shouldEmitPendingCapRestoreMarker) {
+        sendModelRestoreNeededMarker(payload.id, 'pending-cap', outputSeq)
+      }
+      pendingOverflowMarkedPtys.delete(payload.id)
+      try {
+        sendPtyDataToRenderer(
+          payload.id,
+          {
+            id: payload.id,
+            data: nextData,
+            ...(typeof pending.startSeq === 'number'
+              ? {
+                  seq: pending.startSeq + (pending.rawLength ?? nextData.length),
+                  rawLength: pending.rawLength ?? nextData.length
+                }
+              : {}),
+            ...(pending.transformed ? { transformed: true } : {}),
+            ...(pending.containsBackgroundOutput === true ? { background: true } : {}),
+            ...(pending.droppedOutput === true ? { droppedOutput: true } : {})
+          },
+          pending.projectionAdmissionIds
+        )
+      } finally {
+        updateProducerFlowControl(payload.id)
+      }
+      return
+    }
+    setPendingPtyData(payload.id, pending)
+    if (shouldEmitPendingCapRestoreMarker) {
+      sendModelRestoreNeededMarker(payload.id, 'pending-cap', outputSeq)
+    }
+    updateProducerFlowControl(payload.id)
+    if (
+      !canSendPtyDataToRenderer(payload.id, { interactive: activeRendererPtys.has(payload.id) })
+    ) {
+      requestDeliveryResyncForGatedPty()
+    }
+    if (!flushTimer) {
+      schedulePendingDataFlush(PTY_BATCH_INTERVAL_MS)
+    }
+  }
+
+  sshOutputIntakeCleanup?.()
+  sshOutputIntake = new SshPtyOutputIntake({
+    getModelSequence: (id) => runtime?.getPtyOutputSequence(id) ?? 0,
+    acceptModel: (event) => {
+      if (!runtime) {
+        throw new Error('SSH PTY output requires the main terminal model')
+      }
+      return runtime.acceptPtyDataBounded(
+        event.id,
+        event.data,
+        Date.now(),
+        event.rawLength,
+        event.transformed
+      )
+    },
+    project: (event, projection) =>
+      acceptPtyDataForRenderer(
+        {
+          id: event.id,
+          data: event.data,
+          sequenceChars: event.rawLength,
+          transformed: event.transformed
+        },
+        projection.identity.sequenceEnd,
+        projection
+      ),
+    finalizeExit: (event) => {
+      runtime?.onPtyExit(event.id, event.code, event.ptyIncarnation)
+      sendPtyExitToRenderer(event)
+    },
+    pauseProvider: (_generation, id) => {
+      const provider = tryGetProviderForPty(id) as
+        | (IPtyProvider & { hasPtyDeliveryPauseAdapter?: () => boolean })
+        | undefined
+      if (!provider?.hasPtyDeliveryPauseAdapter?.()) {
+        return false
+      }
+      provider.pauseProducer?.(id)
+      return true
+    },
+    resumeProvider: (_generation, id) => tryGetProviderForPty(id)?.resumeProducer?.(id),
+    closeProvider: (generation, reason) => {
+      for (const provider of sshProviders.values()) {
+        if ((provider as { providerGeneration?: number }).providerGeneration === generation) {
+          ;(
+            provider as IPtyProvider & { closeOutputIntake?: (reason: string) => void }
+          ).closeOutputIntake?.(reason)
+          return
+        }
+      }
+    },
+    onGenerationClosed: (providerGeneration) => {
+      for (const id of pendingData.keys()) {
+        const pending = pendingData.get(id)
+        if (
+          pending?.projectionAdmissionIds &&
+          sshOutputIntake?.hasProjectionFromGeneration(
+            pending.projectionAdmissionIds,
+            providerGeneration
+          )
+        ) {
+          pendingData.delete(id)
+          updateProducerFlowControl(id)
+          pendingOverflowMarkedPtys.delete(id)
+        }
+      }
+    }
+  })
+  sshOutputIntakeCleanup = installSshPtyOutputIntake(sshOutputIntake)
+
   async function shutdownProviderAndDetectExit(
     provider: IPtyProvider,
     id: string,
@@ -2928,120 +3207,7 @@ export function registerPtyHandlers(
       const outputSeq = isLocalProvider
         ? runtime?.getPtyOutputSequence(payload.id)
         : runtime?.onPtyData(payload.id, payload.data, Date.now(), rawLength, payload.transformed)
-      const rendererData = payload.data
-      const preservesSeq = !payload.transformed && rawLength === payload.data.length
-      const startSeq =
-        typeof outputSeq === 'number' ? Math.max(0, outputSeq - rawLength) : undefined
-      if (mainWindow.isDestroyed()) {
-        // Why clear the flush timer: macOS app re-activation otherwise leaks orphaned timers from the previous window's registration.
-        if (flushTimer) {
-          clearTimeout(flushTimer)
-          flushTimer = null
-        }
-        producerFlowControl.releaseAll()
-        clearDeliveryResyncProbe()
-        clearPendingPtyData()
-        pendingOverflowMarkedPtys.clear()
-        rendererDeliveryAccountingByPty.clear()
-        rendererInFlightTotalChars = 0
-        clearDispatcherReadyWatchdog()
-        return
-      }
-      if (rendererExitingPtyIds.size > 0 && rendererExitingPtyIds.has(payload.id)) {
-        return
-      }
-      const settings = getSettings?.()
-      // Why drop before the interactive bypass: runtime already ingested the chunk, so gated PTYs skip both renderer paths and reveal restores from the snapshot.
-      if (shouldDropHiddenRendererPtyData(payload.id, settings)) {
-        const drop = recordHiddenRendererPtyDataDrop(payload.id, payload.data.length)
-        warnIfDroppingHiddenBytesForVisiblePty(payload.id, payload.data.length)
-        if (drop.shouldEmitRestoreMarker) {
-          sendModelRestoreNeededMarker(payload.id, 'hidden-drop', outputSeq)
-        }
-        return
-      }
-      if (rendererData.length === 0 && !payload.transformed) {
-        return
-      }
-      const containsBackgroundOutput =
-        rendererPtyIsKnownHidden(payload.id) || ptyHasHiddenRendererResizeOutput(payload.id)
-      if (containsBackgroundOutput) {
-        markHiddenRendererResizeOutputDelivered(payload.id)
-      }
-      const existing = pendingData.get(payload.id)
-      const overflowMarkedBeforeAppend = pendingOverflowMarkedPtys.has(payload.id)
-      const pending = appendPendingPtyData(
-        payload.id,
-        existing,
-        rendererData,
-        startSeq,
-        preservesSeq,
-        containsBackgroundOutput,
-        rawLength,
-        payload.transformed === true
-      )
-      const shouldEmitPendingCapRestoreMarker =
-        pending.droppedOutput === true &&
-        !overflowMarkedBeforeAppend &&
-        pendingOverflowMarkedPtys.has(payload.id)
-      const nextData = pending.data + getDroppedMode2031RendererData(pending)
-      const isInteractiveOutput = shouldSendInteractiveOutputNow(
-        payload.id,
-        nextData,
-        performance.now()
-      )
-      // Why gate the fast path on the handshake too: else boot-window keystroke echo is sent into a listener-less page and pins the gate.
-      if (isInteractiveOutput && rendererPtyDispatcherReady) {
-        // Why the reserve: keep input echo from being pinned behind unrelated bulk output; it's bounded and the per-PTY cap still prevents an active TUI runaway.
-        if (!canSendPtyDataToRenderer(payload.id, { interactive: true })) {
-          setPendingPtyData(payload.id, pending)
-          if (shouldEmitPendingCapRestoreMarker) {
-            sendModelRestoreNeededMarker(payload.id, 'pending-cap', outputSeq)
-          }
-          updateProducerFlowControl(payload.id)
-          requestDeliveryResyncForGatedPty()
-          return
-        }
-        deletePendingPtyData(payload.id)
-        clearFlushTimerIfIdle()
-        if (shouldEmitPendingCapRestoreMarker) {
-          sendModelRestoreNeededMarker(payload.id, 'pending-cap', outputSeq)
-        }
-        pendingOverflowMarkedPtys.delete(payload.id)
-        // Why immediate: agent TUIs redraw small prompt regions per keystroke; the throughput batch timer would add visible input latency.
-        try {
-          sendPtyDataToRenderer(payload.id, {
-            id: payload.id,
-            data: nextData,
-            ...(typeof pending.startSeq === 'number'
-              ? {
-                  seq: pending.startSeq + (pending.rawLength ?? nextData.length),
-                  rawLength: pending.rawLength ?? nextData.length
-                }
-              : {}),
-            ...(pending.transformed ? { transformed: true } : {}),
-            ...(pending.containsBackgroundOutput === true ? { background: true } : {}),
-            ...(pending.droppedOutput === true ? { droppedOutput: true } : {})
-          })
-        } finally {
-          updateProducerFlowControl(payload.id)
-        }
-        return
-      }
-      setPendingPtyData(payload.id, pending)
-      if (shouldEmitPendingCapRestoreMarker) {
-        sendModelRestoreNeededMarker(payload.id, 'pending-cap', outputSeq)
-      }
-      updateProducerFlowControl(payload.id)
-      // Why probe on data arrival (not flush skips): new output for a fully gated PTY is the moment stuck delivery becomes observable.
-      if (
-        !canSendPtyDataToRenderer(payload.id, { interactive: activeRendererPtys.has(payload.id) })
-      ) {
-        requestDeliveryResyncForGatedPty()
-      }
-      if (!flushTimer) {
-        schedulePendingDataFlush(PTY_BATCH_INTERVAL_MS)
-      }
+      acceptPtyDataForRenderer(payload, outputSeq)
     })
     localExitUnsub = localProvider.onExit((payload) => {
       if (!isCurrentPtyExit(payload)) {
@@ -5623,6 +5789,9 @@ export function registerPtyHandlers(
       const pending = pendingData.get(args.id)
       if (pending && transition.droppable) {
         pendingData.delete(args.id)
+        if (pending.projectionAdmissionIds) {
+          sshOutputIntake?.transferProjections(pending.projectionAdmissionIds, 'hidden-drop')
+        }
         updateProducerFlowControl(args.id)
         pendingOverflowMarkedPtys.delete(args.id)
         const drop = recordHiddenRendererPtyDataDrop(args.id, pending.data.length)

src/main/ipc/ssh-pty-legacy-projection-record.ts

@@ -0,0 +1,193 @@
+import type {
+  Mode2031ReplyDecision,
+  Mode2031ReplyScanState
+} from '../../shared/terminal-color-scheme-protocol'
+import { INITIAL_MODE_2031_REPLY_SCAN_STATE } from '../../shared/terminal-color-scheme-protocol'
+
+export type LegacySshProjectionIdentity = Readonly<{
+  projectionSemanticsId: string
+  ptyId: string
+  providerGeneration: number
+  ptyIncarnation: string
+  displayStart: number
+  displayEnd: number
+  sequenceEnd: number
+  rawLength: number
+  transformed: boolean
+}>
+
+export type LegacySshProjectionSemantics = Readonly<{
+  identity: LegacySshProjectionIdentity
+  beforeScanner: Readonly<Mode2031ReplyScanState>
+  afterScanner: Readonly<Mode2031ReplyScanState>
+  decision: Mode2031ReplyDecision
+}>
+
+export type ProjectionState = 'reserved' | 'committed' | 'published'
+
+export type ProjectionRecord = {
+  semantics: LegacySshProjectionSemantics
+  state: ProjectionState
+  publishedDisplay: number
+  publishedAccounting: number
+  settledAccounting: number
+}
+
+export type PtyProjectionCursor = {
+  providerGeneration: number
+  ptyIncarnation: string
+  displayEnd: number
+  scanner: Mode2031ReplyScanState
+}
+
+export type LegacySshProjectionReservation = Readonly<{
+  semantics: LegacySshProjectionSemantics
+}>
+
+export type LegacySshProjectionDebugSnapshot = {
+  reserved: number
+  committed: number
+  published: number
+  settled: number
+  transferred: number
+  rolledBack: number
+  records: number
+  cursors: number
+}
+
+export function scannerSnapshot(state: Mode2031ReplyScanState): Readonly<Mode2031ReplyScanState> {
+  return Object.freeze({ tail: state.tail, pendingSubscribe: state.pendingSubscribe })
+}
+
+export function projectionError(code: string): Error {
+  return Object.assign(new Error(code), { code })
+}
+
+export function projectionDebugSnapshot(
+  records: ReadonlyMap<string, ProjectionRecord>,
+  cursors: ReadonlyMap<string, PtyProjectionCursor>,
+  terminalCounts: { settled: number; transferred: number; rolledBack: number }
+): LegacySshProjectionDebugSnapshot {
+  const result = {
+    reserved: 0,
+    committed: 0,
+    published: 0,
+    ...terminalCounts,
+    records: records.size,
+    cursors: cursors.size
+  }
+  for (const record of records.values()) {
+    result[record.state]++
+  }
+  return result
+}
+
+export function reclaimProjectionRecord(
+  records: Map<string, ProjectionRecord>,
+  idsByPty: Map<string, string[]>,
+  id: string,
+  ptyId: string
+): void {
+  records.delete(id)
+  const ids = idsByPty.get(ptyId)
+  if (!ids) {
+    return
+  }
+  const index = ids.indexOf(id)
+  if (index >= 0) {
+    ids.splice(index, 1)
+  }
+  if (ids.length === 0) {
+    idsByPty.delete(ptyId)
+  }
+}
+
+export function requireProjectionRecord(
+  records: ReadonlyMap<string, ProjectionRecord>,
+  id: string
+): ProjectionRecord {
+  const record = records.get(id)
+  if (!record) {
+    throw projectionError('ssh_projection_reservation_missing')
+  }
+  return record
+}
+
+export function rollbackUnpublishedProjection(
+  records: Map<string, ProjectionRecord>,
+  cursors: ReadonlyMap<string, PtyProjectionCursor>,
+  idsByPty: Map<string, string[]>,
+  id: string
+): boolean {
+  const record = records.get(id)
+  if (!record || record.state !== 'committed') {
+    return false
+  }
+  const { identity, beforeScanner } = record.semantics
+  const cursor = cursors.get(identity.ptyId)
+  if (
+    !cursor ||
+    cursor.providerGeneration !== identity.providerGeneration ||
+    cursor.ptyIncarnation !== identity.ptyIncarnation ||
+    cursor.displayEnd !== identity.displayEnd
+  ) {
+    return false
+  }
+  cursor.displayEnd = identity.displayStart
+  cursor.scanner = { ...beforeScanner }
+  reclaimProjectionRecord(records, idsByPty, id, identity.ptyId)
+  return true
+}
+
+export function closeProjectionPty(
+  cursors: Map<string, PtyProjectionCursor>,
+  ptyId: string,
+  providerGeneration: number,
+  ptyIncarnation: string,
+  beforeDelete: () => void
+): void {
+  const cursor = cursors.get(ptyId)
+  if (
+    !cursor ||
+    cursor.providerGeneration !== providerGeneration ||
+    cursor.ptyIncarnation !== ptyIncarnation
+  ) {
+    return
+  }
+  beforeDelete()
+  cursors.delete(ptyId)
+}
+
+export function getOrCreateProjectionCursor(
+  cursors: Map<string, PtyProjectionCursor>,
+  args: { ptyId: string; providerGeneration: number; ptyIncarnation: string },
+  replaceGeneration: (providerGeneration: number) => void
+): PtyProjectionCursor {
+  const existing = cursors.get(args.ptyId)
+  if (!existing) {
+    const cursor = {
+      providerGeneration: args.providerGeneration,
+      ptyIncarnation: args.ptyIncarnation,
+      displayEnd: 0,
+      scanner: { ...INITIAL_MODE_2031_REPLY_SCAN_STATE }
+    }
+    cursors.set(args.ptyId, cursor)
+    return cursor
+  }
+  if (args.providerGeneration < existing.providerGeneration) {
+    throw projectionError('ssh_projection_stale_generation')
+  }
+  if (
+    args.providerGeneration === existing.providerGeneration &&
+    args.ptyIncarnation !== existing.ptyIncarnation
+  ) {
+    throw projectionError('ssh_projection_stale_incarnation')
+  }
+  if (args.providerGeneration > existing.providerGeneration) {
+    replaceGeneration(existing.providerGeneration)
+    existing.providerGeneration = args.providerGeneration
+    existing.ptyIncarnation = args.ptyIncarnation
+    existing.scanner = { ...INITIAL_MODE_2031_REPLY_SCAN_STATE }
+  }
+  return existing
+}

src/main/ipc/ssh-pty-legacy-projection.test.ts

@@ -0,0 +1,108 @@
+import { describe, expect, it } from 'vitest'
+import { SshPtyLegacyProjectionLedger } from './ssh-pty-legacy-projection'
+
+function reserve(
+  ledger: SshPtyLegacyProjectionLedger,
+  overrides: Partial<Parameters<SshPtyLegacyProjectionLedger['reserve']>[0]> = {}
+) {
+  return ledger.reserve({
+    ptyId: 'pty-1',
+    providerGeneration: 3,
+    ptyIncarnation: 'incarnation-1',
+    data: 'abc',
+    sequenceEnd: 3,
+    rawLength: 3,
+    transformed: false,
+    ...overrides
+  })
+}
+
+describe('SshPtyLegacyProjectionLedger', () => {
+  it('rolls back scanner and display reservations before commit', () => {
+    const ledger = new SshPtyLegacyProjectionLedger()
+    const partial = reserve(ledger, { data: '\x1b[?20', rawLength: 5, sequenceEnd: 5 })
+    expect(ledger.rollback(partial)).toBe(true)
+
+    const next = reserve(ledger, { data: '31h', sequenceEnd: 3 })
+    expect(next.semantics.identity.displayStart).toBe(0)
+    expect(next.semantics.beforeScanner).toEqual({ tail: '', pendingSubscribe: false })
+    expect(next.semantics.decision).toBeNull()
+  })
+
+  it('keeps immutable generation, incarnation, display, sequence, raw length, and scanner facts', () => {
+    const ledger = new SshPtyLegacyProjectionLedger()
+    const first = reserve(ledger, {
+      data: '\x1b[?2031h',
+      rawLength: 11,
+      sequenceEnd: 11
+    })
+    const semantics = ledger.commit(first)
+
+    expect(semantics.identity).toMatchObject({
+      providerGeneration: 3,
+      ptyIncarnation: 'incarnation-1',
+      displayStart: 0,
+      displayEnd: 8,
+      sequenceEnd: 11,
+      rawLength: 11
+    })
+    expect(semantics.decision).toBe('subscribed')
+    expect(Object.isFrozen(semantics)).toBe(true)
+    expect(Object.isFrozen(semantics.identity)).toBe(true)
+  })
+
+  it('rejects stale generations and resets cross-chunk scanner state on gaps', () => {
+    const ledger = new SshPtyLegacyProjectionLedger()
+    ledger.commit(reserve(ledger, { data: '\x1b[?20', rawLength: 5, sequenceEnd: 5 }))
+    ledger.resetForGap('pty-1')
+    const next = reserve(ledger, { data: '31h', sequenceEnd: 8 })
+    expect(next.semantics.beforeScanner).toEqual({ tail: '', pendingSubscribe: false })
+
+    expect(() => reserve(ledger, { providerGeneration: 2 })).toThrow(
+      'ssh_projection_stale_generation'
+    )
+  })
+
+  it('publishes, settles, and transfers explicit ranges', () => {
+    const ledger = new SshPtyLegacyProjectionLedger()
+    const first = ledger.commit(reserve(ledger))
+    ledger.publishPrefix([first.identity.projectionSemanticsId], 3, 3)
+    expect(ledger.settlePublishedPrefix('pty-1', 2)).toBe(2)
+    expect(ledger.transfer([first.identity.projectionSemanticsId], 'renderer-reload')).toBe(1)
+    expect(ledger.getDebugSnapshot()).toMatchObject({ transferred: 1, records: 0 })
+  })
+
+  it('publishes and settles transformed source accounting with no display text', () => {
+    const ledger = new SshPtyLegacyProjectionLedger()
+    const projection = ledger.commit(
+      reserve(ledger, {
+        data: '',
+        sequenceEnd: 9,
+        rawLength: 9,
+        transformed: true
+      })
+    )
+
+    ledger.publishPrefix([projection.identity.projectionSemanticsId], 0, 9)
+    expect(ledger.settlePublishedPrefix('pty-1', 9)).toBe(9)
+    expect(ledger.getDebugSnapshot()).toMatchObject({ settled: 1, records: 0 })
+  })
+
+  it('reclaims a closed PTY cursor so its id can be reused by a new incarnation', () => {
+    const ledger = new SshPtyLegacyProjectionLedger()
+    ledger.commit(reserve(ledger))
+    ledger.closePty('pty-1', 3, 'incarnation-1', 'pty-exit')
+
+    const next = reserve(ledger, {
+      ptyIncarnation: 'incarnation-2',
+      data: 'next',
+      sequenceEnd: 4,
+      rawLength: 4
+    })
+    expect(next.semantics.identity).toMatchObject({
+      ptyIncarnation: 'incarnation-2',
+      displayStart: 0
+    })
+    expect(ledger.getDebugSnapshot()).toMatchObject({ records: 1, cursors: 1 })
+  })
+})

src/main/ipc/ssh-pty-legacy-projection.ts

@@ -0,0 +1,290 @@
+import {
+  INITIAL_MODE_2031_REPLY_SCAN_STATE,
+  scanMode2031ReplyDecision
+} from '../../shared/terminal-color-scheme-protocol'
+import {
+  closeProjectionPty,
+  getOrCreateProjectionCursor,
+  projectionDebugSnapshot,
+  projectionError,
+  reclaimProjectionRecord,
+  requireProjectionRecord,
+  rollbackUnpublishedProjection,
+  scannerSnapshot,
+  type LegacySshProjectionDebugSnapshot,
+  type LegacySshProjectionReservation,
+  type LegacySshProjectionSemantics,
+  type ProjectionRecord,
+  type PtyProjectionCursor
+} from './ssh-pty-legacy-projection-record'
+
+export type {
+  LegacySshProjectionIdentity,
+  LegacySshProjectionReservation,
+  LegacySshProjectionSemantics
+} from './ssh-pty-legacy-projection-record'
+
+export class SshPtyLegacyProjectionLedger {
+  private nextId = 1
+  private readonly records = new Map<string, ProjectionRecord>()
+  private readonly cursorByPty = new Map<string, PtyProjectionCursor>()
+  private readonly idsByPty = new Map<string, string[]>()
+  private settledCount = 0
+  private transferredCount = 0
+  private rolledBackCount = 0
+
+  reserve(args: {
+    ptyId: string
+    providerGeneration: number
+    ptyIncarnation: string
+    data: string
+    sequenceEnd: number
+    rawLength: number
+    transformed: boolean
+  }): LegacySshProjectionReservation {
+    const cursor = getOrCreateProjectionCursor(this.cursorByPty, args, (generation) =>
+      this.transferGeneration(generation, 'provider-generation-replaced')
+    )
+    if (
+      cursor.providerGeneration !== args.providerGeneration ||
+      cursor.ptyIncarnation !== args.ptyIncarnation
+    ) {
+      throw projectionError('ssh_projection_stale_generation')
+    }
+    const scan = scanMode2031ReplyDecision(cursor.scanner, args.data)
+    const projectionSemanticsId = `ssh-projection:${args.providerGeneration}:${this.nextId++}`
+    const identity = Object.freeze({
+      projectionSemanticsId,
+      ptyId: args.ptyId,
+      providerGeneration: args.providerGeneration,
+      ptyIncarnation: args.ptyIncarnation,
+      displayStart: cursor.displayEnd,
+      displayEnd: cursor.displayEnd + args.data.length,
+      sequenceEnd: args.sequenceEnd,
+      rawLength: args.rawLength,
+      transformed: args.transformed
+    })
+    const semantics = Object.freeze({
+      identity,
+      beforeScanner: scannerSnapshot(cursor.scanner),
+      afterScanner: scannerSnapshot(scan.state),
+      decision: scan.decision
+    })
+    this.records.set(projectionSemanticsId, {
+      semantics,
+      state: 'reserved',
+      publishedDisplay: 0,
+      publishedAccounting: 0,
+      settledAccounting: 0
+    })
+    return Object.freeze({ semantics })
+  }
+
+  commit(reservation: LegacySshProjectionReservation): LegacySshProjectionSemantics {
+    const record = requireProjectionRecord(
+      this.records,
+      reservation.semantics.identity.projectionSemanticsId
+    )
+    if (record.state !== 'reserved') {
+      throw projectionError('ssh_projection_commit_invalid')
+    }
+    const { identity, afterScanner } = record.semantics
+    const cursor = this.cursorByPty.get(identity.ptyId)
+    if (
+      !cursor ||
+      cursor.providerGeneration !== identity.providerGeneration ||
+      cursor.ptyIncarnation !== identity.ptyIncarnation ||
+      cursor.displayEnd !== identity.displayStart
+    ) {
+      this.records.delete(identity.projectionSemanticsId)
+      throw projectionError('ssh_projection_commit_stale')
+    }
+    cursor.displayEnd = identity.displayEnd
+    cursor.scanner = { ...afterScanner }
+    record.state = 'committed'
+    const ids = this.idsByPty.get(identity.ptyId) ?? []
+    ids.push(identity.projectionSemanticsId)
+    this.idsByPty.set(identity.ptyId, ids)
+    return record.semantics
+  }
+
+  rollback(reservation: LegacySshProjectionReservation): boolean {
+    const id = reservation.semantics.identity.projectionSemanticsId
+    const record = this.records.get(id)
+    if (!record || record.state !== 'reserved') {
+      return false
+    }
+    this.records.delete(id)
+    this.rolledBackCount++
+    return true
+  }
+
+  publishPrefix(ids: readonly string[], displayChars: number, accountingChars: number): void {
+    let displayRemaining = Math.max(0, displayChars)
+    let accountingRemaining = Math.max(0, accountingChars)
+    for (const id of ids) {
+      const record = this.records.get(id)
+      if (!record) {
+        continue
+      }
+      const displayLength =
+        record.semantics.identity.displayEnd - record.semantics.identity.displayStart
+      const unpublishedDisplay = displayLength - record.publishedDisplay
+      const unpublishedAccounting = record.semantics.identity.rawLength - record.publishedAccounting
+      if (displayLength === 0 && unpublishedAccounting > 0) {
+        const publishAccounting = Math.min(accountingRemaining, unpublishedAccounting)
+        if (publishAccounting !== unpublishedAccounting) {
+          throw projectionError('ssh_projection_indivisible_split')
+        }
+        record.publishedAccounting += publishAccounting
+        record.state = 'published'
+        accountingRemaining -= publishAccounting
+        continue
+      }
+      if (unpublishedDisplay <= 0) {
+        continue
+      }
+      const publishDisplay = Math.min(displayRemaining, unpublishedDisplay)
+      if (publishDisplay <= 0) {
+        break
+      }
+      const indivisible = record.semantics.identity.transformed
+      if (indivisible && publishDisplay !== unpublishedDisplay) {
+        throw projectionError('ssh_projection_indivisible_split')
+      }
+      const publishAccounting =
+        publishDisplay === unpublishedDisplay
+          ? Math.min(accountingRemaining, unpublishedAccounting)
+          : publishDisplay
+      record.publishedDisplay += publishDisplay
+      record.publishedAccounting += publishAccounting
+      record.state = 'published'
+      displayRemaining -= publishDisplay
+      accountingRemaining -= publishAccounting
+    }
+    if (displayRemaining !== 0 || accountingRemaining !== 0) {
+      throw projectionError('ssh_projection_publish_range_mismatch')
+    }
+  }
+
+  settlePublishedPrefix(ptyId: string, accountingChars: number): number {
+    let remaining = Math.max(0, accountingChars)
+    let settled = 0
+    for (const id of this.idsByPty.get(ptyId)?.slice() ?? []) {
+      const record = this.records.get(id)
+      if (!record) {
+        continue
+      }
+      const available = record.publishedAccounting - record.settledAccounting
+      if (available <= 0) {
+        continue
+      }
+      const take = Math.min(remaining, available)
+      record.settledAccounting += take
+      settled += take
+      remaining -= take
+      if (
+        record.settledAccounting === record.semantics.identity.rawLength &&
+        record.publishedDisplay ===
+          record.semantics.identity.displayEnd - record.semantics.identity.displayStart
+      ) {
+        this.settledCount++
+        reclaimProjectionRecord(this.records, this.idsByPty, id, ptyId)
+      }
+      if (remaining === 0) {
+        break
+      }
+    }
+    return settled
+  }
+
+  transfer(ids: readonly string[], _reason: string): number {
+    let transferred = 0
+    for (const id of ids.slice()) {
+      const record = this.records.get(id)
+      if (!record || record.state === 'reserved') {
+        continue
+      }
+      this.transferredCount++
+      reclaimProjectionRecord(this.records, this.idsByPty, id, record.semantics.identity.ptyId)
+      transferred++
+    }
+    return transferred
+  }
+
+  transferUnpublished(id: string, _reason: string): boolean {
+    const transferred = rollbackUnpublishedProjection(
+      this.records,
+      this.cursorByPty,
+      this.idsByPty,
+      id
+    )
+    this.transferredCount += Number(transferred)
+    return transferred
+  }
+
+  transferGeneration(providerGeneration: number, reason: string): number {
+    const ids: string[] = []
+    for (const [id, record] of this.records) {
+      if (record.semantics.identity.providerGeneration === providerGeneration) {
+        ids.push(id)
+      }
+    }
+    return this.transfer(ids, reason)
+  }
+
+  transferPty(ptyId: string, reason: string): number {
+    return this.transfer(this.idsByPty.get(ptyId) ?? [], reason)
+  }
+
+  closePty(
+    ptyId: string,
+    providerGeneration: number,
+    ptyIncarnation: string,
+    reason: string
+  ): void {
+    closeProjectionPty(this.cursorByPty, ptyId, providerGeneration, ptyIncarnation, () => {
+      this.transferPty(ptyId, reason)
+      this.idsByPty.delete(ptyId)
+    })
+  }
+
+  closeGeneration(providerGeneration: number, reason: string): void {
+    for (const [id, record] of this.records) {
+      if (record.semantics.identity.providerGeneration !== providerGeneration) {
+        continue
+      }
+      if (record.state === 'reserved') {
+        this.records.delete(id)
+        this.rolledBackCount++
+      } else {
+        this.transfer([id], reason)
+      }
+    }
+    for (const [ptyId, cursor] of this.cursorByPty) {
+      if (cursor.providerGeneration === providerGeneration) {
+        this.cursorByPty.delete(ptyId)
+        this.idsByPty.delete(ptyId)
+      }
+    }
+  }
+
+  resetForGap(ptyId: string): void {
+    const cursor = this.cursorByPty.get(ptyId)
+    if (cursor) {
+      cursor.scanner = { ...INITIAL_MODE_2031_REPLY_SCAN_STATE }
+    }
+  }
+
+  get(id: string): LegacySshProjectionSemantics | undefined {
+    return this.records.get(id)?.semantics
+  }
+
+  getDebugSnapshot(): LegacySshProjectionDebugSnapshot {
+    return projectionDebugSnapshot(this.records, this.cursorByPty, {
+      settled: this.settledCount,
+      transferred: this.transferredCount,
+      rolledBack: this.rolledBackCount
+    })
+  }
+}

src/main/ipc/ssh-pty-model-admission-contract.ts

@@ -0,0 +1,34 @@
+export type SshPtyModelAdmissionKey = Readonly<{
+  ptyId: string
+  providerGeneration: number
+}>
+
+export type SshPtyModelAdmissionReceipt = Readonly<{
+  ptyId: string
+  providerGeneration: number
+  sequence: number
+}>
+
+export type SshPtyModelAdmissionDebugSnapshot = Readonly<{
+  sourceUnits: number
+  bytes: number
+  pressureFrames: number
+  pressureBytes: number
+  pausedPtys: number
+}>
+
+export type SshPtyModelAdmissionOptions = {
+  perPtyHighSourceUnits?: number
+  perPtyHighBytes?: number
+  perPtyLowSourceUnits?: number
+  perPtyLowBytes?: number
+  globalHighSourceUnits?: number
+  globalHighBytes?: number
+  globalLowSourceUnits?: number
+  globalLowBytes?: number
+  pressureMaxFrames?: number
+  pressureMaxBytes?: number
+  pauseProvider?: (key: SshPtyModelAdmissionKey) => boolean
+  resumeProvider?: (key: SshPtyModelAdmissionKey) => void
+  closeProvider?: (providerGeneration: number, reason: string) => void
+}

src/main/ipc/ssh-pty-model-admission-entry.ts

@@ -0,0 +1,110 @@
+import type {
+  SshPtyModelAdmissionKey,
+  SshPtyModelAdmissionReceipt
+} from './ssh-pty-model-admission-contract'
+import type { SshPtyModelAdmissionLimits } from './ssh-pty-model-admission-limits'
+
+export type AdmissionCharge = { sourceUnits: number; bytes: number }
+
+export type AdmissionEntry = {
+  key: SshPtyModelAdmissionKey
+  charge: AdmissionCharge
+  run: () => { sequence: number; completion: Promise<void> }
+  resolve: (receipt: SshPtyModelAdmissionReceipt) => void
+  reject: (error: Error) => void
+  state: 'queued' | 'running' | 'pressure'
+}
+
+export type PtyUsage = AdmissionCharge & {
+  queued: AdmissionEntry[]
+  running: AdmissionEntry | null
+}
+
+export function admissionError(code: string): Error {
+  return Object.assign(new Error(code), { code })
+}
+
+export function retainedBytes(data: string): number {
+  return Math.max(Buffer.byteLength(data, 'utf8'), 2 * data.length) + 128
+}
+
+export function admissionKeyId(key: SshPtyModelAdmissionKey): string {
+  return `${key.providerGeneration}\0${key.ptyId}`
+}
+
+export function canReserveAdmission(args: {
+  key: SshPtyModelAdmissionKey
+  charge: AdmissionCharge
+  limits: SshPtyModelAdmissionLimits
+  usageByPty: ReadonlyMap<string, PtyUsage>
+  closingGenerations: ReadonlySet<number>
+  globalSourceUnits: number
+  globalBytes: number
+}): boolean {
+  if (args.closingGenerations.has(args.key.providerGeneration)) {
+    return false
+  }
+  const usage = args.usageByPty.get(admissionKeyId(args.key))
+  return (
+    (usage?.sourceUnits ?? 0) + args.charge.sourceUnits <= args.limits.perPtyHighSourceUnits &&
+    (usage?.bytes ?? 0) + args.charge.bytes <= args.limits.perPtyHighBytes &&
+    args.globalSourceUnits + args.charge.sourceUnits <= args.limits.globalHighSourceUnits &&
+    args.globalBytes + args.charge.bytes <= args.limits.globalHighBytes
+  )
+}
+
+export function pressureHasAdmissionKey(
+  entries: readonly AdmissionEntry[],
+  key: SshPtyModelAdmissionKey
+): boolean {
+  const id = admissionKeyId(key)
+  return entries.some((entry) => admissionKeyId(entry.key) === id)
+}
+
+export function takePressureEntriesForGeneration(
+  entries: AdmissionEntry[],
+  providerGeneration: number
+): AdmissionEntry[] {
+  const removed: AdmissionEntry[] = []
+  for (let index = entries.length - 1; index >= 0; index--) {
+    if (entries[index]!.key.providerGeneration === providerGeneration) {
+      removed.unshift(entries.splice(index, 1)[0]!)
+    }
+  }
+  return removed
+}
+
+export function takePausedGeneration(
+  paused: Map<string, SshPtyModelAdmissionKey>,
+  providerGeneration: number
+): SshPtyModelAdmissionKey[] {
+  const removed: SshPtyModelAdmissionKey[] = []
+  for (const [id, key] of paused) {
+    if (key.providerGeneration === providerGeneration) {
+      paused.delete(id)
+      removed.push(key)
+    }
+  }
+  return removed
+}
+
+export function resolveAdmissionIdleWaiters(
+  usageByPty: ReadonlyMap<string, PtyUsage>,
+  pressure: readonly AdmissionEntry[],
+  waitersById: Map<string, Set<() => void>>,
+  id: string
+): void {
+  const usage = usageByPty.get(id)
+  const stillPressured = pressure.some((entry) => admissionKeyId(entry.key) === id)
+  if (usage?.running || (usage?.queued.length ?? 0) > 0 || stillPressured) {
+    return
+  }
+  const waiters = waitersById.get(id)
+  if (!waiters) {
+    return
+  }
+  waitersById.delete(id)
+  for (const resolve of waiters) {
+    resolve()
+  }
+}

src/main/ipc/ssh-pty-model-admission-limits.ts

@@ -0,0 +1,38 @@
+import type { SshPtyModelAdmissionOptions } from './ssh-pty-model-admission-contract'
+
+export type SshPtyModelAdmissionLimits = Readonly<{
+  perPtyHighSourceUnits: number
+  perPtyHighBytes: number
+  perPtyLowSourceUnits: number
+  perPtyLowBytes: number
+  globalHighSourceUnits: number
+  globalHighBytes: number
+  globalLowSourceUnits: number
+  globalLowBytes: number
+  pressureMaxFrames: number
+  pressureMaxBytes: number
+}>
+
+const DEFAULT_PER_PTY_HIGH_SOURCE_UNITS = 256 * 1024
+const DEFAULT_PER_PTY_HIGH_BYTES = 2 * 1024 * 1024
+const DEFAULT_GLOBAL_HIGH_SOURCE_UNITS = 50 * DEFAULT_PER_PTY_HIGH_SOURCE_UNITS
+const DEFAULT_GLOBAL_HIGH_BYTES = 64 * 1024 * 1024
+
+export function resolveSshPtyModelAdmissionLimits(
+  options: SshPtyModelAdmissionOptions
+): SshPtyModelAdmissionLimits {
+  const perPtyHighSourceUnits = options.perPtyHighSourceUnits ?? DEFAULT_PER_PTY_HIGH_SOURCE_UNITS
+  const perPtyHighBytes = options.perPtyHighBytes ?? DEFAULT_PER_PTY_HIGH_BYTES
+  return {
+    perPtyHighSourceUnits,
+    perPtyHighBytes,
+    perPtyLowSourceUnits: options.perPtyLowSourceUnits ?? Math.floor(perPtyHighSourceUnits / 2),
+    perPtyLowBytes: options.perPtyLowBytes ?? Math.floor(perPtyHighBytes / 2),
+    globalHighSourceUnits: options.globalHighSourceUnits ?? DEFAULT_GLOBAL_HIGH_SOURCE_UNITS,
+    globalHighBytes: options.globalHighBytes ?? DEFAULT_GLOBAL_HIGH_BYTES,
+    globalLowSourceUnits: options.globalLowSourceUnits ?? 8 * 1024 * 1024,
+    globalLowBytes: options.globalLowBytes ?? 48 * 1024 * 1024,
+    pressureMaxFrames: options.pressureMaxFrames ?? 64,
+    pressureMaxBytes: options.pressureMaxBytes ?? 1024 * 1024
+  }
+}

src/main/ipc/ssh-pty-model-admission.test.ts

@@ -0,0 +1,34 @@
+import { describe, expect, it } from 'vitest'
+import { SshPtyModelAdmission } from './ssh-pty-model-admission'
+
+function deferred() {
+  let resolve!: () => void
+  const promise = new Promise<void>((promiseResolve) => {
+    resolve = promiseResolve
+  })
+  return { promise, resolve }
+}
+
+describe('SshPtyModelAdmission', () => {
+  it('owns the sole running entry through disposal', async () => {
+    const completion = deferred()
+    const admission = new SshPtyModelAdmission()
+    const receipt = admission.accept({ ptyId: 'pty-1', providerGeneration: 7 }, 'data', 4, () => ({
+      sequence: 4,
+      completion: completion.promise
+    }))
+    let idle = false
+    const idleReceipt = admission.whenIdle({ ptyId: 'pty-1', providerGeneration: 7 }).then(() => {
+      idle = true
+    })
+
+    admission.dispose()
+    await Promise.resolve()
+    expect(idle).toBe(false)
+    completion.resolve()
+
+    await expect(receipt).rejects.toThrow('ssh_model_admission_generation_closed')
+    await idleReceipt
+    expect(admission.getDebugSnapshot()).toMatchObject({ sourceUnits: 0, bytes: 0 })
+  })
+})

src/main/ipc/ssh-pty-model-admission.ts

@@ -0,0 +1,315 @@
+import type {
+  SshPtyModelAdmissionDebugSnapshot,
+  SshPtyModelAdmissionKey,
+  SshPtyModelAdmissionOptions,
+  SshPtyModelAdmissionReceipt
+} from './ssh-pty-model-admission-contract'
+import {
+  admissionError,
+  admissionKeyId,
+  canReserveAdmission,
+  pressureHasAdmissionKey,
+  retainedBytes,
+  resolveAdmissionIdleWaiters,
+  takePausedGeneration,
+  takePressureEntriesForGeneration,
+  type AdmissionCharge,
+  type AdmissionEntry,
+  type PtyUsage
+} from './ssh-pty-model-admission-entry'
+import {
+  resolveSshPtyModelAdmissionLimits,
+  type SshPtyModelAdmissionLimits
+} from './ssh-pty-model-admission-limits'
+
+export type {
+  SshPtyModelAdmissionKey,
+  SshPtyModelAdmissionOptions,
+  SshPtyModelAdmissionReceipt
+} from './ssh-pty-model-admission-contract'
+
+export class SshPtyModelAdmission {
+  private readonly limits: SshPtyModelAdmissionLimits
+  private readonly pauseProvider: (key: SshPtyModelAdmissionKey) => boolean
+  private readonly resumeProvider: (key: SshPtyModelAdmissionKey) => void
+  private readonly closeProvider: (providerGeneration: number, reason: string) => void
+  private readonly usageByPty = new Map<string, PtyUsage>()
+  private readonly pressure: AdmissionEntry[] = []
+  private readonly pausedKeys = new Map<string, SshPtyModelAdmissionKey>()
+  private readonly idleWaiters = new Map<string, Set<() => void>>()
+  private readonly closingGenerations = new Set<number>()
+  private globalSourceUnits = 0
+  private globalBytes = 0
+  private pressureBytes = 0
+  private disposed = false
+
+  constructor(options: SshPtyModelAdmissionOptions = {}) {
+    this.limits = resolveSshPtyModelAdmissionLimits(options)
+    this.pauseProvider = options.pauseProvider ?? (() => false)
+    this.resumeProvider = options.resumeProvider ?? (() => {})
+    this.closeProvider = options.closeProvider ?? (() => {})
+  }
+
+  accept(
+    key: SshPtyModelAdmissionKey,
+    data: string,
+    sourceUnits: number,
+    run: () => { sequence: number; completion: Promise<void> }
+  ): Promise<SshPtyModelAdmissionReceipt> {
+    if (this.disposed) {
+      return Promise.reject(admissionError('ssh_model_admission_disposed'))
+    }
+    if (this.closingGenerations.has(key.providerGeneration)) {
+      return Promise.reject(admissionError('ssh_model_admission_generation_closed'))
+    }
+    const charge = { sourceUnits, bytes: retainedBytes(data) }
+    return new Promise((resolve, reject) => {
+      const entry: AdmissionEntry = {
+        key: { ...key },
+        charge,
+        run,
+        resolve,
+        reject,
+        state: 'queued'
+      }
+      if (this.canReserve(key, charge) && !pressureHasAdmissionKey(this.pressure, key)) {
+        this.reserveAndQueue(entry)
+        return
+      }
+      this.admitPressure(entry)
+    })
+  }
+
+  closeGeneration(providerGeneration: number, reason = 'provider-generation-closed'): void {
+    this.closingGenerations.add(providerGeneration)
+    const error = admissionError(reason)
+    for (const entry of takePressureEntriesForGeneration(this.pressure, providerGeneration)) {
+      this.pressureBytes -= entry.charge.bytes
+      entry.reject(error)
+    }
+    for (const [id, usage] of this.usageByPty) {
+      const queued = usage.queued.filter(
+        (entry) => entry.key.providerGeneration === providerGeneration
+      )
+      usage.queued = usage.queued.filter(
+        (entry) => entry.key.providerGeneration !== providerGeneration
+      )
+      for (const entry of queued) {
+        this.release(entry.key, entry.charge)
+        entry.reject(error)
+      }
+      if (!usage.running && usage.queued.length === 0 && usage.sourceUnits === 0) {
+        this.usageByPty.delete(id)
+      }
+      resolveAdmissionIdleWaiters(this.usageByPty, this.pressure, this.idleWaiters, id)
+    }
+    for (const key of takePausedGeneration(this.pausedKeys, providerGeneration)) {
+      try {
+        this.resumeProvider(key)
+      } catch {}
+    }
+    const generationPrefix = `${providerGeneration}\0`
+    for (const id of this.idleWaiters.keys()) {
+      if (id.startsWith(generationPrefix)) {
+        resolveAdmissionIdleWaiters(this.usageByPty, this.pressure, this.idleWaiters, id)
+      }
+    }
+  }
+
+  whenIdle(key: SshPtyModelAdmissionKey): Promise<void> {
+    const id = admissionKeyId(key)
+    const usage = this.usageByPty.get(id)
+    const hasPressure = this.pressure.some(
+      (entry) =>
+        entry.key.providerGeneration === key.providerGeneration && entry.key.ptyId === key.ptyId
+    )
+    if ((!usage || (!usage.running && usage.queued.length === 0)) && !hasPressure) {
+      return Promise.resolve()
+    }
+    return new Promise((resolve) => {
+      const waiters = this.idleWaiters.get(id) ?? new Set<() => void>()
+      waiters.add(resolve)
+      this.idleWaiters.set(id, waiters)
+    })
+  }
+
+  dispose(): void {
+    if (this.disposed) {
+      return
+    }
+    this.disposed = true
+    const generations = new Set<number>()
+    for (const usage of this.usageByPty.values()) {
+      if (usage.running) {
+        generations.add(usage.running.key.providerGeneration)
+      }
+      for (const entry of usage.queued) {
+        generations.add(entry.key.providerGeneration)
+      }
+    }
+    for (const entry of this.pressure) {
+      generations.add(entry.key.providerGeneration)
+    }
+    for (const generation of generations) {
+      this.closeGeneration(generation, 'ssh_model_admission_disposed')
+    }
+  }
+
+  getDebugSnapshot(): SshPtyModelAdmissionDebugSnapshot {
+    return {
+      sourceUnits: this.globalSourceUnits,
+      bytes: this.globalBytes,
+      pressureFrames: this.pressure.length,
+      pressureBytes: this.pressureBytes,
+      pausedPtys: this.pausedKeys.size
+    }
+  }
+
+  private canReserve(key: SshPtyModelAdmissionKey, charge: AdmissionCharge): boolean {
+    return canReserveAdmission({
+      key,
+      charge,
+      limits: this.limits,
+      usageByPty: this.usageByPty,
+      closingGenerations: this.closingGenerations,
+      globalSourceUnits: this.globalSourceUnits,
+      globalBytes: this.globalBytes
+    })
+  }
+
+  private reserveAndQueue(entry: AdmissionEntry): void {
+    const id = admissionKeyId(entry.key)
+    let usage = this.usageByPty.get(id)
+    if (!usage) {
+      usage = { sourceUnits: 0, bytes: 0, queued: [], running: null }
+      this.usageByPty.set(id, usage)
+    }
+    usage.sourceUnits += entry.charge.sourceUnits
+    usage.bytes += entry.charge.bytes
+    this.globalSourceUnits += entry.charge.sourceUnits
+    this.globalBytes += entry.charge.bytes
+    entry.state = 'queued'
+    usage.queued.push(entry)
+    this.startNext(id, usage)
+  }
+
+  private admitPressure(entry: AdmissionEntry): void {
+    const id = admissionKeyId(entry.key)
+    const paused = this.pausedKeys.has(id) || this.pauseProvider(entry.key)
+    if (paused) {
+      this.pausedKeys.set(id, entry.key)
+    }
+    if (
+      !paused ||
+      this.pressure.length >= this.limits.pressureMaxFrames ||
+      this.pressureBytes + entry.charge.bytes > this.limits.pressureMaxBytes
+    ) {
+      entry.reject(admissionError('ssh_model_admission_pressure_exhausted'))
+      this.closeProvider(entry.key.providerGeneration, 'model-admission-pressure')
+      return
+    }
+    entry.state = 'pressure'
+    this.pressure.push(entry)
+    this.pressureBytes += entry.charge.bytes
+  }
+
+  private startNext(id: string, usage: PtyUsage): void {
+    if (usage.running) {
+      return
+    }
+    const entry = usage.queued.shift()
+    if (!entry) {
+      this.maybeResumeAndPromote()
+      return
+    }
+    usage.running = entry
+    entry.state = 'running'
+    let execution: { sequence: number; completion: Promise<void> }
+    try {
+      execution = entry.run()
+    } catch (error) {
+      this.finishEntry(id, usage, entry)
+      entry.reject(error instanceof Error ? error : new Error(String(error)))
+      return
+    }
+    void execution.completion.then(
+      () => {
+        this.finishEntry(id, usage, entry)
+        if (this.closingGenerations.has(entry.key.providerGeneration)) {
+          entry.reject(admissionError('ssh_model_admission_generation_closed'))
+        } else {
+          entry.resolve({
+            ptyId: entry.key.ptyId,
+            providerGeneration: entry.key.providerGeneration,
+            sequence: execution.sequence
+          })
+        }
+      },
+      (error) => {
+        this.finishEntry(id, usage, entry)
+        entry.reject(error instanceof Error ? error : new Error(String(error)))
+      }
+    )
+  }
+
+  private finishEntry(id: string, usage: PtyUsage, entry: AdmissionEntry): void {
+    if (usage.running === entry) {
+      usage.running = null
+    }
+    this.release(entry.key, entry.charge)
+    this.startNext(id, usage)
+    if (!usage.running && usage.queued.length === 0 && usage.sourceUnits === 0) {
+      this.usageByPty.delete(id)
+    }
+    resolveAdmissionIdleWaiters(this.usageByPty, this.pressure, this.idleWaiters, id)
+  }
+
+  private release(key: SshPtyModelAdmissionKey, charge: AdmissionCharge): void {
+    const usage = this.usageByPty.get(admissionKeyId(key))
+    if (usage) {
+      usage.sourceUnits = Math.max(0, usage.sourceUnits - charge.sourceUnits)
+      usage.bytes = Math.max(0, usage.bytes - charge.bytes)
+    }
+    this.globalSourceUnits = Math.max(0, this.globalSourceUnits - charge.sourceUnits)
+    this.globalBytes = Math.max(0, this.globalBytes - charge.bytes)
+    this.maybeResumeAndPromote()
+  }
+
+  private maybeResumeAndPromote(): void {
+    for (let index = 0; index < this.pressure.length; ) {
+      const entry = this.pressure[index]!
+      const hasEarlierEntryForPty = this.pressure
+        .slice(0, index)
+        .some((earlier) => admissionKeyId(earlier.key) === admissionKeyId(entry.key))
+      if (hasEarlierEntryForPty || !this.canReserve(entry.key, entry.charge)) {
+        index++
+        continue
+      }
+      this.pressure.splice(index, 1)
+      this.pressureBytes -= entry.charge.bytes
+      this.reserveAndQueue(entry)
+    }
+    if (
+      this.globalSourceUnits > this.limits.globalLowSourceUnits ||
+      this.globalBytes > this.limits.globalLowBytes
+    ) {
+      return
+    }
+    for (const [id, key] of this.pausedKeys) {
+      const usage = this.usageByPty.get(id)
+      const stillPressured = this.pressure.some(
+        (entry) =>
+          entry.key.providerGeneration === key.providerGeneration && entry.key.ptyId === key.ptyId
+      )
+      if (
+        stillPressured ||
+        (usage?.sourceUnits ?? 0) > this.limits.perPtyLowSourceUnits ||
+        (usage?.bytes ?? 0) > this.limits.perPtyLowBytes
+      ) {
+        continue
+      }
+      this.pausedKeys.delete(id)
+      this.resumeProvider(key)
+    }
+  }
+}

src/main/ipc/ssh-pty-output-intake-contract.ts

@@ -0,0 +1,40 @@
+import type { LegacySshProjectionSemantics } from './ssh-pty-legacy-projection'
+import type {
+  SshPtyModelAdmissionOptions,
+  SshPtyModelAdmissionReceipt
+} from './ssh-pty-model-admission-contract'
+
+export type SshPtyOutputDataEvent = Readonly<{
+  id: string
+  data: string
+  providerGeneration: number
+  ptyIncarnation: string
+  rawLength: number
+  transformed: boolean
+  sequence?: number
+}>
+
+export type SshPtyOutputExitEvent = Readonly<{
+  id: string
+  code: number
+  providerGeneration: number
+  ptyIncarnation: string
+}>
+
+export type SshPtyOutputReceipt = SshPtyModelAdmissionReceipt &
+  Readonly<{ projection: LegacySshProjectionSemantics }>
+
+export type SshPtyOutputIntakeDependencies = {
+  getModelSequence: (id: string) => number
+  acceptModel: (event: SshPtyOutputDataEvent) => { sequence: number; completion: Promise<void> }
+  project: (event: SshPtyOutputDataEvent, projection: LegacySshProjectionSemantics) => void
+  finalizeExit: (event: SshPtyOutputExitEvent) => void
+  pauseProvider?: (providerGeneration: number, id: string) => boolean
+  resumeProvider?: (providerGeneration: number, id: string) => void
+  closeProvider?: (providerGeneration: number, reason: string) => void
+  onGenerationClosed?: (providerGeneration: number, reason: string) => void
+}
+
+export type SshPtyOutputIntakeOptions = SshPtyModelAdmissionOptions & {
+  exitBarrierMs?: number
+}

src/main/ipc/ssh-pty-output-intake-registry.ts

@@ -0,0 +1,47 @@
+import type { SshPtyOutputIntake } from './ssh-pty-output-intake'
+import type {
+  SshPtyOutputDataEvent,
+  SshPtyOutputExitEvent,
+  SshPtyOutputReceipt
+} from './ssh-pty-output-intake-contract'
+
+let installedIntake: SshPtyOutputIntake | null = null
+let nextProviderGeneration = 1
+
+export function allocateSshPtyProviderGeneration(): number {
+  return nextProviderGeneration++
+}
+
+export function installSshPtyOutputIntake(intake: SshPtyOutputIntake): () => void {
+  const previous = installedIntake
+  installedIntake = intake
+  previous?.dispose()
+  return () => {
+    if (installedIntake === intake) {
+      installedIntake = null
+      intake.dispose()
+    }
+  }
+}
+
+export function acceptSshPtyOutputData(event: SshPtyOutputDataEvent): Promise<SshPtyOutputReceipt> {
+  return installedIntake
+    ? installedIntake.acceptData(event)
+    : Promise.reject(outputIntakeUnavailableError())
+}
+
+export function acceptSshPtyOutputExit(event: SshPtyOutputExitEvent): Promise<void> {
+  return installedIntake
+    ? installedIntake.acceptExit(event)
+    : Promise.reject(outputIntakeUnavailableError())
+}
+
+export function closeSshPtyOutputGeneration(providerGeneration: number, reason: string): void {
+  installedIntake?.closeGeneration(providerGeneration, reason)
+}
+
+function outputIntakeUnavailableError(): Error {
+  return Object.assign(new Error('ssh_output_intake_unavailable'), {
+    code: 'ssh_output_intake_unavailable'
+  })
+}

src/main/ipc/ssh-pty-output-intake-validation.ts

@@ -0,0 +1,16 @@
+export type SshPtyExitBarrier = {
+  timer: ReturnType<typeof setTimeout>
+  reject: (error: Error) => void
+}
+
+export function outputIntakeError(code: string): Error {
+  return Object.assign(new Error(code), { code })
+}
+
+export function validOutputLength(value: number): boolean {
+  return Number.isSafeInteger(value) && value >= 0
+}
+
+export function sshPtyGenerationKey(ptyId: string, providerGeneration: number): string {
+  return `${providerGeneration}\0${ptyId}`
+}

src/main/ipc/ssh-pty-output-intake.test.ts

@@ -0,0 +1,349 @@
+import { describe, expect, it, vi } from 'vitest'
+import type { LegacySshProjectionSemantics } from './ssh-pty-legacy-projection'
+import {
+  SshPtyOutputIntake,
+  type SshPtyOutputDataEvent,
+  type SshPtyOutputIntakeDependencies
+} from './ssh-pty-output-intake'
+
+function deferred() {
+  let resolve!: () => void
+  let reject!: (error: Error) => void
+  const promise = new Promise<void>((promiseResolve, promiseReject) => {
+    resolve = promiseResolve
+    reject = promiseReject
+  })
+  return { promise, resolve, reject }
+}
+
+function event(overrides: Partial<SshPtyOutputDataEvent> = {}): SshPtyOutputDataEvent {
+  return {
+    id: 'pty-1',
+    data: 'aaaa',
+    providerGeneration: 1,
+    ptyIncarnation: 'incarnation-1',
+    rawLength: 4,
+    transformed: false,
+    ...overrides
+  }
+}
+
+function createHarness(
+  overrides: Partial<SshPtyOutputIntakeDependencies> = {},
+  options: ConstructorParameters<typeof SshPtyOutputIntake>[1] = {}
+) {
+  let sequence = 0
+  const completions: ReturnType<typeof deferred>[] = []
+  const order: string[] = []
+  const dependencies: SshPtyOutputIntakeDependencies = {
+    getModelSequence: () => sequence,
+    acceptModel: (input) => {
+      order.push(`model:${input.data}`)
+      sequence += input.rawLength
+      const completion = deferred()
+      completions.push(completion)
+      return { sequence, completion: completion.promise }
+    },
+    project: (input) => order.push(`project:${input.data}`),
+    finalizeExit: () => order.push('exit'),
+    pauseProvider: vi.fn(() => true),
+    resumeProvider: vi.fn(),
+    closeProvider: vi.fn(),
+    ...overrides
+  }
+  return {
+    intake: new SshPtyOutputIntake(dependencies, options),
+    dependencies,
+    completions,
+    order
+  }
+}
+
+describe('SshPtyOutputIntake', () => {
+  it('plateaus at the model and pressure budgets, then resumes below low water', async () => {
+    const harness = createHarness(
+      {},
+      {
+        perPtyHighSourceUnits: 4,
+        perPtyHighBytes: 1024,
+        perPtyLowSourceUnits: 1,
+        perPtyLowBytes: 256,
+        globalHighSourceUnits: 4,
+        globalHighBytes: 1024,
+        globalLowSourceUnits: 1,
+        globalLowBytes: 256,
+        pressureMaxFrames: 2,
+        pressureMaxBytes: 1024
+      }
+    )
+    const first = harness.intake.acceptData(event())
+    const second = harness.intake.acceptData(event({ data: 'bbbb' }))
+    const third = harness.intake.acceptData(event({ data: 'cccc' }))
+    const rejected = harness.intake.acceptData(event({ data: 'dddd' }))
+
+    expect(harness.intake.getDebugSnapshot().model).toMatchObject({
+      sourceUnits: 4,
+      pressureFrames: 2
+    })
+    await expect(rejected).rejects.toThrow('ssh_model_admission_pressure_exhausted')
+    expect(harness.dependencies.closeProvider).toHaveBeenCalledWith(1, expect.any(String))
+
+    harness.completions[0]!.resolve()
+    await first
+    harness.completions[1]!.resolve()
+    await second
+    harness.completions[2]!.resolve()
+    await third
+
+    expect(harness.intake.getDebugSnapshot().model).toMatchObject({
+      sourceUnits: 0,
+      bytes: 0,
+      pressureFrames: 0
+    })
+    expect(harness.dependencies.resumeProvider).toHaveBeenCalled()
+  })
+
+  it('preserves per-PTY FIFO while differently sized pressure entries wait', async () => {
+    const harness = createHarness(
+      {},
+      {
+        perPtyHighSourceUnits: 4,
+        perPtyHighBytes: 4096,
+        globalHighSourceUnits: 4,
+        globalHighBytes: 4096,
+        pressureMaxFrames: 4,
+        pressureMaxBytes: 4096
+      }
+    )
+    const receipts = [
+      harness.intake.acceptData(event({ data: 'a', rawLength: 1 })),
+      harness.intake.acceptData(event({ data: 'bbb', rawLength: 3 })),
+      harness.intake.acceptData(event({ data: 'cc', rawLength: 2 })),
+      harness.intake.acceptData(event({ data: 'd', rawLength: 1 }))
+    ]
+
+    harness.completions[0]!.resolve()
+    await receipts[0]
+    harness.completions[1]!.resolve()
+    await receipts[1]
+    harness.completions[2]!.resolve()
+    await receipts[2]
+    harness.completions[3]!.resolve()
+    await receipts[3]
+
+    expect(harness.order).toEqual([
+      'model:a',
+      'project:a',
+      'model:bbb',
+      'project:bbb',
+      'model:cc',
+      'project:cc',
+      'model:d',
+      'project:d'
+    ])
+  })
+
+  it('assigns projection sequence ends when queued model capture begins', async () => {
+    const harness = createHarness(
+      {},
+      {
+        perPtyHighSourceUnits: 12,
+        perPtyHighBytes: 4096,
+        globalHighSourceUnits: 12,
+        globalHighBytes: 4096
+      }
+    )
+    const first = harness.intake.acceptData(event({ data: 'aaaa' }))
+    const second = harness.intake.acceptData(event({ data: 'bbbb' }))
+    const third = harness.intake.acceptData(event({ data: 'cccc' }))
+
+    harness.completions[0]!.resolve()
+    expect((await first).projection.identity.sequenceEnd).toBe(4)
+    harness.completions[1]!.resolve()
+    expect((await second).projection.identity.sequenceEnd).toBe(8)
+    harness.completions[2]!.resolve()
+    expect((await third).projection.identity.sequenceEnd).toBe(12)
+  })
+
+  it('transfers projection state and closes the provider on model failure', async () => {
+    const harness = createHarness()
+    const receipt = harness.intake.acceptData(event())
+    harness.completions[0]!.reject(new Error('emulator failed'))
+
+    await expect(receipt).rejects.toThrow('emulator failed')
+    expect(harness.intake.getDebugSnapshot().projection.transferred).toBe(1)
+    expect(harness.dependencies.closeProvider).toHaveBeenCalledWith(1, 'model-admission-failed')
+  })
+
+  it('transfers a committed projection when desktop admission throws', async () => {
+    const projections: LegacySshProjectionSemantics[] = []
+    const harness = createHarness({
+      project: (_event, projection) => {
+        projections.push(projection)
+        if (projections.length === 1) {
+          throw new Error('send failed')
+        }
+      }
+    })
+    const receipt = harness.intake.acceptData(event({ data: '\x1b[?20', rawLength: 5 }))
+    harness.completions[0]!.resolve()
+
+    await expect(receipt).resolves.toMatchObject({ sequence: 5 })
+    expect(harness.intake.getDebugSnapshot().projection.transferred).toBe(1)
+
+    const next = harness.intake.acceptData(event({ data: '31h', rawLength: 3 }))
+    harness.completions[1]!.resolve()
+    await next
+    expect(projections[1]).toMatchObject({
+      identity: { displayStart: 0 },
+      beforeScanner: { tail: '', pendingSubscribe: false }
+    })
+  })
+
+  it('rolls back projection staging when model capture throws synchronously', async () => {
+    const project = vi.fn()
+    const harness = createHarness({
+      acceptModel: () => {
+        throw new Error('model reservation failed')
+      },
+      project
+    })
+
+    await expect(harness.intake.acceptData(event())).rejects.toThrow('model reservation failed')
+    expect(project).not.toHaveBeenCalled()
+    expect(harness.intake.getDebugSnapshot().projection).toMatchObject({
+      rolledBack: 1,
+      records: 0
+    })
+    expect(harness.dependencies.closeProvider).toHaveBeenCalledWith(1, 'model-admission-failed')
+  })
+
+  it('rejects stale provider generations without model capture', async () => {
+    const harness = createHarness()
+    const current = harness.intake.acceptData(event({ providerGeneration: 2 }))
+    harness.completions[0]!.resolve()
+    await current
+
+    await expect(harness.intake.acceptData(event({ providerGeneration: 1 }))).rejects.toThrow(
+      'ssh_output_stale_generation'
+    )
+    expect(harness.completions).toHaveLength(1)
+  })
+
+  it('keeps exit behind accepted model and projection work', async () => {
+    const harness = createHarness({}, { exitBarrierMs: 1000 })
+    const dataReceipt = harness.intake.acceptData(event())
+    const exitReceipt = harness.intake.acceptExit({
+      id: 'pty-1',
+      code: 0,
+      providerGeneration: 1,
+      ptyIncarnation: 'incarnation-1'
+    })
+    await Promise.resolve()
+    expect(harness.order).toEqual(['model:aaaa', 'project:aaaa'])
+
+    harness.completions[0]!.resolve()
+    await Promise.all([dataReceipt, exitReceipt])
+    expect(harness.order).toEqual(['model:aaaa', 'project:aaaa', 'exit'])
+  })
+
+  it('rejects late same-generation data after ordered exit cleanup', async () => {
+    const harness = createHarness({}, { exitBarrierMs: 1000 })
+    const first = harness.intake.acceptData(event())
+    harness.completions[0]!.resolve()
+    await first
+    await harness.intake.acceptExit({
+      id: 'pty-1',
+      code: 0,
+      providerGeneration: 1,
+      ptyIncarnation: 'incarnation-1'
+    })
+
+    await expect(harness.intake.acceptData(event())).rejects.toThrow('ssh_output_after_exit')
+    await expect(
+      harness.intake.acceptExit({
+        id: 'pty-1',
+        code: 0,
+        providerGeneration: 1,
+        ptyIncarnation: 'incarnation-1'
+      })
+    ).rejects.toThrow('ssh_output_duplicate_exit')
+
+    const next = harness.intake.acceptData(
+      event({ providerGeneration: 2, ptyIncarnation: 'incarnation-2' })
+    )
+    harness.completions[1]!.resolve()
+    await expect(next).resolves.toMatchObject({
+      projection: { identity: { ptyIncarnation: 'incarnation-2', displayStart: 0 } }
+    })
+  })
+
+  it('closes the provider when exit finalization fails', async () => {
+    const harness = createHarness({
+      finalizeExit: () => {
+        throw new Error('renderer exit send failed')
+      }
+    })
+
+    await expect(
+      harness.intake.acceptExit({
+        id: 'pty-1',
+        code: 0,
+        providerGeneration: 1,
+        ptyIncarnation: 'incarnation-1'
+      })
+    ).rejects.toThrow('renderer exit send failed')
+    expect(harness.dependencies.closeProvider).toHaveBeenCalledWith(1, 'pty-exit-finalize-failed')
+  })
+
+  it('cancels queued work and exit waiters on generation cleanup', async () => {
+    const harness = createHarness(
+      {},
+      {
+        perPtyHighSourceUnits: 8,
+        perPtyHighBytes: 2048,
+        globalHighSourceUnits: 8,
+        globalHighBytes: 2048
+      }
+    )
+    const running = harness.intake.acceptData(event())
+    const queued = harness.intake.acceptData(event({ data: 'bbbb' }))
+    const exit = harness.intake.acceptExit({
+      id: 'pty-1',
+      code: 0,
+      providerGeneration: 1,
+      ptyIncarnation: 'incarnation-1'
+    })
+    harness.intake.closeGeneration(1, 'provider-closed')
+
+    await expect(queued).rejects.toThrow('provider-closed')
+    await expect(exit).rejects.toThrow('provider-closed')
+    await expect(harness.intake.acceptData(event())).rejects.toThrow('ssh_output_stale_generation')
+    harness.completions[0]!.resolve()
+    await expect(running).rejects.toThrow('ssh_model_admission_generation_closed')
+    expect(harness.intake.getDebugSnapshot().exitBarriers).toBe(0)
+  })
+
+  it('resumes a paused provider during generation cleanup', async () => {
+    const harness = createHarness(
+      {},
+      {
+        perPtyHighSourceUnits: 4,
+        perPtyHighBytes: 1024,
+        globalHighSourceUnits: 4,
+        globalHighBytes: 1024,
+        pressureMaxFrames: 2,
+        pressureMaxBytes: 1024
+      }
+    )
+    const running = harness.intake.acceptData(event())
+    const pressured = harness.intake.acceptData(event({ data: 'bbbb' }))
+
+    harness.intake.closeGeneration(1, 'provider-closed')
+
+    await expect(pressured).rejects.toThrow('provider-closed')
+    expect(harness.dependencies.resumeProvider).toHaveBeenCalledWith(1, 'pty-1')
+    harness.completions[0]!.resolve()
+    await expect(running).rejects.toThrow('ssh_model_admission_generation_closed')
+  })
+})

src/main/ipc/ssh-pty-output-intake.ts

@@ -0,0 +1,315 @@
+import {
+  SshPtyLegacyProjectionLedger,
+  type LegacySshProjectionSemantics
+} from './ssh-pty-legacy-projection'
+import { SshPtyModelAdmission } from './ssh-pty-model-admission'
+import type {
+  SshPtyOutputDataEvent,
+  SshPtyOutputExitEvent,
+  SshPtyOutputIntakeDependencies,
+  SshPtyOutputIntakeOptions,
+  SshPtyOutputReceipt
+} from './ssh-pty-output-intake-contract'
+import {
+  outputIntakeError,
+  sshPtyGenerationKey,
+  validOutputLength,
+  type SshPtyExitBarrier
+} from './ssh-pty-output-intake-validation'
+
+export type {
+  SshPtyOutputDataEvent,
+  SshPtyOutputExitEvent,
+  SshPtyOutputIntakeDependencies,
+  SshPtyOutputIntakeOptions,
+  SshPtyOutputReceipt
+} from './ssh-pty-output-intake-contract'
+
+export class SshPtyOutputIntake {
+  private readonly projections = new SshPtyLegacyProjectionLedger()
+  private readonly admission: SshPtyModelAdmission
+  private readonly exitBarrierMs: number
+  private readonly latestGenerationByPty = new Map<string, number>()
+  private readonly incarnationByPty = new Map<string, string>()
+  private readonly sealedPtys = new Set<string>()
+  private readonly closedGenerations = new Set<number>()
+  private readonly exitBarriersByGeneration = new Map<number, Set<SshPtyExitBarrier>>()
+  private disposed = false
+
+  constructor(
+    private readonly dependencies: SshPtyOutputIntakeDependencies,
+    options: SshPtyOutputIntakeOptions = {}
+  ) {
+    this.exitBarrierMs = options.exitBarrierMs ?? 30_000
+    this.admission = new SshPtyModelAdmission({
+      ...options,
+      pauseProvider: (key) =>
+        this.dependencies.pauseProvider?.(key.providerGeneration, key.ptyId) ?? false,
+      resumeProvider: (key) =>
+        this.dependencies.resumeProvider?.(key.providerGeneration, key.ptyId),
+      closeProvider: (providerGeneration, reason) =>
+        this.dependencies.closeProvider?.(providerGeneration, reason)
+    })
+  }
+
+  acceptData(event: SshPtyOutputDataEvent): Promise<SshPtyOutputReceipt> {
+    try {
+      this.validateDataEvent(event)
+    } catch (error) {
+      return Promise.reject(error)
+    }
+    let projection: LegacySshProjectionSemantics | undefined
+    const receipt = this.admission.accept(
+      { ptyId: event.id, providerGeneration: event.providerGeneration },
+      event.data,
+      event.rawLength,
+      () => {
+        const expectedSequence = this.dependencies.getModelSequence(event.id) + event.rawLength
+        const reservation = this.projections.reserve({
+          ptyId: event.id,
+          providerGeneration: event.providerGeneration,
+          ptyIncarnation: event.ptyIncarnation,
+          data: event.data,
+          sequenceEnd: expectedSequence,
+          rawLength: event.rawLength,
+          transformed: event.transformed
+        })
+        let model: { sequence: number; completion: Promise<void> }
+        try {
+          model = this.dependencies.acceptModel(event)
+        } catch (error) {
+          this.projections.rollback(reservation)
+          throw error
+        }
+        projection = this.projections.commit(reservation)
+        try {
+          this.dependencies.project(event, projection)
+        } catch {
+          const id = projection.identity.projectionSemanticsId
+          if (!this.projections.transferUnpublished(id, 'projection-admission-failed')) {
+            this.projections.transfer([id], 'projection-admission-failed')
+          }
+        }
+        return model
+      }
+    )
+    return receipt.then(
+      (modelReceipt) => {
+        if (!projection) {
+          throw outputIntakeError('ssh_projection_receipt_missing')
+        }
+        return Object.freeze({ ...modelReceipt, projection })
+      },
+      (error) => {
+        if (projection) {
+          this.projections.transfer(
+            [projection.identity.projectionSemanticsId],
+            'model-admission-failed'
+          )
+        }
+        this.dependencies.closeProvider?.(event.providerGeneration, 'model-admission-failed')
+        throw error
+      }
+    )
+  }
+
+  async acceptExit(event: SshPtyOutputExitEvent): Promise<void> {
+    this.validateGeneration(event)
+    const sealKey = sshPtyGenerationKey(event.id, event.providerGeneration)
+    if (this.sealedPtys.has(sealKey)) {
+      throw outputIntakeError('ssh_output_duplicate_exit')
+    }
+    this.sealedPtys.add(sealKey)
+    await this.withExitDeadline(
+      event.providerGeneration,
+      this.admission.whenIdle({
+        ptyId: event.id,
+        providerGeneration: event.providerGeneration
+      })
+    )
+    this.validateGeneration(event)
+    try {
+      this.dependencies.finalizeExit(event)
+      this.projections.closePty(
+        event.id,
+        event.providerGeneration,
+        event.ptyIncarnation,
+        'pty-exit'
+      )
+    } catch (error) {
+      this.projections.closePty(
+        event.id,
+        event.providerGeneration,
+        event.ptyIncarnation,
+        'pty-exit-finalize-failed'
+      )
+      this.dependencies.closeProvider?.(event.providerGeneration, 'pty-exit-finalize-failed')
+      throw error
+    }
+  }
+
+  publishProjectionPrefix(
+    ids: readonly string[],
+    displayChars: number,
+    accountingChars: number
+  ): void {
+    this.projections.publishPrefix(ids, displayChars, accountingChars)
+  }
+
+  settleProjectionPrefix(ptyId: string, accountingChars: number): number {
+    return this.projections.settlePublishedPrefix(ptyId, accountingChars)
+  }
+
+  transferProjections(ids: readonly string[], reason: string): number {
+    return this.projections.transfer(ids, reason)
+  }
+
+  transferPtyProjections(ptyId: string, reason: string): number {
+    return this.projections.transferPty(ptyId, reason)
+  }
+
+  hasProjectionFromGeneration(ids: readonly string[], providerGeneration: number): boolean {
+    return ids.some(
+      (id) => this.projections.get(id)?.identity.providerGeneration === providerGeneration
+    )
+  }
+
+  closeGeneration(providerGeneration: number, reason: string): void {
+    this.closedGenerations.add(providerGeneration)
+    this.admission.closeGeneration(providerGeneration, reason)
+    this.dependencies.onGenerationClosed?.(providerGeneration, reason)
+    this.projections.closeGeneration(providerGeneration, reason)
+    const barriers = this.exitBarriersByGeneration.get(providerGeneration)
+    if (barriers) {
+      for (const barrier of barriers) {
+        clearTimeout(barrier.timer)
+        barrier.reject(outputIntakeError(reason))
+      }
+      this.exitBarriersByGeneration.delete(providerGeneration)
+    }
+    for (const [ptyId, generation] of this.latestGenerationByPty) {
+      if (generation === providerGeneration) {
+        this.latestGenerationByPty.delete(ptyId)
+        this.incarnationByPty.delete(ptyId)
+        this.sealedPtys.delete(sshPtyGenerationKey(ptyId, generation))
+      }
+    }
+    const generationPrefix = `${providerGeneration}\0`
+    for (const key of this.sealedPtys) {
+      if (key.startsWith(generationPrefix)) {
+        this.sealedPtys.delete(key)
+      }
+    }
+  }
+
+  dispose(): void {
+    if (this.disposed) {
+      return
+    }
+    this.disposed = true
+    for (const generation of new Set(this.latestGenerationByPty.values())) {
+      this.closeGeneration(generation, 'ssh_output_intake_disposed')
+    }
+    this.admission.dispose()
+  }
+
+  getDebugSnapshot() {
+    return {
+      model: this.admission.getDebugSnapshot(),
+      projection: this.projections.getDebugSnapshot(),
+      exitBarriers: Array.from(this.exitBarriersByGeneration.values()).reduce(
+        (total, barriers) => total + barriers.size,
+        0
+      )
+    }
+  }
+
+  private validateDataEvent(event: SshPtyOutputDataEvent): void {
+    if (this.disposed) {
+      throw outputIntakeError('ssh_output_intake_disposed')
+    }
+    if (
+      !event.id ||
+      !event.ptyIncarnation ||
+      !Number.isSafeInteger(event.providerGeneration) ||
+      event.providerGeneration <= 0 ||
+      !validOutputLength(event.rawLength) ||
+      (event.transformed && event.rawLength < 0)
+    ) {
+      throw outputIntakeError('ssh_output_invalid_event')
+    }
+    this.validateGeneration(event)
+    if (this.sealedPtys.has(sshPtyGenerationKey(event.id, event.providerGeneration))) {
+      throw outputIntakeError('ssh_output_after_exit')
+    }
+  }
+
+  private validateGeneration(event: {
+    id: string
+    providerGeneration: number
+    ptyIncarnation: string
+  }): void {
+    if (this.disposed) {
+      throw outputIntakeError('ssh_output_intake_disposed')
+    }
+    if (this.closedGenerations.has(event.providerGeneration)) {
+      throw outputIntakeError('ssh_output_stale_generation')
+    }
+    const generation = this.latestGenerationByPty.get(event.id)
+    if (generation !== undefined && event.providerGeneration < generation) {
+      throw outputIntakeError('ssh_output_stale_generation')
+    }
+    const incarnation = this.incarnationByPty.get(event.id)
+    if (
+      generation === event.providerGeneration &&
+      incarnation !== undefined &&
+      incarnation !== event.ptyIncarnation
+    ) {
+      throw outputIntakeError('ssh_output_stale_incarnation')
+    }
+    if (generation === undefined || event.providerGeneration > generation) {
+      this.latestGenerationByPty.set(event.id, event.providerGeneration)
+      this.incarnationByPty.set(event.id, event.ptyIncarnation)
+    }
+  }
+
+  private withExitDeadline(providerGeneration: number, promise: Promise<void>): Promise<void> {
+    return new Promise((resolve, reject) => {
+      const barrier: SshPtyExitBarrier = {
+        timer: setTimeout(() => {
+          this.removeExitBarrier(providerGeneration, barrier)
+          this.dependencies.closeProvider?.(providerGeneration, 'ssh-exit-barrier-timeout')
+          reject(outputIntakeError('ssh_exit_barrier_timeout'))
+        }, this.exitBarrierMs),
+        reject
+      }
+      barrier.timer.unref?.()
+      let barriers = this.exitBarriersByGeneration.get(providerGeneration)
+      if (!barriers) {
+        barriers = new Set()
+        this.exitBarriersByGeneration.set(providerGeneration, barriers)
+      }
+      barriers.add(barrier)
+      void promise.then(
+        () => {
+          clearTimeout(barrier.timer)
+          this.removeExitBarrier(providerGeneration, barrier)
+          resolve()
+        },
+        (error) => {
+          clearTimeout(barrier.timer)
+          this.removeExitBarrier(providerGeneration, barrier)
+          reject(error)
+        }
+      )
+    })
+  }
+
+  private removeExitBarrier(providerGeneration: number, barrier: SshPtyExitBarrier): void {
+    const barriers = this.exitBarriersByGeneration.get(providerGeneration)
+    barriers?.delete(barrier)
+    if (barriers?.size === 0) {
+      this.exitBarriersByGeneration.delete(providerGeneration)
+    }
+  }
+}

src/main/providers/ssh-pty-notification-routing.test.ts

@@ -23,7 +23,9 @@ function createSubscription() {
     replayListeners: replayListeners as never,
     exitListeners: exitListeners as never,
     livePtyIds,
-    recordExit
+    recordExit,
+    providerGeneration: 7,
+    resolvePtyIncarnation: (id) => `incarnation:${id}`
   })
 
   const handler = mux.onNotification.mock.calls[0]?.[0] as (
@@ -70,6 +72,8 @@ describe('subscribeSshPtyNotifications', () => {
     expect(onData).toHaveBeenCalledWith({
       id: 'ssh:conn@@pty-1',
       data: 'hello',
+      providerGeneration: 7,
+      ptyIncarnation: 'incarnation:pty-1',
       sequenceChars: 5,
       seq: 9
     })
@@ -92,6 +96,8 @@ describe('subscribeSshPtyNotifications', () => {
     expect(onExit).toHaveBeenCalledWith({
       id: 'ssh:conn@@pty-1',
       code: 0,
+      providerGeneration: 7,
+      ptyIncarnation: 'incarnation:pty-1',
       incarnationId: 'incarnation-1'
     })
   })

src/main/providers/ssh-pty-notification-routing.ts

@@ -16,6 +16,8 @@ export function subscribeSshPtyNotifications(args: {
   exitListeners: Set<SshPtyExitCallback>
   livePtyIds: Set<string>
   recordExit: (relayPtyId: string, incarnationId: unknown) => void
+  providerGeneration: number
+  resolvePtyIncarnation: (relayPtyId: string, incarnationId?: unknown) => string
 }): () => void {
   return args.mux.onNotification((method, params) => {
     // Why: mux delivers every method to generic handlers; non-PTY payloads
@@ -30,12 +32,15 @@ export function subscribeSshPtyNotifications(args: {
     const relayPtyId = params.id
     const id = args.toAppPtyId(relayPtyId)
     if (method === 'pty.exit') {
+      const ptyIncarnation = args.resolvePtyIncarnation(relayPtyId, params.incarnationId)
       args.recordExit(relayPtyId, params.incarnationId)
       args.livePtyIds.delete(id)
       for (const listener of args.exitListeners) {
         listener({
           id,
           code: params.code as number,
+          providerGeneration: args.providerGeneration,
+          ptyIncarnation,
           ...(isPtyIncarnationId(params.incarnationId)
             ? { incarnationId: params.incarnationId }
             : {})
@@ -54,6 +59,8 @@ export function subscribeSshPtyNotifications(args: {
       listener({
         id,
         data: params.data as string,
+        providerGeneration: args.providerGeneration,
+        ptyIncarnation: args.resolvePtyIncarnation(relayPtyId, params.incarnationId),
         ...(typeof params.rawLength === 'number' ? { sequenceChars: params.rawLength } : {}),
         ...(params.transformed === true ? { transformed: true } : {}),
         ...(typeof params.seq === 'number' ? { seq: params.seq } : {})

src/main/providers/ssh-pty-provider-contract.ts

@@ -11,6 +11,8 @@ export type RemoteCliBridgeEnv = {
 export type SshPtyDataCallback = (payload: {
   id: string
   data: string
+  providerGeneration: number
+  ptyIncarnation: string
   sequenceChars?: number
   transformed?: boolean
   seq?: number
@@ -19,5 +21,13 @@ export type SshPtyReplayCallback = (payload: { id: string; data: string }) => vo
 export type SshPtyExitCallback = (payload: {
   id: string
   code: number
+  providerGeneration: number
+  ptyIncarnation: string
   incarnationId?: PtyIncarnationId
 }) => void
+
+export type SshPtyDeliveryPauseAdapter = (args: {
+  id: string
+  providerGeneration: number
+  paused: boolean
+}) => void

src/main/providers/ssh-pty-provider-exit-race.test.ts

@@ -32,7 +32,9 @@ it('rejects a fresh SSH PTY whose exit shares the spawn response batch', async (
   expect(exitListener).toHaveBeenCalledWith({
     id: 'ssh:conn-1@@pty-raced',
     code: 0,
-    incarnationId: 'incarnation-raced'
+    incarnationId: 'incarnation-raced',
+    providerGeneration: expect.any(Number),
+    ptyIncarnation: 'incarnation-raced'
   })
   mux.request.mockResolvedValue({ id: 'pty-next', incarnationId: 'incarnation-next' })
   await expect(provider.spawn({ cols: 80, rows: 24 })).resolves.toMatchObject({

src/main/providers/ssh-pty-provider-output-state.ts

@@ -0,0 +1,132 @@
+import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer'
+import type {
+  SshPtyDataCallback,
+  SshPtyDeliveryPauseAdapter,
+  SshPtyExitCallback,
+  SshPtyReplayCallback
+} from './ssh-pty-provider-contract'
+import { subscribeSshPtyNotifications } from './ssh-pty-notification-routing'
+
+export class SshPtyProviderOutputState {
+  private readonly dataListeners = new Set<SshPtyDataCallback>()
+  private readonly replayListeners = new Set<SshPtyReplayCallback>()
+  private readonly exitListeners = new Set<SshPtyExitCallback>()
+  private readonly incarnationByRelayPtyId = new Map<string, string>()
+  private readonly pausedRelayPtyIds = new Set<string>()
+  private deliveryPauseAdapter: SshPtyDeliveryPauseAdapter | null = null
+  private legacyIncarnationSerial = 1
+  private unsubscribeNotifications: (() => void) | null
+
+  constructor(
+    private readonly providerGeneration: number,
+    args: {
+      mux: SshChannelMultiplexer
+      toAppPtyId: (id: string) => string
+      livePtyIds: Set<string>
+      recordExit: (relayPtyId: string, incarnationId: unknown) => void
+    }
+  ) {
+    this.unsubscribeNotifications = subscribeSshPtyNotifications({
+      ...args,
+      dataListeners: this.dataListeners,
+      replayListeners: this.replayListeners,
+      exitListeners: this.exitListeners,
+      providerGeneration,
+      resolvePtyIncarnation: (relayPtyId, incarnationId) =>
+        this.resolvePtyIncarnation(relayPtyId, incarnationId),
+      recordExit: (relayPtyId, incarnationId) => {
+        args.recordExit(relayPtyId, incarnationId)
+        this.incarnationByRelayPtyId.delete(relayPtyId)
+        this.pausedRelayPtyIds.delete(relayPtyId)
+      }
+    })
+  }
+
+  dispose(): void {
+    this.resumePausedDeliveries()
+    this.unsubscribeNotifications?.()
+    this.unsubscribeNotifications = null
+    this.dataListeners.clear()
+    this.replayListeners.clear()
+    this.exitListeners.clear()
+    this.incarnationByRelayPtyId.clear()
+    this.deliveryPauseAdapter = null
+  }
+
+  onData(callback: SshPtyDataCallback): () => void {
+    this.dataListeners.add(callback)
+    return () => this.dataListeners.delete(callback)
+  }
+
+  onReplay(callback: SshPtyReplayCallback): () => void {
+    this.replayListeners.add(callback)
+    return () => this.replayListeners.delete(callback)
+  }
+
+  onExit(callback: SshPtyExitCallback): () => void {
+    this.exitListeners.add(callback)
+    return () => this.exitListeners.delete(callback)
+  }
+
+  setDeliveryPauseAdapter(adapter: SshPtyDeliveryPauseAdapter | null): void {
+    if (adapter !== this.deliveryPauseAdapter) {
+      this.resumePausedDeliveries()
+    }
+    this.deliveryPauseAdapter = adapter
+  }
+
+  hasDeliveryPauseAdapter(): boolean {
+    return this.deliveryPauseAdapter !== null
+  }
+
+  pause(id: string): void {
+    if (!this.deliveryPauseAdapter || this.pausedRelayPtyIds.has(id)) {
+      return
+    }
+    this.pausedRelayPtyIds.add(id)
+    this.deliveryPauseAdapter({ id, providerGeneration: this.providerGeneration, paused: true })
+  }
+
+  resume(id: string): void {
+    if (!this.deliveryPauseAdapter || !this.pausedRelayPtyIds.delete(id)) {
+      return
+    }
+    this.deliveryPauseAdapter({ id, providerGeneration: this.providerGeneration, paused: false })
+  }
+
+  rememberPtyIncarnation(relayPtyId: string, incarnationId: unknown): void {
+    if (
+      !this.incarnationByRelayPtyId.has(relayPtyId) &&
+      typeof incarnationId === 'string' &&
+      incarnationId.length > 0
+    ) {
+      this.incarnationByRelayPtyId.set(relayPtyId, incarnationId)
+    }
+  }
+
+  private resolvePtyIncarnation(relayPtyId: string, incarnationId: unknown): string {
+    this.rememberPtyIncarnation(relayPtyId, incarnationId)
+    let resolved = this.incarnationByRelayPtyId.get(relayPtyId)
+    if (!resolved) {
+      resolved = `legacy:${this.providerGeneration}:${this.legacyIncarnationSerial++}:${relayPtyId}`
+      this.incarnationByRelayPtyId.set(relayPtyId, resolved)
+    }
+    return resolved
+  }
+
+  private resumePausedDeliveries(): void {
+    const adapter = this.deliveryPauseAdapter
+    if (!adapter) {
+      this.pausedRelayPtyIds.clear()
+      return
+    }
+    for (const id of this.pausedRelayPtyIds) {
+      try {
+        adapter({ id, providerGeneration: this.providerGeneration, paused: false })
+      } catch {
+        /* Generation close is the fallback cleanup proof. */
+      }
+    }
+    this.pausedRelayPtyIds.clear()
+  }
+}

src/main/providers/ssh-pty-provider-process-events.test.ts

@@ -135,10 +135,17 @@ describe('SshPtyProvider process listings and events', () => {
     notify('pty.replay', { id: 'pty-1', data: 'buffered output' })
     notify('pty.exit', { id: 'pty-1', code: 0, incarnationId: 'incarnation-1' })
 
-    expect(dataHandler).toHaveBeenNthCalledWith(1, { id: scopedPty1, data: 'output' })
+    expect(dataHandler).toHaveBeenNthCalledWith(1, {
+      id: scopedPty1,
+      data: 'output',
+      providerGeneration: 1,
+      ptyIncarnation: 'legacy:1:1:pty-1'
+    })
     expect(dataHandler).toHaveBeenNthCalledWith(2, {
       id: scopedPty1,
       data: '',
+      providerGeneration: 1,
+      ptyIncarnation: 'legacy:1:1:pty-1',
       sequenceChars: 9,
       seq: 9,
       transformed: true
@@ -147,10 +154,28 @@ describe('SshPtyProvider process listings and events', () => {
     expect(exitHandler).toHaveBeenCalledWith({
       id: scopedPty1,
       code: 0,
+      providerGeneration: 1,
+      ptyIncarnation: 'legacy:1:1:pty-1',
       incarnationId: 'incarnation-1'
     })
   })
 
+  it('keeps fallback admission identity stable when spawn metadata arrives later', async () => {
+    const dataHandler = vi.fn()
+    provider.onData(dataHandler)
+    const notify = mux.onNotification.mock.calls[0][0]
+
+    notify('pty.data', { id: 'pty-1', data: 'before-response' })
+    mux.request.mockResolvedValue({ id: 'pty-1', incarnationId: 'incarnation-1' })
+    await provider.spawn({ cols: 80, rows: 24 })
+    notify('pty.data', { id: 'pty-1', data: 'after-response' })
+
+    expect(dataHandler.mock.calls.map(([payload]) => payload.ptyIncarnation)).toEqual([
+      'legacy:1:1:pty-1',
+      'legacy:1:1:pty-1'
+    ])
+  })
+
   it('supports listener removal, fanout, and connection namespaces', () => {
     const removed = vi.fn()
     const first = vi.fn()
@@ -170,6 +195,29 @@ describe('SshPtyProvider process listings and events', () => {
     const other = vi.fn()
     otherProvider.onData(other)
     otherMux.onNotification.mock.calls[0][0]('pty.data', { id: 'pty-1', data: 'second' })
-    expect(other).toHaveBeenCalledWith({ id: 'ssh:conn-2@@pty-1', data: 'second' })
+    expect(other).toHaveBeenCalledWith({
+      id: 'ssh:conn-2@@pty-1',
+      data: 'second',
+      providerGeneration: 1,
+      ptyIncarnation: 'legacy:1:1:pty-1'
+    })
+  })
+
+  it('scopes pause delivery adapters and resumes them during cleanup', () => {
+    const adapter = vi.fn()
+    provider.setPtyDeliveryPauseAdapter(adapter)
+
+    provider.pauseProducer(scopedPty1)
+    provider.pauseProducer(scopedPty1)
+    provider.resumeProducer(scopedPty1)
+    provider.pauseProducer(scopedPty1)
+    provider.dispose()
+
+    expect(adapter.mock.calls).toEqual([
+      [{ id: 'pty-1', providerGeneration: 1, paused: true }],
+      [{ id: 'pty-1', providerGeneration: 1, paused: false }],
+      [{ id: 'pty-1', providerGeneration: 1, paused: true }],
+      [{ id: 'pty-1', providerGeneration: 1, paused: false }]
+    ])
   })
 })

src/main/providers/ssh-pty-provider.ts

@@ -5,10 +5,11 @@ import { createSshPtyAppliedSizeReader } from './ssh-pty-applied-size'
 import type {
   RemoteCliBridgeEnv,
   SshPtyDataCallback,
+  SshPtyDeliveryPauseAdapter,
   SshPtyExitCallback,
   SshPtyReplayCallback
 } from './ssh-pty-provider-contract'
-import { subscribeSshPtyNotifications } from './ssh-pty-notification-routing'
+import { SshPtyProviderOutputState } from './ssh-pty-provider-output-state'
 import { validateClaimedSshSpawn } from './ssh-agent-session-claim-validation'
 import {
   assertSshAgentSessionCreateResult,
@@ -34,46 +35,35 @@ function relayTimeoutOptions(deadlineMs: number | undefined): { timeoutMs: numbe
 export class SshPtyProvider implements IPtyProvider {
   private mux: SshChannelMultiplexer
   private connectionId: string
-  private dataListeners = new Set<SshPtyDataCallback>()
-  private replayListeners = new Set<SshPtyReplayCallback>()
-  private exitListeners = new Set<SshPtyExitCallback>()
   private livePtyIds = new Set<string>()
-  // Why: stale notification callbacks must not outlive a disconnected provider.
-  private unsubscribeNotifications: (() => void) | null = null
   readonly getAppliedSize: NonNullable<IPtyProvider['getAppliedSize']>
   private readonly agentSessionCapabilities: SshAgentSessionCapabilities
   private spawnExitRaces = new SshPtySpawnExitRaceTracker()
+  private readonly outputState: SshPtyProviderOutputState
 
   constructor(
     connectionId: string,
     mux: SshChannelMultiplexer,
-    private readonly remoteCliBridgeEnv?: RemoteCliBridgeEnv
+    private readonly remoteCliBridgeEnv?: RemoteCliBridgeEnv,
+    readonly providerGeneration = 1
   ) {
     this.connectionId = connectionId
     this.mux = mux
     this.agentSessionCapabilities = new SshAgentSessionCapabilities(mux)
     this.getAppliedSize = createSshPtyAppliedSizeReader(mux, connectionId)
 
-    this.unsubscribeNotifications = subscribeSshPtyNotifications({
+    this.outputState = new SshPtyProviderOutputState(providerGeneration, {
       mux,
       toAppPtyId: (id) => this.toAppPtyId(id),
-      dataListeners: this.dataListeners,
-      replayListeners: this.replayListeners,
-      exitListeners: this.exitListeners,
       livePtyIds: this.livePtyIds,
-      recordExit: (relayPtyId, incarnationId) =>
+      recordExit: (relayPtyId, incarnationId) => {
         this.spawnExitRaces.recordExit(relayPtyId, incarnationId)
+      }
     })
   }
 
   dispose(): void {
-    if (this.unsubscribeNotifications) {
-      this.unsubscribeNotifications()
-      this.unsubscribeNotifications = null
-    }
-    this.dataListeners.clear()
-    this.replayListeners.clear()
-    this.exitListeners.clear()
+    this.outputState.dispose()
     this.livePtyIds.clear()
   }
 
@@ -161,6 +151,7 @@ export class SshPtyProvider implements IPtyProvider {
         }
       }
       const id = this.toAppPtyId(spawnResult.id)
+      this.outputState.rememberPtyIncarnation(spawnResult.id, spawnResult.incarnationId)
       this.livePtyIds.add(id)
       return {
         ...spawnResult,
@@ -209,7 +200,7 @@ export class SshPtyProvider implements IPtyProvider {
     // be filtered before they reach the renderer. The expected identity lets the
     // relay reject a cross-generation id collision instead of reattaching this
     // lease to a different pane's freshly spawned PTY.
-    return parseSshPtyAttachResult(
+    const result = parseSshPtyAttachResult(
       await this.mux.request('pty.attach', {
         id: this.toRelayPtyId(id),
         suppressReplayNotification: true,
@@ -217,6 +208,8 @@ export class SshPtyProvider implements IPtyProvider {
         ...(expected?.tabId ? { expectedTabId: expected.tabId } : {})
       })
     )
+    this.outputState.rememberPtyIncarnation(this.toRelayPtyId(id), result.incarnationId)
+    return result
   }
 
   write(id: string, data: string): void {
@@ -308,6 +301,8 @@ export class SshPtyProvider implements IPtyProvider {
     const processes = mapSshPtyProcessList(result as PtyProcessInfo[], (id) => this.toAppPtyId(id))
     for (const process of processes) {
       this.livePtyIds.add(process.id)
+      const relayPtyId = this.toRelayPtyId(process.id)
+      this.outputState.rememberPtyIncarnation(relayPtyId, process.incarnationId)
     }
     return processes
   }
@@ -326,18 +321,28 @@ export class SshPtyProvider implements IPtyProvider {
     return result as { name: string; path: string }[]
   }
 
-  onData(callback: SshPtyDataCallback): () => void {
-    this.dataListeners.add(callback)
-    return () => this.dataListeners.delete(callback)
+  onData = (callback: SshPtyDataCallback): (() => void) => this.outputState.onData(callback)
+  onReplay = (callback: SshPtyReplayCallback): (() => void) => this.outputState.onReplay(callback)
+  onExit = (callback: SshPtyExitCallback): (() => void) => this.outputState.onExit(callback)
+
+  setPtyDeliveryPauseAdapter(adapter: SshPtyDeliveryPauseAdapter | null): void {
+    this.outputState.setDeliveryPauseAdapter(adapter)
   }
 
-  onReplay(callback: SshPtyReplayCallback): () => void {
-    this.replayListeners.add(callback)
-    return () => this.replayListeners.delete(callback)
+  hasPtyDeliveryPauseAdapter(): boolean {
+    return this.outputState.hasDeliveryPauseAdapter()
   }
 
-  onExit(callback: SshPtyExitCallback): () => void {
-    this.exitListeners.add(callback)
-    return () => this.exitListeners.delete(callback)
+  pauseProducer(id: string): void {
+    this.outputState.pause(this.toRelayPtyId(id))
+  }
+
+  resumeProducer(id: string): void {
+    this.outputState.resume(this.toRelayPtyId(id))
+  }
+
+  closeOutputIntake(reason: string): void {
+    this.mux.dispose('connection_lost')
+    console.error('[ssh-pty-provider] closed after bounded output intake failure', { reason })
   }
 }

src/main/runtime/orca-runtime.ts

@@ -1417,6 +1417,11 @@ type RuntimeHeadlessTerminal = {
   writeChain: Promise<void>
 }
 
+export type RuntimePtyDataAdmission = Readonly<{
+  sequence: number
+  completion: Promise<void>
+}>
+
 type RuntimeVisibleTerminalState = {
   lines: string[]
   isAlternateScreen: boolean
@@ -7714,12 +7719,30 @@ export class OrcaRuntimeService {
    * Handles incoming data from a PTY process, running agent detection,
    * updating terminal tail buffers, and triggering foreground agent refreshes.
    */
-  onPtyData(
+  acceptPtyDataBounded(
     ptyId: string,
     data: string,
     at: number,
     sequenceChars = data.length,
     transformed = false
+  ): RuntimePtyDataAdmission {
+    let completion: Promise<void> | null = null
+    const sequence = this.onPtyData(ptyId, data, at, sequenceChars, transformed, (receipt) => {
+      completion = receipt
+    })
+    if (!completion) {
+      throw new Error('PTY model admission receipt was not captured')
+    }
+    return Object.freeze({ sequence, completion })
+  }
+
+  onPtyData(
+    ptyId: string,
+    data: string,
+    at: number,
+    sequenceChars = data.length,
+    transformed = false,
+    captureModelReceipt?: (completion: Promise<void>) => void
   ): number {
     const outputSequence = (this.ptyOutputSequenceById.get(ptyId) ?? 0) + sequenceChars
     this.ptyOutputSequenceById.set(ptyId, outputSequence)
@@ -7758,7 +7781,13 @@ export class OrcaRuntimeService {
     // applyTrackedPtyTitle) in byte order, superseding main's inline
     // extractLastOscTitleForPty block (#7880/#7852 title/status semantics are
     // preserved via the tracker + detectAgentStatusFromTitle path).
-    this.trackHeadlessTerminalData(ptyId, data, outputSequence, forwardQueryReplies)
+    const modelCompletion = this.trackHeadlessTerminalData(
+      ptyId,
+      data,
+      outputSequence,
+      forwardQueryReplies
+    )
+    captureModelReceipt?.(modelCompletion)
 
     const pty = this.getOrCreatePtyWorktreeRecord(ptyId)
     const ptyTailBefore = pty
@@ -9278,19 +9307,17 @@ export class OrcaRuntimeService {
     data: string,
     outputSequence: number,
     forwardQueryReplies = false
-  ): void {
+  ): Promise<void> {
     const state = this.getOrCreateHeadlessTerminal(ptyId)
-    state.writeChain = state.writeChain
-      .then(async () => {
-        // Why: the ingestion-time ownership decision is closed over this
-        // chain link; async scheduling cannot retroactively change it.
-        await state.emulator.write(data, { forwardQueryReplies })
-        state.outputSequence = outputSequence
-      })
-      .catch(() => {
-        // Best-effort state tracking; live streaming must continue even if
-        // xterm rejects a malformed or raced write during shutdown.
-      })
+    const completion = state.writeChain.then(async () => {
+      // Why: the ingestion-time ownership decision is closed over this
+      // chain link; async scheduling cannot retroactively change it.
+      await state.emulator.write(data, { forwardQueryReplies })
+      state.outputSequence = outputSequence
+    })
+    // Legacy callers remain best-effort; bounded SSH admission observes the raw receipt.
+    state.writeChain = completion.catch(() => {})
+    return completion
   }
 
   /** Shared factory for the per-PTY runtime emulators (seed, hydration, and

src/main/ssh/ssh-relay-session-data-delivery.test.ts

@@ -2,9 +2,18 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { SshRelaySession } from './ssh-relay-session'
 import { createMockDeps, mockDeploySuccess } from './ssh-relay-session-test-fixtures'
 
-const { muxRequestMock } = vi.hoisted(() => ({ muxRequestMock: vi.fn() }))
+const { acceptOutputDataMock, muxRequestMock } = vi.hoisted(() => ({
+  acceptOutputDataMock: vi.fn().mockResolvedValue(undefined),
+  muxRequestMock: vi.fn()
+}))
 
 vi.mock('./ssh-relay-deploy', () => ({ deployAndLaunchRelay: vi.fn() }))
+vi.mock('../ipc/ssh-pty-output-intake-registry', () => ({
+  acceptSshPtyOutputData: acceptOutputDataMock,
+  acceptSshPtyOutputExit: vi.fn().mockResolvedValue(undefined),
+  allocateSshPtyProviderGeneration: vi.fn(() => 23),
+  closeSshPtyOutputGeneration: vi.fn()
+}))
 
 vi.mock('./ssh-channel-multiplexer', () => ({
   SshChannelMultiplexer: class MockSshChannelMultiplexer {
@@ -95,18 +104,28 @@ describe('SshRelaySession data delivery', () => {
       data: string
       sequenceChars?: number
       transformed?: boolean
+      providerGeneration: number
+      ptyIncarnation: string
     }) => void
 
-    onData({ id: 'ssh-pty-1', data: '', sequenceChars: 9, transformed: true })
-
-    expect(runtime.onPtyData).toHaveBeenCalledWith('ssh-pty-1', '', expect.any(Number), 9, true)
-    expect(mockWindow.webContents.send).toHaveBeenCalledWith('pty:data', {
+    onData({
       id: 'ssh-pty-1',
       data: '',
       sequenceChars: 9,
       transformed: true,
-      seq: 17,
-      rawLength: 9
+      providerGeneration: 23,
+      ptyIncarnation: 'incarnation-1'
     })
+
+    expect(acceptOutputDataMock).toHaveBeenCalledWith({
+      id: 'ssh-pty-1',
+      data: '',
+      providerGeneration: 23,
+      ptyIncarnation: 'incarnation-1',
+      rawLength: 9,
+      transformed: true
+    })
+    expect(runtime.onPtyData).not.toHaveBeenCalled()
+    expect(mockWindow.webContents.send).not.toHaveBeenCalledWith('pty:data', expect.anything())
   })
 })

src/main/ssh/ssh-relay-session-incarnation.test.ts

@@ -2,9 +2,18 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { SshRelaySession } from './ssh-relay-session'
 import { createMockDeps, mockDeploySuccess } from './ssh-relay-session-test-fixtures'
 
-const { muxRequestMock } = vi.hoisted(() => ({ muxRequestMock: vi.fn() }))
+const { acceptOutputExitMock, muxRequestMock } = vi.hoisted(() => ({
+  acceptOutputExitMock: vi.fn().mockResolvedValue(undefined),
+  muxRequestMock: vi.fn()
+}))
 
 vi.mock('./ssh-relay-deploy', () => ({ deployAndLaunchRelay: vi.fn() }))
+vi.mock('../ipc/ssh-pty-output-intake-registry', () => ({
+  acceptSshPtyOutputData: vi.fn().mockResolvedValue(undefined),
+  acceptSshPtyOutputExit: acceptOutputExitMock,
+  allocateSshPtyProviderGeneration: vi.fn(() => 31),
+  closeSshPtyOutputGeneration: vi.fn()
+}))
 vi.mock('./ssh-relay-deploy-helpers', () => ({ execCommand: vi.fn().mockResolvedValue('') }))
 vi.mock('./ssh-channel-multiplexer', () => ({
   SshChannelMultiplexer: class MockSshChannelMultiplexer {
@@ -90,22 +99,41 @@ describe('SSH relay PTY incarnation exits', () => {
       id: string
       code: number
       incarnationId: string
+      providerGeneration: number
+      ptyIncarnation: string
     }) => void
     vi.mocked(isCurrentPtyExit).mockReturnValueOnce(false)
 
-    onExit({ id: 'ssh:target-1@@pty-reused', code: 0, incarnationId: 'old-incarnation' })
+    onExit({
+      id: 'ssh:target-1@@pty-reused',
+      code: 0,
+      incarnationId: 'old-incarnation',
+      providerGeneration: 31,
+      ptyIncarnation: 'old-incarnation'
+    })
 
     expect(clearProviderPtyState).not.toHaveBeenCalled()
     expect(deletePtyOwnership).not.toHaveBeenCalled()
     expect(mockStore.markSshRemotePtyLease).not.toHaveBeenCalled()
+    expect(acceptOutputExitMock).not.toHaveBeenCalled()
     expect(runtime.onPtyExit).not.toHaveBeenCalled()
     expect(mockWindow.webContents.send).not.toHaveBeenCalledWith('pty:exit', expect.anything())
 
-    onExit({ id: 'ssh:target-1@@pty-reused', code: 7, incarnationId: 'current-incarnation' })
-    expect(runtime.onPtyExit).toHaveBeenCalledWith(
-      'ssh:target-1@@pty-reused',
-      7,
-      'current-incarnation'
+    onExit({
+      id: 'ssh:target-1@@pty-reused',
+      code: 7,
+      incarnationId: 'current-incarnation',
+      providerGeneration: 31,
+      ptyIncarnation: 'current-incarnation'
+    })
+    await vi.waitFor(() =>
+      expect(acceptOutputExitMock).toHaveBeenCalledWith({
+        id: 'ssh:target-1@@pty-reused',
+        code: 7,
+        providerGeneration: 31,
+        ptyIncarnation: 'current-incarnation'
+      })
     )
+    expect(runtime.onPtyExit).not.toHaveBeenCalled()
   })
 })

src/main/ssh/ssh-relay-session-reconnect-incarnation.test.ts

@@ -2,9 +2,18 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { SshRelaySession } from './ssh-relay-session'
 import { createMockDeps, mockDeploySuccess } from './ssh-relay-session-test-fixtures'
 
-const { muxRequestMock } = vi.hoisted(() => ({ muxRequestMock: vi.fn() }))
+const { acceptOutputExitMock, muxRequestMock } = vi.hoisted(() => ({
+  acceptOutputExitMock: vi.fn().mockResolvedValue(undefined),
+  muxRequestMock: vi.fn()
+}))
 
 vi.mock('./ssh-relay-deploy', () => ({ deployAndLaunchRelay: vi.fn() }))
+vi.mock('../ipc/ssh-pty-output-intake-registry', () => ({
+  acceptSshPtyOutputData: vi.fn().mockResolvedValue(undefined),
+  acceptSshPtyOutputExit: acceptOutputExitMock,
+  allocateSshPtyProviderGeneration: vi.fn(() => 17),
+  closeSshPtyOutputGeneration: vi.fn()
+}))
 vi.mock('./ssh-relay-deploy-helpers', () => ({ execCommand: vi.fn().mockResolvedValue('') }))
 vi.mock('./ssh-channel-multiplexer', () => ({
   SshChannelMultiplexer: class MockSshChannelMultiplexer {
@@ -85,14 +94,26 @@ function detachedLease() {
   }
 }
 
-function emitExitDuringAttach(payload: { id: string; code: number; incarnationId?: string }): void {
+function emitExitDuringAttach(payload: {
+  id: string
+  code: number
+  incarnationId?: string
+  providerGeneration?: number
+  ptyIncarnation?: string
+}): void {
   const registeredProvider = vi.mocked(registerSshPtyProvider).mock.calls[0]?.[1] as unknown as {
     onExit: ReturnType<typeof vi.fn>
   }
   const exitHandler = registeredProvider.onExit.mock.calls[0]?.[0] as
     | ((exit: typeof payload) => void)
     | undefined
-  queueMicrotask(() => exitHandler?.(payload))
+  queueMicrotask(() =>
+    exitHandler?.({
+      providerGeneration: 17,
+      ptyIncarnation: payload.incarnationId ?? `legacy:${payload.id}`,
+      ...payload
+    })
+  )
 }
 
 describe('SshRelaySession reconnect incarnation ordering', () => {
@@ -175,7 +196,13 @@ describe('SshRelaySession reconnect incarnation ordering', () => {
 
     await session.establish(mockConn)
 
-    expect(runtime.onPtyExit).toHaveBeenCalledWith(APP_PTY_ID, 0, incarnationId)
+    expect(acceptOutputExitMock).toHaveBeenCalledWith({
+      id: APP_PTY_ID,
+      code: 0,
+      providerGeneration: 17,
+      ptyIncarnation: incarnationId
+    })
+    expect(runtime.onPtyExit).not.toHaveBeenCalled()
     expect(runtime.acceptPtyIncarnationForExit).toHaveBeenCalledWith(APP_PTY_ID, incarnationId)
     expect(runtime.registerPty).not.toHaveBeenCalled()
     expect(restorePtyIncarnation).toHaveBeenCalledWith(APP_PTY_ID, incarnationId)

src/main/ssh/ssh-relay-session.test.ts

@@ -8,12 +8,22 @@ import {
 import { SSH_RELAY_CONFIGURE_GRACE_TIME_METHOD } from '../../shared/ssh-types'
 import { createMockDeps, mockDeploySuccess } from './ssh-relay-session-test-fixtures'
 
-const { muxRequestMock } = vi.hoisted(() => ({ muxRequestMock: vi.fn() }))
+const { acceptOutputDataMock, muxRequestMock } = vi.hoisted(() => ({
+  acceptOutputDataMock: vi.fn().mockResolvedValue(undefined),
+  muxRequestMock: vi.fn()
+}))
 
 vi.mock('./ssh-relay-deploy', () => ({
   deployAndLaunchRelay: vi.fn()
 }))
 
+vi.mock('../ipc/ssh-pty-output-intake-registry', () => ({
+  acceptSshPtyOutputData: acceptOutputDataMock,
+  acceptSshPtyOutputExit: vi.fn().mockResolvedValue(undefined),
+  allocateSshPtyProviderGeneration: vi.fn(() => 41),
+  closeSshPtyOutputGeneration: vi.fn()
+}))
+
 vi.mock('./ssh-relay-deploy-helpers', () => ({
   execCommand: vi.fn().mockResolvedValue('')
 }))
@@ -86,12 +96,6 @@ vi.mock('../providers/ssh-git-dispatch', () => ({
 }))
 
 const { deployAndLaunchRelay } = await import('./ssh-relay-deploy')
-// Why: the hidden-delivery gate module is intentionally real (pure state, no
-// electron deps) so the SSH parity tests exercise the same gate main uses.
-const { markHiddenRendererPty, setRendererPtyDeliveryInterest } =
-  await import('../ipc/pty-hidden-delivery-gate')
-const { _resetHiddenRendererPtyDeliveryGateForTest } =
-  await import('../ipc/pty-hidden-delivery-gate')
 const { execCommand } = await import('./ssh-relay-deploy-helpers')
 const { getRemoteHostPlatform } = await import('./ssh-remote-platform')
 const {
@@ -115,15 +119,11 @@ describe('SshRelaySession', () => {
     muxRequestMock.mockResolvedValue([])
     mockDeploySuccess()
     vi.mocked(getPtyIdsForConnection).mockReturnValue([])
-    _resetHiddenRendererPtyDeliveryGateForTest()
   })
 
-  it('drops hidden-gated PTY data after runtime ingestion with one restore marker', async () => {
+  it('hands each PTY data event exactly once to the bounded main intake', async () => {
     const { mockConn, mockStore, mockPortForward, getMainWindow, mockWindow } = createMockDeps()
-    const runtime = {
-      onPtyData: vi.fn(() => 99),
-      onPtyExit: vi.fn()
-    }
+    const runtime = { onPtyData: vi.fn(), onPtyExit: vi.fn() }
     const session = new SshRelaySession(
       'target-1',
       getMainWindow,
@@ -138,53 +138,31 @@ describe('SshRelaySession', () => {
     const onData = ptyProvider.onData.mock.calls[0]?.[0] as (payload: {
       id: string
       data: string
+      providerGeneration: number
+      ptyIncarnation: string
     }) => void
 
-    markHiddenRendererPty('ssh-pty-1')
-    onData({ id: 'ssh-pty-1', data: 'hidden ssh output' })
-
-    // Runtime ingestion still ran; renderer delivery shrank to one marker.
-    expect(runtime.onPtyData).toHaveBeenCalledWith(
-      'ssh-pty-1',
-      'hidden ssh output',
-      expect.any(Number),
-      'hidden ssh output'.length,
-      undefined
-    )
-    expect(mockWindow.webContents.send).toHaveBeenCalledTimes(1)
-    // Why out-of-band: an in-band empty pty:data sentinel is ambiguous with
-    // chunks fully consumed by renderer OSC-9999 stripping.
-    expect(mockWindow.webContents.send).toHaveBeenCalledWith('pty:modelRestoreNeeded', {
+    onData({
       id: 'ssh-pty-1',
-      reason: 'hidden-drop',
-      markerSeq: 99
+      data: 'ssh output',
+      providerGeneration: 41,
+      ptyIncarnation: 'incarnation-1'
     })
 
-    onData({ id: 'ssh-pty-1', data: 'more hidden ssh output' })
-    expect(mockWindow.webContents.send).toHaveBeenCalledTimes(1)
-
-    // Delivery interest (renderer sidecars) suppresses the gate — parity with
-    // the local path in ipc/pty.ts.
-    setRendererPtyDeliveryInterest('ssh-pty-1', true)
-    onData({ id: 'ssh-pty-1', data: 'sidecar ssh bytes' })
-    expect(mockWindow.webContents.send).toHaveBeenLastCalledWith('pty:data', {
+    expect(acceptOutputDataMock).toHaveBeenCalledTimes(1)
+    expect(acceptOutputDataMock).toHaveBeenCalledWith({
       id: 'ssh-pty-1',
-      data: 'sidecar ssh bytes',
-      seq: 99,
-      rawLength: 'sidecar ssh bytes'.length
-    })
-
-    // Non-hidden PTYs are unaffected.
-    onData({ id: 'ssh-pty-2', data: 'visible ssh output' })
-    expect(mockWindow.webContents.send).toHaveBeenLastCalledWith('pty:data', {
-      id: 'ssh-pty-2',
-      data: 'visible ssh output',
-      seq: 99,
-      rawLength: 'visible ssh output'.length
+      data: 'ssh output',
+      providerGeneration: 41,
+      ptyIncarnation: 'incarnation-1',
+      rawLength: 'ssh output'.length,
+      transformed: false
     })
+    expect(runtime.onPtyData).not.toHaveBeenCalled()
+    expect(mockWindow.webContents.send).not.toHaveBeenCalledWith('pty:data', expect.anything())
   })
 
-  it('keeps hidden SSH delivery when the gate kill switch is off', async () => {
+  it('keeps the bounded intake path when hidden delivery gating is disabled', async () => {
     const { mockConn, mockStore, mockPortForward, getMainWindow, mockWindow } = createMockDeps()
     ;(mockStore as unknown as { getSettings: () => unknown }).getSettings = vi.fn(() => ({
       terminalHiddenDeliveryGate: false
@@ -197,16 +175,26 @@ describe('SshRelaySession', () => {
     const onData = ptyProvider.onData.mock.calls[0]?.[0] as (payload: {
       id: string
       data: string
+      providerGeneration: number
+      ptyIncarnation: string
     }) => void
 
-    markHiddenRendererPty('ssh-pty-1')
-    onData({ id: 'ssh-pty-1', data: 'still delivered' })
-
-    expect(mockWindow.webContents.send).toHaveBeenCalledWith('pty:data', {
+    onData({
       id: 'ssh-pty-1',
       data: 'still delivered',
-      rawLength: 'still delivered'.length
+      providerGeneration: 41,
+      ptyIncarnation: 'incarnation-1'
     })
+
+    expect(acceptOutputDataMock).toHaveBeenCalledWith({
+      id: 'ssh-pty-1',
+      data: 'still delivered',
+      providerGeneration: 41,
+      ptyIncarnation: 'incarnation-1',
+      rawLength: 'still delivered'.length,
+      transformed: false
+    })
+    expect(mockWindow.webContents.send).not.toHaveBeenCalledWith('pty:data', expect.anything())
   })
 
   it('starts in idle state', () => {

src/main/ssh/ssh-relay-session.ts

@@ -37,10 +37,11 @@ import {
   isCurrentPtyExit
 } from '../ipc/pty'
 import {
-  recordHiddenRendererPtyDataDrop,
-  shouldDropHiddenRendererPtyData
-} from '../ipc/pty-hidden-delivery-gate'
-import type { PtyModelRestoreNeededEvent } from '../../shared/pty-model-restore-marker'
+  acceptSshPtyOutputData,
+  acceptSshPtyOutputExit,
+  allocateSshPtyProviderGeneration,
+  closeSshPtyOutputGeneration
+} from '../ipc/ssh-pty-output-intake-registry'
 import {
   registerSshFilesystemProvider,
   unregisterSshFilesystemProvider,
@@ -151,6 +152,7 @@ export class SshRelaySession {
   private remoteCliBridgeEnv: RemoteCliBridgeEnv | null = null
   private forwardedReattachReplayByPty = new Map<string, ForwardedReplayFingerprint>()
   private pendingPtyReattaches = new Map<string, PendingPtyReattach>()
+  private activePtyProviderGeneration: number | null = null
 
   constructor(
     readonly targetId: string,
@@ -532,7 +534,14 @@ export class SshRelaySession {
 
     this.wireUpRemoteOrcaCli(mux)
 
-    const ptyProvider = new SshPtyProvider(this.targetId, mux, this.remoteCliBridgeEnv ?? undefined)
+    const providerGeneration = allocateSshPtyProviderGeneration()
+    const ptyProvider = new SshPtyProvider(
+      this.targetId,
+      mux,
+      this.remoteCliBridgeEnv ?? undefined,
+      providerGeneration
+    )
+    this.activePtyProviderGeneration = providerGeneration
     registerSshPtyProvider(this.targetId, ptyProvider)
 
     const connection = this.requireReadyConnection()
@@ -826,6 +835,10 @@ export class SshRelaySession {
       this.mux.dispose(reason)
     }
     this.mux = null
+    if (this.activePtyProviderGeneration !== null) {
+      closeSshPtyOutputGeneration(this.activePtyProviderGeneration, reason)
+      this.activePtyProviderGeneration = null
+    }
 
     if (reason === 'shutdown') {
       clearPtyOwnershipForConnection(this.targetId)
@@ -920,38 +933,15 @@ export class SshRelaySession {
   private wireUpPtyEvents(ptyProvider: SshPtyProvider): void {
     ptyProvider.onData((payload) => {
       const rawLength = payload.sequenceChars ?? payload.data.length
-      const seq = this.runtime?.onPtyData(
-        payload.id,
-        payload.data,
-        Date.now(),
+      void acceptSshPtyOutputData({
+        id: payload.id,
+        data: payload.data,
+        providerGeneration: payload.providerGeneration,
+        ptyIncarnation: payload.ptyIncarnation,
         rawLength,
-        payload.transformed
-      )
-      const win = this.getMainWindow()
-      if (!win || win.isDestroyed()) {
-        return
-      }
-      // Why: hidden-delivery gate parity with ipc/pty.ts — latch model-restore out-of-band, never an in-band pty:data sentinel (OSC-9999-only chunks strip to empty).
-      const store = this.store as { getSettings?: Store['getSettings'] }
-      if (shouldDropHiddenRendererPtyData(payload.id, store.getSettings?.())) {
-        const drop = recordHiddenRendererPtyDataDrop(payload.id, rawLength)
-        if (drop.shouldEmitRestoreMarker) {
-          win.webContents.send('pty:modelRestoreNeeded', {
-            id: payload.id,
-            reason: 'hidden-drop',
-            ...(typeof seq === 'number' ? { markerSeq: seq } : {})
-          } satisfies PtyModelRestoreNeededEvent)
-        }
-        return
-      }
-      if (payload.data.length > 0 || payload.transformed) {
-        win.webContents.send('pty:data', {
-          ...payload,
-          ...(typeof seq === 'number' ? { seq } : {}),
-          rawLength,
-          ...(payload.transformed ? { transformed: true } : {})
-        })
-      }
+        transformed: payload.transformed === true,
+        ...(typeof payload.seq === 'number' ? { sequence: payload.seq } : {})
+      }).catch(() => {})
     })
     ptyProvider.onReplay((payload) => {
       const win = this.getMainWindow()
@@ -969,16 +959,31 @@ export class SshRelaySession {
       if (!isCurrentPtyExit(payload)) {
         return
       }
-      this.retireExitedPty(payload)
+      void this.acceptPtyExit(payload).catch(() => {})
     })
   }
 
-  private retireExitedPty(payload: SshPtyExitPayload): void {
+  private async acceptPtyExit(payload: SshPtyExitPayload): Promise<void> {
+    await acceptSshPtyOutputExit({
+      id: payload.id,
+      code: payload.code,
+      providerGeneration: payload.providerGeneration,
+      ptyIncarnation: payload.ptyIncarnation
+    })
+    if (isCurrentPtyExit(payload)) {
+      this.retireExitedPty(payload, true)
+    }
+  }
+
+  private retireExitedPty(payload: SshPtyExitPayload, deliveryHandled = false): void {
     const relayPtyId = toRelaySshPtyId(this.targetId, payload.id)
     clearProviderPtyState(payload.id)
     deletePtyOwnership(payload.id)
     this.forwardedReattachReplayByPty.delete(payload.id)
     this.store.markSshRemotePtyLease(this.targetId, relayPtyId, 'terminated')
+    if (deliveryHandled) {
+      return
+    }
     this.runtime?.onPtyExit(payload.id, payload.code, payload.incarnationId)
     const win = this.getMainWindow()
     if (win && !win.isDestroyed()) {
@@ -1069,7 +1074,7 @@ export class SshRelaySession {
             restorePtyIncarnation(appPtyId, attachResult.incarnationId)
             this.runtime?.acceptPtyIncarnationForExit(appPtyId, attachResult.incarnationId)
           }
-          this.retireExitedPty(exitDuringAttach)
+          await this.acceptPtyExit(exitDuringAttach)
           continue
         }
         setPtyOwnership(appPtyId, this.targetId)