fix(terminal): preserve input during parked reveal

This commit is contained in:
Brennan Benson
2026-08-05 13:50:41 -07:00
parent ba21cc8087
commit ebd80cc894
5 changed files with 260 additions and 15 deletions
@@ -20622,14 +20622,22 @@ describe('connectPanePty', () => {
expect(transport.sendInput).toHaveBeenCalledWith('a')
})
// Why: xterm auto-replies during replay must not count as user interaction, or a BELed pane would self-dismiss unseen.
it('does not clear unread when onData fires during replay', async () => {
it('drops parser replies but forwards real user input during replay', async () => {
const { connectPanePty } = await import('./pty-connection')
const transport = createMockTransport()
transportFactoryQueue.push(transport)
const pane = createPane(1)
let onDataHandler: ((data: string) => void) | null = null
let userInputListener: (() => void) | null = null
;(pane.terminal as unknown as { _core: unknown })._core = {
coreService: {
onUserInput: vi.fn((listener: () => void) => {
userInputListener = listener
return { dispose: vi.fn() }
})
}
}
pane.terminal.onData = vi.fn(((handler: (data: string) => void) => {
onDataHandler = handler
return { dispose: vi.fn() }
@@ -20640,13 +20648,24 @@ describe('connectPanePty', () => {
connectPanePty(pane as never, manager as never, deps as never)
if (!onDataHandler) {
throw new Error('expected onData handler to be registered')
if (!onDataHandler || !userInputListener) {
throw new Error('expected terminal input handlers to be registered')
}
;(onDataHandler as (data: string) => void)('\x1b[?1;2c')
expect(transport.sendInput).not.toHaveBeenCalled()
expect(deps.clearTerminalTabUnread).not.toHaveBeenCalled()
expect(deps.clearWorktreeUnread).not.toHaveBeenCalled()
;(userInputListener as () => void)()
await Promise.resolve()
;(onDataHandler as (data: string) => void)('\x1b[?6c')
expect(transport.sendInput).not.toHaveBeenCalled()
;(userInputListener as () => void)()
;(onDataHandler as (data: string) => void)('a')
expect(transport.sendInput).toHaveBeenCalledWith('a')
})
// Why: on a stale-codex pane (pending account-switch restart), onData bytes must not count as user interaction; production also blocks sendInput here.
@@ -3750,10 +3750,18 @@ export function connectPanePty(
// Takeover must never fire from the onData fallback below: it mixes in auto-replies.
reportWorkerTerminalUserInput(cacheKey, runtimeEnvironmentId)
}
const userInputActivityDisposable = subscribeToTerminalUserInput(
pane.terminal,
recordRealUserTerminalInput
)
let pendingRealUserInputToken = 0
let nextRealUserInputToken = 0
const userInputActivityDisposable = subscribeToTerminalUserInput(pane.terminal, () => {
recordRealUserTerminalInput()
const token = ++nextRealUserInputToken
pendingRealUserInputToken = token
queueMicrotask(() => {
if (pendingRealUserInputToken === token) {
pendingRealUserInputToken = 0
}
})
})
const recordTerminalInputForHibernationFallback = (): void => {
if (userInputActivityDisposable === null) {
recordTerminalInputForHibernation()
@@ -4054,6 +4062,8 @@ export function connectPanePty(
)
const onDataDisposable = pane.terminal.onData((data) => {
const isRealUserInput = pendingRealUserInputToken !== 0
pendingRealUserInputToken = 0
// Why: xterm auto-replies to embedded query sequences (DA1, DECRQM,
// OSC 10/11, focus, CPR) via onData. When we replay recorded PTY bytes
// into xterm for scrollback/cold-restore/snapshot, those queries would
@@ -4061,7 +4071,7 @@ export function connectPanePty(
// ("?1;2c", "2026;2$y", OSC color fragments, ...). The replay sites
// engage the guard via replayIntoTerminal; here we drop everything
// xterm emits while the guard is active. See replay-guard.ts.
if (isPaneReplaying(deps.replayingPanesRef, pane.id)) {
if (isPaneReplaying(deps.replayingPanesRef, pane.id) && !isRealUserInput) {
return
}
const currentPtyId = transport.getPtyId()
@@ -837,6 +837,91 @@ describe('createRemoteRuntimePtyTransport', () => {
)
})
it('queues typed input and Enter until a web mirror handle is authoritative', async () => {
const healthyRuntimeCall = runtimeCall.getMockImplementation()
let releaseActivation = (): void => {}
const activationGate = new Promise<void>((resolve) => {
releaseActivation = resolve
})
runtimeCall.mockImplementation((request: { method: string; params?: unknown }) => {
if (request.method === 'session.tabs.activate') {
return activationGate.then(() => healthyRuntimeCall?.(request))
}
if (request.method === 'terminal.send') {
return Promise.resolve({ ok: true, result: { send: { accepted: true } } })
}
return healthyRuntimeCall?.(request)
})
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
const transport = createRemoteRuntimePtyTransport('env-1', {
worktreeId: 'wt-1',
tabId: 'web-terminal-host-tab-1',
leafId: 'pane:1'
})
expect(transport.sendInput('typed-before-connect\r')).toBe(true)
const connect = transport.connect({ url: '', cols: 100, rows: 30, callbacks: {} })
await vi.waitFor(() =>
expect(runtimeCall).toHaveBeenCalledWith(
expect.objectContaining({ method: 'session.tabs.activate' })
)
)
expect(transport.sendInput('typed-before-attach\r')).toBe(true)
const enterAccepted = transport.sendInputAccepted?.('accepted-enter\r')
expect(latestFrameForOpcode(TerminalStreamOpcode.Input)).toBeUndefined()
releaseActivation()
await connect
expect(transport.sendInput('typed-after-handle\r')).toBe(true)
expect(latestFrameForOpcode(TerminalStreamOpcode.Input)).toBeUndefined()
emitSnapshot(latestSubscribePayload().streamId, 'authoritative state')
await vi.waitFor(() =>
expect(runtimeCall).toHaveBeenCalledWith(
expect.objectContaining({
method: 'terminal.send',
params: expect.objectContaining({
text: 'typed-before-connect\rtyped-before-attach\raccepted-enter\rtyped-after-handle\r'
})
})
)
)
await expect(enterAccepted).resolves.toBe(true)
transport.destroy?.()
})
it('drops pre-connect web mirror input when attach takes ownership', async () => {
const healthyRuntimeCall = runtimeCall.getMockImplementation()
let releasePaneAuthority = (): void => {}
const paneAuthorityGate = new Promise<void>((resolve) => {
releasePaneAuthority = resolve
})
runtimeCall.mockImplementation((request: { method: string; params?: unknown }) => {
if (request.method === 'terminal.resolvePane') {
return paneAuthorityGate.then(() => healthyRuntimeCall?.(request))
}
return healthyRuntimeCall?.(request)
})
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
const transport = createRemoteRuntimePtyTransport('env-1', {
worktreeId: 'wt-1',
tabId: 'web-terminal-host-tab-1',
leafId: 'pane:1'
})
expect(transport.sendInput('stale typed input')).toBe(true)
const staleAcceptedInput = transport.sendInputAccepted?.('stale Enter\r')
transport.attach({
existingPtyId: 'remote:env-1@@terminal-1',
callbacks: {}
})
await expect(staleAcceptedInput).resolves.toBe(false)
expect(transport.sendInput('input before attach authority')).toBe(false)
expect(latestFrameForOpcode(TerminalStreamOpcode.Input)).toBeUndefined()
releasePaneAuthority()
transport.destroy?.()
})
it('retries initial web mirror inventory after a transient runtime close', async () => {
const healthyRuntimeCall = runtimeCall.getMockImplementation()
let activateAttempts = 0
@@ -86,6 +86,7 @@ import { getRuntimeEnvironmentRevision } from '@/runtime/runtime-environment-rev
const REMOTE_TERMINAL_INPUT_FLUSH_MS = 8
const REMOTE_TERMINAL_VIEWPORT_FLUSH_MS = 33
const REMOTE_TERMINAL_CONNECTING_INPUT_MAX_CHARS = 256 * 1024
const HOST_SESSION_ATTACH_POLL_MS = 150
const HOST_SESSION_REPLACEMENT_POLL_MAX_MS = 1_000
const HOST_SESSION_ATTACH_TIMEOUT_MS = 15_000
@@ -169,7 +170,14 @@ export function createRemoteRuntimePtyTransport(
let attachmentReady = false
let destroyed = false
let terminalEnded = false
let connecting = false
let connecting = isWebTerminalSurfaceTabId(tabId ?? '')
// Why: cached park frames paint before host authority resolves, so bounded typing waits for the verified subscription instead of disappearing.
let acceptsInputWhileConnecting = connecting
let inputQueuedWhileConnecting = ''
let inputQueuedWhileConnectingChars = 0
let inputQueuedWhileConnectingGeneration = 0
let inputQueuedWhileConnectingFlushActive = false
let inputQueuedWhileConnectingWaiters: ((accepted: boolean) => void)[] = []
const attachmentReadyWaiters = new Set<(ready: boolean) => void>()
// Why: transport methods overlap during remounts; only the latest pane lifecycle may install a returned PTY.
let lifecycleEpoch = 0
@@ -203,6 +211,85 @@ export function createRemoteRuntimePtyTransport(
let attachGeneration = 0
let subscriptionGeneration = 0
function clearInputQueuedWhileConnecting(): void {
inputQueuedWhileConnectingGeneration += 1
acceptsInputWhileConnecting = false
inputQueuedWhileConnecting = ''
inputQueuedWhileConnectingChars = 0
inputQueuedWhileConnectingFlushActive = false
for (const resolve of inputQueuedWhileConnectingWaiters) {
resolve(false)
}
inputQueuedWhileConnectingWaiters = []
}
function queueInputWhileConnecting(
data: string,
resolveAccepted?: (accepted: boolean) => void
): boolean {
if (
!acceptsInputWhileConnecting ||
(!connecting && !inputQueuedWhileConnectingFlushActive) ||
recoveryBlocksIo() ||
inputQueuedWhileConnectingChars + data.length > REMOTE_TERMINAL_CONNECTING_INPUT_MAX_CHARS
) {
return false
}
inputQueuedWhileConnecting += data
inputQueuedWhileConnectingChars += data.length
if (resolveAccepted) {
inputQueuedWhileConnectingWaiters.push(resolveAccepted)
}
return true
}
async function drainInputQueuedWhileConnecting(generation: number): Promise<void> {
while (generation === inputQueuedWhileConnectingGeneration) {
const queuedInput = inputQueuedWhileConnecting
const acceptedWaiters = inputQueuedWhileConnectingWaiters
if (!queuedInput) {
acceptsInputWhileConnecting = false
inputQueuedWhileConnectingFlushActive = false
return
}
inputQueuedWhileConnecting = ''
inputQueuedWhileConnectingWaiters = []
const accepted = acceptedWaiters.length
? await sendInputAcceptedToRuntime(queuedInput, true)
: inputBatcher.push(queuedInput)
if (generation !== inputQueuedWhileConnectingGeneration) {
for (const resolve of acceptedWaiters) {
resolve(false)
}
return
}
inputQueuedWhileConnectingChars -= queuedInput.length
for (const resolve of acceptedWaiters) {
resolve(accepted)
}
if (!accepted && acceptedWaiters.length === 0) {
notifyWriteUnavailable()
}
}
}
function flushInputQueuedWhileConnecting(): void {
if (
!acceptsInputWhileConnecting ||
inputQueuedWhileConnectingFlushActive ||
!attachmentReady ||
!connected ||
!handle ||
recoveryBlocksIo() ||
!multiplexedStream
) {
return
}
inputQueuedWhileConnectingFlushActive = true
const generation = inputQueuedWhileConnectingGeneration
void drainInputQueuedWhileConnecting(generation)
}
function setAttachmentReady(ready: boolean): void {
attachmentReady = ready
if (!ready) {
@@ -247,6 +334,7 @@ export function createRemoteRuntimePtyTransport(
clearPublishedHandleWait()
}
if (recovery.currentPhase === 'disconnected') {
clearInputQueuedWhileConnecting()
autoRecoveryWindowSpent = true
// Why: cached pixels may remain, but no stream from the exhausted epoch may keep delivering or accepting terminal traffic.
subscriptionGeneration += 1
@@ -1195,7 +1283,17 @@ export function createRemoteRuntimePtyTransport(
return recovery.isActive || recovery.currentPhase === 'disconnected'
}
async function sendInputAcceptedToRuntime(data: string): Promise<boolean> {
async function sendInputAcceptedToRuntime(
data: string,
bypassConnectingQueue = false
): Promise<boolean> {
if (data && acceptsInputWhileConnecting && !bypassConnectingQueue) {
return new Promise((resolve) => {
if (!queueInputWhileConnecting(data, resolve)) {
resolve(false)
}
})
}
const targetHandle = handle
if (!connected || !targetHandle || recoveryBlocksIo()) {
return false
@@ -1365,6 +1463,7 @@ export function createRemoteRuntimePtyTransport(
recovery.cancel()
resetRecoveryReplacementPolicy()
resetSameHandleEndReuse()
clearInputQueuedWhileConnecting()
connected = false
connecting = false
terminalEnded = true
@@ -1494,6 +1593,7 @@ export function createRemoteRuntimePtyTransport(
return
}
connecting = false
clearInputQueuedWhileConnecting()
emitRecoveryState()
surfaceErrorMessage(message)
}
@@ -1799,6 +1899,7 @@ export function createRemoteRuntimePtyTransport(
emitRecoveryState()
storedCallbacks.onConnect?.()
storedCallbacks.onStatus?.('shell')
flushInputQueuedWhileConnecting()
},
onEnd: () => {
if (!isCurrentSubscription()) {
@@ -1819,6 +1920,7 @@ export function createRemoteRuntimePtyTransport(
unregisterShutdownHandlers(subscribedPtyId)
connected = false
connecting = false
clearInputQueuedWhileConnecting()
handle = null
remotePtyId = null
multiplexedStream = null
@@ -1876,6 +1978,7 @@ export function createRemoteRuntimePtyTransport(
}
} else {
connecting = false
clearInputQueuedWhileConnecting()
recovery.cancel()
setAttachmentUnavailable()
emitRecoveryState()
@@ -1923,16 +2026,22 @@ export function createRemoteRuntimePtyTransport(
) {
nextStream.resize(desiredViewport.cols, desiredViewport.rows)
}
flushInputQueuedWhileConnecting()
}
const transport: PtyTransport = {
async connect(options) {
cancelTerminalCreateRetryWait()
const acceptsPreConnectInput = lifecycleEpoch === 0 && acceptsInputWhileConnecting
const connectLifecycleEpoch = ++lifecycleEpoch
const createEnvironmentId = currentRuntimeEnvironmentId
lastConnectOptions = options
lastAttachOptions = null
lastSurfacedErrorMessage = null
if (!acceptsPreConnectInput) {
clearInputQueuedWhileConnecting()
}
acceptsInputWhileConnecting = isWebTerminalSurfaceTabId(tabId ?? '')
storedCallbacks = options.callbacks
resetRecoveryReplacementPolicy()
resetSameHandleEndReuse()
@@ -1940,12 +2049,22 @@ export function createRemoteRuntimePtyTransport(
connecting = true
emitRecoveryState(true)
if (destroyed || !worktreeId) {
clearInputQueuedWhileConnecting()
return
}
try {
if (isWebTerminalSurfaceTabId(tabId ?? '')) {
return await attachHostSessionMirror(options, true, undefined, connectLifecycleEpoch)
const result = await attachHostSessionMirror(
options,
true,
undefined,
connectLifecycleEpoch
)
if (!result) {
clearInputQueuedWhileConnecting()
}
return result
}
if (options.sessionId && !getRemoteRuntimeTerminalHandle(options.sessionId)) {
@@ -2069,6 +2188,7 @@ export function createRemoteRuntimePtyTransport(
if (!created) {
if (!destroyed && lifecycleEpoch === connectLifecycleEpoch) {
connecting = false
clearInputQueuedWhileConnecting()
recovery.markDisconnected()
}
return
@@ -2154,6 +2274,7 @@ export function createRemoteRuntimePtyTransport(
surfaceErrorMessage(message)
}
}
clearInputQueuedWhileConnecting()
return undefined
}
},
@@ -2168,6 +2289,7 @@ export function createRemoteRuntimePtyTransport(
clearPublishedHandleWait()
lastAttachOptions = options
lastSurfacedErrorMessage = null
clearInputQueuedWhileConnecting()
storedCallbacks = options.callbacks
terminalEnded = false
connecting = true
@@ -2268,6 +2390,7 @@ export function createRemoteRuntimePtyTransport(
clearPublishedHandleWait()
inputBatcher.flush()
inputBatcher.clear()
clearInputQueuedWhileConnecting()
viewportBatcher.flush()
outputProcessor.clearAccumulatedState()
if (!connected && !handle) {
@@ -2300,6 +2423,7 @@ export function createRemoteRuntimePtyTransport(
clearPublishedHandleWait()
inputBatcher.flush()
inputBatcher.clear()
clearInputQueuedWhileConnecting()
viewportBatcher.flush()
outputProcessor.clearAccumulatedState()
unregisterShutdownHandlers(remotePtyId)
@@ -2313,6 +2437,9 @@ export function createRemoteRuntimePtyTransport(
},
sendInput(data: string): boolean {
if (acceptsInputWhileConnecting) {
return data ? queueInputWhileConnecting(data) : false
}
if (!connected || !handle || recoveryBlocksIo()) {
return false
}
@@ -2519,6 +2646,7 @@ export function createRemoteRuntimePtyTransport(
this.disconnect()
recovery.dispose()
inputBatcher.clear()
clearInputQueuedWhileConnecting()
viewportBatcher.clear()
}
}
@@ -20,20 +20,23 @@ describe('subscribeToTerminalUserInput', () => {
const listener = vi.fn()
const subscription = subscribeToTerminalUserInput(terminal, listener)
expect(subscription).not.toBeNull()
const eventOrder: string[] = []
const onData = vi.fn(() => eventOrder.push('data'))
terminal.onData(onData)
listener.mockImplementation(() => eventOrder.push('user-input'))
const coreService = (terminal as unknown as CoreServiceAccess)._core.coreService
// Keyboard/IME/paste/mouse paths mark their data as user input.
coreService.triggerDataEvent('a', true)
expect(listener).toHaveBeenCalledTimes(1)
expect(eventOrder).toEqual(['user-input', 'data'])
// Parser-generated replies (focus reports, DA/DSR/CPR responses) are not
// user input and must not fire the signal, while still flowing to onData.
const onData = vi.fn()
terminal.onData(onData)
coreService.triggerDataEvent('\x1b[O', false)
coreService.triggerDataEvent('\x1b[?1;2c')
expect(listener).toHaveBeenCalledTimes(1)
expect(onData).toHaveBeenCalledTimes(2)
expect(onData).toHaveBeenCalledTimes(3)
subscription?.dispose()
coreService.triggerDataEvent('b', true)