diff --git a/src/main/ipc/pty-pending-data-drain-contract.ts b/src/main/ipc/pty-pending-data-drain-contract.ts index 05fbd2ff77e..9ddfa755ffb 100644 --- a/src/main/ipc/pty-pending-data-drain-contract.ts +++ b/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' diff --git a/src/main/ipc/pty.ts b/src/main/ipc/pty.ts index feae29a0096..1944425f4bf 100644 --- a/src/main/ipc/pty.ts +++ b/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() const rendererDeliveryRestoreNeededPtys = new Set() @@ -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) diff --git a/src/main/ipc/ssh-pty-legacy-projection-record.ts b/src/main/ipc/ssh-pty-legacy-projection-record.ts new file mode 100644 index 00000000000..9f5077c8d4d --- /dev/null +++ b/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 + afterScanner: Readonly + 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 { + 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, + cursors: ReadonlyMap, + 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, + idsByPty: Map, + 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, + id: string +): ProjectionRecord { + const record = records.get(id) + if (!record) { + throw projectionError('ssh_projection_reservation_missing') + } + return record +} + +export function rollbackUnpublishedProjection( + records: Map, + cursors: ReadonlyMap, + idsByPty: Map, + 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, + 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, + 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 +} diff --git a/src/main/ipc/ssh-pty-legacy-projection.test.ts b/src/main/ipc/ssh-pty-legacy-projection.test.ts new file mode 100644 index 00000000000..c396592d1a5 --- /dev/null +++ b/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[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 }) + }) +}) diff --git a/src/main/ipc/ssh-pty-legacy-projection.ts b/src/main/ipc/ssh-pty-legacy-projection.ts new file mode 100644 index 00000000000..c1042d4815f --- /dev/null +++ b/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() + private readonly cursorByPty = new Map() + private readonly idsByPty = new Map() + 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 + }) + } +} diff --git a/src/main/ipc/ssh-pty-model-admission-contract.ts b/src/main/ipc/ssh-pty-model-admission-contract.ts new file mode 100644 index 00000000000..98df5b1c34d --- /dev/null +++ b/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 +} diff --git a/src/main/ipc/ssh-pty-model-admission-entry.ts b/src/main/ipc/ssh-pty-model-admission-entry.ts new file mode 100644 index 00000000000..e73884f1efa --- /dev/null +++ b/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 } + 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 + closingGenerations: ReadonlySet + 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, + 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, + pressure: readonly AdmissionEntry[], + waitersById: Map 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() + } +} diff --git a/src/main/ipc/ssh-pty-model-admission-limits.ts b/src/main/ipc/ssh-pty-model-admission-limits.ts new file mode 100644 index 00000000000..865242de284 --- /dev/null +++ b/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 + } +} diff --git a/src/main/ipc/ssh-pty-model-admission.test.ts b/src/main/ipc/ssh-pty-model-admission.test.ts new file mode 100644 index 00000000000..95178659c44 --- /dev/null +++ b/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((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 }) + }) +}) diff --git a/src/main/ipc/ssh-pty-model-admission.ts b/src/main/ipc/ssh-pty-model-admission.ts new file mode 100644 index 00000000000..792c5a9cc5b --- /dev/null +++ b/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() + private readonly pressure: AdmissionEntry[] = [] + private readonly pausedKeys = new Map() + private readonly idleWaiters = new Map void>>() + private readonly closingGenerations = new Set() + 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 } + ): Promise { + 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 { + 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() + 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 } + 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) + } + } +} diff --git a/src/main/ipc/ssh-pty-output-intake-contract.ts b/src/main/ipc/ssh-pty-output-intake-contract.ts new file mode 100644 index 00000000000..00d76c78b43 --- /dev/null +++ b/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 } + 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 +} diff --git a/src/main/ipc/ssh-pty-output-intake-registry.ts b/src/main/ipc/ssh-pty-output-intake-registry.ts new file mode 100644 index 00000000000..732d94d04cb --- /dev/null +++ b/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 { + return installedIntake + ? installedIntake.acceptData(event) + : Promise.reject(outputIntakeUnavailableError()) +} + +export function acceptSshPtyOutputExit(event: SshPtyOutputExitEvent): Promise { + 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' + }) +} diff --git a/src/main/ipc/ssh-pty-output-intake-validation.ts b/src/main/ipc/ssh-pty-output-intake-validation.ts new file mode 100644 index 00000000000..5c9b346860f --- /dev/null +++ b/src/main/ipc/ssh-pty-output-intake-validation.ts @@ -0,0 +1,16 @@ +export type SshPtyExitBarrier = { + timer: ReturnType + 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}` +} diff --git a/src/main/ipc/ssh-pty-output-intake.test.ts b/src/main/ipc/ssh-pty-output-intake.test.ts new file mode 100644 index 00000000000..82c7c428943 --- /dev/null +++ b/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((promiseResolve, promiseReject) => { + resolve = promiseResolve + reject = promiseReject + }) + return { promise, resolve, reject } +} + +function event(overrides: Partial = {}): SshPtyOutputDataEvent { + return { + id: 'pty-1', + data: 'aaaa', + providerGeneration: 1, + ptyIncarnation: 'incarnation-1', + rawLength: 4, + transformed: false, + ...overrides + } +} + +function createHarness( + overrides: Partial = {}, + options: ConstructorParameters[1] = {} +) { + let sequence = 0 + const completions: ReturnType[] = [] + 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') + }) +}) diff --git a/src/main/ipc/ssh-pty-output-intake.ts b/src/main/ipc/ssh-pty-output-intake.ts new file mode 100644 index 00000000000..667864d6c33 --- /dev/null +++ b/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() + private readonly incarnationByPty = new Map() + private readonly sealedPtys = new Set() + private readonly closedGenerations = new Set() + private readonly exitBarriersByGeneration = new Map>() + 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 { + 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 } + 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 { + 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): Promise { + 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) + } + } +} diff --git a/src/main/providers/ssh-pty-notification-routing.test.ts b/src/main/providers/ssh-pty-notification-routing.test.ts index 2c1ab863295..e70a6dad137 100644 --- a/src/main/providers/ssh-pty-notification-routing.test.ts +++ b/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' }) }) diff --git a/src/main/providers/ssh-pty-notification-routing.ts b/src/main/providers/ssh-pty-notification-routing.ts index 1169674211f..8b5328465ec 100644 --- a/src/main/providers/ssh-pty-notification-routing.ts +++ b/src/main/providers/ssh-pty-notification-routing.ts @@ -16,6 +16,8 @@ export function subscribeSshPtyNotifications(args: { exitListeners: Set livePtyIds: Set 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 } : {}) diff --git a/src/main/providers/ssh-pty-provider-contract.ts b/src/main/providers/ssh-pty-provider-contract.ts index 9043133b03c..9b709d9e08d 100644 --- a/src/main/providers/ssh-pty-provider-contract.ts +++ b/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 diff --git a/src/main/providers/ssh-pty-provider-exit-race.test.ts b/src/main/providers/ssh-pty-provider-exit-race.test.ts index 12616e4d133..411a3373cd8 100644 --- a/src/main/providers/ssh-pty-provider-exit-race.test.ts +++ b/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({ diff --git a/src/main/providers/ssh-pty-provider-output-state.ts b/src/main/providers/ssh-pty-provider-output-state.ts new file mode 100644 index 00000000000..e6559e8cac0 --- /dev/null +++ b/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() + private readonly replayListeners = new Set() + private readonly exitListeners = new Set() + private readonly incarnationByRelayPtyId = new Map() + private readonly pausedRelayPtyIds = new Set() + 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 + 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() + } +} diff --git a/src/main/providers/ssh-pty-provider-process-events.test.ts b/src/main/providers/ssh-pty-provider-process-events.test.ts index 2d033aa40d3..2d7e5846e61 100644 --- a/src/main/providers/ssh-pty-provider-process-events.test.ts +++ b/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 }] + ]) }) }) diff --git a/src/main/providers/ssh-pty-provider.ts b/src/main/providers/ssh-pty-provider.ts index 59f20bfdf0a..7c8c3438c75 100644 --- a/src/main/providers/ssh-pty-provider.ts +++ b/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() - private replayListeners = new Set() - private exitListeners = new Set() private livePtyIds = new Set() - // Why: stale notification callbacks must not outlive a disconnected provider. - private unsubscribeNotifications: (() => void) | null = null readonly getAppliedSize: NonNullable 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 }) } } diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index aa4d66ada6d..1282bc08813 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -1417,6 +1417,11 @@ type RuntimeHeadlessTerminal = { writeChain: Promise } +export type RuntimePtyDataAdmission = Readonly<{ + sequence: number + completion: Promise +}> + 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 | 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 ): 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 { 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 diff --git a/src/main/ssh/ssh-relay-session-data-delivery.test.ts b/src/main/ssh/ssh-relay-session-data-delivery.test.ts index 901f7940d31..06c09d1b082 100644 --- a/src/main/ssh/ssh-relay-session-data-delivery.test.ts +++ b/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()) }) }) diff --git a/src/main/ssh/ssh-relay-session-incarnation.test.ts b/src/main/ssh/ssh-relay-session-incarnation.test.ts index a98180308b2..b06a78bfca1 100644 --- a/src/main/ssh/ssh-relay-session-incarnation.test.ts +++ b/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() }) }) diff --git a/src/main/ssh/ssh-relay-session-reconnect-incarnation.test.ts b/src/main/ssh/ssh-relay-session-reconnect-incarnation.test.ts index 1a3e8721064..0fd63d68d6a 100644 --- a/src/main/ssh/ssh-relay-session-reconnect-incarnation.test.ts +++ b/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 } 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) diff --git a/src/main/ssh/ssh-relay-session.test.ts b/src/main/ssh/ssh-relay-session.test.ts index 8fe7a292bcf..653bb28237d 100644 --- a/src/main/ssh/ssh-relay-session.test.ts +++ b/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', () => { diff --git a/src/main/ssh/ssh-relay-session.ts b/src/main/ssh/ssh-relay-session.ts index fdd0b115203..af35f6416a9 100644 --- a/src/main/ssh/ssh-relay-session.ts +++ b/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() private pendingPtyReattaches = new Map() + 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 { + 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)