mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
refactor(terminal): one remote input hold (#27781)
* refactor(terminal): one remote input hold * fix(terminal): keep restored typing scoped to its terminal until the connect settles * refactor(terminal): name the restored-input expectation type --------- Co-authored-by: m4air <m4air@Mac.localdomain>
This commit is contained in:
@@ -8583,7 +8583,6 @@
|
||||
"src/renderer/src/components/terminal-pane/remote-layout-scrollback-retention.test.ts": 64,
|
||||
"src/renderer/src/components/terminal-pane/remote-pane-layout-push.test.ts": 71,
|
||||
"src/renderer/src/components/terminal-pane/remote-park-reveal-unverifiable-retry.test.ts": 988,
|
||||
"src/renderer/src/components/terminal-pane/remote-reattach-input-buffer.test.ts": 84,
|
||||
"src/renderer/src/components/terminal-pane/remote-runtime-connect-failure-recovery.test.ts": 1519,
|
||||
"src/renderer/src/components/terminal-pane/remote-runtime-disconnected-input-grace-retry.test.ts": 2324,
|
||||
"src/renderer/src/components/terminal-pane/remote-runtime-disconnected-input-grace.test.ts": 43,
|
||||
|
||||
@@ -233,8 +233,6 @@ export type PtyTransport = {
|
||||
retryRecovery?: () => boolean
|
||||
/** True while the transport has a retry armed or parked; pane-level remounts must defer to it. */
|
||||
ownsRecovery?: () => boolean
|
||||
/** Lets a wrapper retain input when recovery re-enters connect internally. */
|
||||
setConnectForRecovery?: (connect: PtyTransport['connect']) => void
|
||||
/** The user dismissed the error surface; the next occurrence of the same message must surface again. */
|
||||
notifyErrorSurfaceDismissed?: () => void
|
||||
getPtyId: () => string | null
|
||||
|
||||
@@ -1,150 +0,0 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { createDeferred, flushAsyncTicks } from './pty-connection-test-async'
|
||||
import type { PtyConnectResult, PtyTransport } from './pty-transport-types'
|
||||
import { withRemoteReattachInputBuffer } from './remote-reattach-input-buffer'
|
||||
|
||||
const ORIGINAL_ID = 'remote:environment@@term_original'
|
||||
|
||||
function createHarness() {
|
||||
const connected = createDeferred<PtyConnectResult | undefined>()
|
||||
let ptyId: string | null = null
|
||||
const writes: string[] = []
|
||||
const delegate: PtyTransport = {
|
||||
connect: vi.fn(() => connected.promise),
|
||||
attach: vi.fn(),
|
||||
disconnect: vi.fn(),
|
||||
detach: vi.fn(),
|
||||
destroy: vi.fn(),
|
||||
isConnected: () => ptyId !== null,
|
||||
getPtyId: () => ptyId,
|
||||
resize: () => true,
|
||||
sendInput: vi.fn((data) => {
|
||||
if (!ptyId) {
|
||||
return false
|
||||
}
|
||||
writes.push(data)
|
||||
return true
|
||||
}),
|
||||
sendInputImmediate: vi.fn(() => false),
|
||||
sendInputAccepted: vi.fn(async (data) => {
|
||||
if (!ptyId) {
|
||||
return false
|
||||
}
|
||||
writes.push(data)
|
||||
return true
|
||||
})
|
||||
}
|
||||
return {
|
||||
delegate,
|
||||
connected,
|
||||
writes,
|
||||
transport: withRemoteReattachInputBuffer(delegate),
|
||||
finish(id: string | null = ORIGINAL_ID) {
|
||||
ptyId = id
|
||||
connected.resolve(id ? { id, isReattach: true } : undefined)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
describe('remote reattach type-ahead', () => {
|
||||
it('preserves typing from the restored screen before connect starts', async () => {
|
||||
const harness = createHarness()
|
||||
expect(harness.transport.sendInput('before', 'driving')).toBe(true)
|
||||
const accepted = harness.transport.sendInputAccepted?.('\x03', 'driving')
|
||||
const connecting = harness.transport.connect({ url: '', sessionId: ORIGINAL_ID, callbacks: {} })
|
||||
expect(harness.transport.sendInput('after\r', 'driving')).toBe(true)
|
||||
harness.finish()
|
||||
await connecting
|
||||
await expect(accepted).resolves.toBe(true)
|
||||
expect(harness.writes).toEqual(['before', '\x03', 'after\r'])
|
||||
})
|
||||
|
||||
it('discards pre-entry typing when the first connection creates a fresh terminal', async () => {
|
||||
const harness = createHarness()
|
||||
const accepted = harness.transport.sendInputAccepted?.('old tail\r', 'driving')
|
||||
const connecting = harness.transport.connect({ url: '', callbacks: {} })
|
||||
await expect(accepted).resolves.toBe(false)
|
||||
harness.finish('remote:environment@@fresh')
|
||||
await connecting
|
||||
expect(harness.writes).toEqual([])
|
||||
})
|
||||
|
||||
it('delivers typing and an acknowledged interrupt once the original endpoint reattaches', async () => {
|
||||
const harness = createHarness()
|
||||
const connecting = harness.transport.connect({ url: '', sessionId: ORIGINAL_ID, callbacks: {} })
|
||||
expect(harness.transport.sendInput('partial', 'driving')).toBe(true)
|
||||
const accepted = harness.transport.sendInputAccepted?.('\x03', 'driving')
|
||||
expect(harness.transport.sendInput('next\r', 'driving')).toBe(true)
|
||||
expect(harness.writes).toEqual([])
|
||||
harness.finish()
|
||||
await connecting
|
||||
await expect(accepted).resolves.toBe(true)
|
||||
expect(harness.writes).toEqual(['partial', '\x03', 'next\r'])
|
||||
expect(harness.transport.sendInput('later', 'driving')).toBe(true)
|
||||
expect(harness.writes.at(-1)).toBe('later')
|
||||
})
|
||||
|
||||
it.each(['remote:environment@@term_replacement', null])(
|
||||
'never forwards retained input when the original endpoint is unavailable: %s',
|
||||
async (replacementId) => {
|
||||
const harness = createHarness()
|
||||
const connecting = harness.transport.connect({
|
||||
url: '',
|
||||
sessionId: ORIGINAL_ID,
|
||||
callbacks: {}
|
||||
})
|
||||
const accepted = harness.transport.sendInputAccepted?.('dangerous tail\r', 'driving')
|
||||
harness.finish(replacementId)
|
||||
await connecting
|
||||
await expect(accepted).resolves.toBe(false)
|
||||
expect(harness.writes).toEqual([])
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['disconnect', 'detach', 'destroy'] as const)(
|
||||
'drops input and settles accepted writes on %s before a late reattach',
|
||||
async (action) => {
|
||||
const harness = createHarness()
|
||||
const connecting = harness.transport.connect({
|
||||
url: '',
|
||||
sessionId: ORIGINAL_ID,
|
||||
callbacks: {}
|
||||
})
|
||||
const accepted = harness.transport.sendInputAccepted?.('stale\r', 'driving')
|
||||
harness.transport[action]?.()
|
||||
await expect(accepted).resolves.toBe(false)
|
||||
harness.finish()
|
||||
await connecting
|
||||
expect(harness.writes).toEqual([])
|
||||
}
|
||||
)
|
||||
|
||||
it('keeps newly typed bytes behind a pending acknowledged write', async () => {
|
||||
const harness = createHarness()
|
||||
const acknowledgement = createDeferred<boolean>()
|
||||
harness.delegate.sendInputAccepted = () => acknowledgement.promise
|
||||
const transport = withRemoteReattachInputBuffer(harness.delegate)
|
||||
const connecting = transport.connect({ url: '', sessionId: ORIGINAL_ID, callbacks: {} })
|
||||
const accepted = transport.sendInputAccepted?.('\x03', 'driving')
|
||||
harness.finish()
|
||||
await flushAsyncTicks()
|
||||
expect(transport.sendInput('after interrupt\r', 'driving')).toBe(true)
|
||||
expect(harness.writes).toEqual([])
|
||||
acknowledgement.resolve(true)
|
||||
await connecting
|
||||
await expect(accepted).resolves.toBe(true)
|
||||
expect(harness.writes).toEqual(['after interrupt\r'])
|
||||
})
|
||||
|
||||
it('rejects pending writes when connection fails and leaves query replies unbuffered', async () => {
|
||||
const harness = createHarness()
|
||||
const connecting = harness.transport.connect({ url: '', sessionId: ORIGINAL_ID, callbacks: {} })
|
||||
const accepted = harness.transport.sendInputAccepted?.('typed', 'driving')
|
||||
expect(harness.transport.sendInput('\x1b[I', 'query-reply')).toBe(false)
|
||||
expect(harness.transport.sendInputImmediate('\x1b[1;1R')).toBe(false)
|
||||
harness.connected.reject(new Error('offline'))
|
||||
await expect(connecting).rejects.toThrow('offline')
|
||||
await expect(accepted).resolves.toBe(false)
|
||||
expect(harness.writes).toEqual([])
|
||||
})
|
||||
})
|
||||
@@ -1,90 +0,0 @@
|
||||
import { parseRemoteRuntimePtyId } from '../../../../shared/remote-runtime-pty-id'
|
||||
import { createPtyPreconnectInputBuffer } from './pty-preconnect-input-buffer'
|
||||
import type { PtyTransport } from './pty-transport-types'
|
||||
|
||||
/** Keeps type-ahead on a restored screen bound to its original remote terminal. */
|
||||
export function withRemoteReattachInputBuffer(transport: PtyTransport): PtyTransport {
|
||||
const sendAccepted = transport.sendInputAccepted?.bind(transport)
|
||||
let pending: ReturnType<typeof createPtyPreconnectInputBuffer> | null =
|
||||
createPtyPreconnectInputBuffer()
|
||||
let entered = false
|
||||
const clear = (): void => {
|
||||
pending?.clear()
|
||||
pending = null
|
||||
}
|
||||
const wrapped: PtyTransport = {
|
||||
...transport,
|
||||
async connect(options) {
|
||||
const expectedId = options.sessionId
|
||||
const initialBuffer = entered ? null : pending
|
||||
entered = true
|
||||
const buffer =
|
||||
expectedId && parseRemoteRuntimePtyId(expectedId)
|
||||
? (initialBuffer ?? createPtyPreconnectInputBuffer())
|
||||
: null
|
||||
// Restored pixels can accept typing before the first connect call begins.
|
||||
if (pending !== buffer) {
|
||||
clear()
|
||||
}
|
||||
pending = buffer
|
||||
try {
|
||||
const result = await transport.connect(options)
|
||||
if (buffer) {
|
||||
await buffer.flush({
|
||||
// A replacement endpoint must never receive the old terminal's unfinished command.
|
||||
isCurrent: () => pending === buffer && transport.getPtyId() === expectedId,
|
||||
sendInput: (data, kind) => transport.sendInput(data, kind),
|
||||
sendInputImmediate: (data) => transport.sendInputImmediate(data),
|
||||
...(sendAccepted ? { sendInputAccepted: sendAccepted } : {})
|
||||
})
|
||||
}
|
||||
return result
|
||||
} finally {
|
||||
buffer?.clear()
|
||||
if (pending === buffer) {
|
||||
pending = null
|
||||
}
|
||||
}
|
||||
},
|
||||
sendInput(data, kind) {
|
||||
return kind !== 'query-reply' && pending?.isBuffering()
|
||||
? pending.enqueue(data, 'ordinary', kind)
|
||||
: transport.sendInput(data, kind)
|
||||
},
|
||||
// Emulator replies stay on the immediate path; replay must not retain them as user input.
|
||||
...(sendAccepted
|
||||
? {
|
||||
sendInputAccepted: (data, kind, options) =>
|
||||
kind !== 'query-reply' && pending?.isBuffering()
|
||||
? pending.enqueueAccepted(data, kind, undefined, options)
|
||||
: sendAccepted(data, kind, options)
|
||||
}
|
||||
: {}),
|
||||
attach(options) {
|
||||
entered = true
|
||||
clear()
|
||||
transport.attach(options)
|
||||
},
|
||||
disconnect() {
|
||||
entered = true
|
||||
clear()
|
||||
transport.disconnect()
|
||||
},
|
||||
...(transport.detach
|
||||
? {
|
||||
detach: (options) => {
|
||||
entered = true
|
||||
clear()
|
||||
transport.detach?.(options)
|
||||
}
|
||||
}
|
||||
: {}),
|
||||
destroy(options) {
|
||||
entered = true
|
||||
clear()
|
||||
return transport.destroy?.(options)
|
||||
}
|
||||
}
|
||||
transport.setConnectForRecovery?.((options) => wrapped.connect(options))
|
||||
return wrapped
|
||||
}
|
||||
+203
-45
@@ -1,68 +1,226 @@
|
||||
import { beforeEach, expect, it, vi } from 'vitest'
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
TerminalStreamOpcode,
|
||||
decodeTerminalStreamText
|
||||
} from '../../../../shared/terminal-stream-protocol'
|
||||
import { createDeferred } from './pty-connection-test-async'
|
||||
import { createDeferred, flushAsyncTicks } from './pty-connection-test-async'
|
||||
import {
|
||||
createRemoteRuntimeTransportMocks,
|
||||
type MultiplexSubscriptionCallbacks
|
||||
} from './remote-runtime-pty-transport-test-harness'
|
||||
|
||||
const ORIGINAL_ID = 'remote:env-1@@terminal-1'
|
||||
|
||||
let callbacks: MultiplexSubscriptionCallbacks = null
|
||||
let resolvedHandle = 'terminal-1'
|
||||
const { runtimeCall, latestFrameForOpcode, resetRemoteRuntimeTransport } =
|
||||
createRemoteRuntimeTransportMocks({
|
||||
getCallbacks: () => callbacks,
|
||||
setCallbacks: (value) => {
|
||||
callbacks = value
|
||||
},
|
||||
getResolvedPaneHandle: () => resolvedHandle,
|
||||
setResolvedPaneHandle: (value) => {
|
||||
resolvedHandle = value
|
||||
}
|
||||
})
|
||||
const {
|
||||
runtimeCall,
|
||||
runtimeSubscribe,
|
||||
latestFrameForOpcode,
|
||||
latestSubscribePayload,
|
||||
emitSnapshot,
|
||||
inputFrameTexts,
|
||||
resetRemoteRuntimeTransport
|
||||
} = createRemoteRuntimeTransportMocks({
|
||||
getCallbacks: () => callbacks,
|
||||
setCallbacks: (value) => {
|
||||
callbacks = value
|
||||
},
|
||||
getResolvedPaneHandle: () => resolvedHandle,
|
||||
setResolvedPaneHandle: (value) => {
|
||||
resolvedHandle = value
|
||||
}
|
||||
})
|
||||
|
||||
beforeEach(resetRemoteRuntimeTransport)
|
||||
|
||||
it('sends typing from a restored screen after the host finishes resolving its original pane', async () => {
|
||||
const resolution = createDeferred<void>()
|
||||
/** Real hosts publish `subscribed` and the initial snapshot that completes the attach. */
|
||||
async function completeAttach(): Promise<void> {
|
||||
await vi.waitFor(() => expect(latestSubscribePayload()).toBeDefined())
|
||||
const { streamId } = latestSubscribePayload()
|
||||
callbacks?.onResponse({ ok: true, result: { type: 'subscribed', streamId } })
|
||||
emitSnapshot(streamId, 'prompt$ ')
|
||||
}
|
||||
|
||||
/** Holds `terminal.create` open until the spec settles it with the handle the host returns. */
|
||||
async function createGatedTransport() {
|
||||
const created = createDeferred<string | Error>()
|
||||
// Acknowledged writes to a host without input acks go out as `terminal.send` calls.
|
||||
const acceptedSends: string[] = []
|
||||
const call = runtimeCall.getMockImplementation()
|
||||
runtimeCall.mockImplementation(async (request: { method: string }) => {
|
||||
if (request.method === 'terminal.resolvePane') {
|
||||
await resolution.promise
|
||||
runtimeCall.mockImplementation(
|
||||
async (request: { method: string; params?: { text?: string } }) => {
|
||||
if (request.method === 'terminal.send') {
|
||||
acceptedSends.push(request.params?.text ?? '')
|
||||
return { ok: true, result: { send: { accepted: true } } }
|
||||
}
|
||||
if (request.method !== 'terminal.create') {
|
||||
return call?.(request)
|
||||
}
|
||||
const outcome = await created.promise
|
||||
if (outcome instanceof Error) {
|
||||
throw outcome
|
||||
}
|
||||
return { ok: true, result: { terminal: { handle: outcome, isReattach: true } } }
|
||||
}
|
||||
return call?.(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'
|
||||
})
|
||||
try {
|
||||
expect(transport.sendInput('probe-', 'driving')).toBe(true)
|
||||
const connecting = transport.connect({
|
||||
url: '',
|
||||
sessionId: 'remote:env-1@@terminal-1',
|
||||
callbacks: {}
|
||||
})
|
||||
await vi.waitFor(() => {
|
||||
const transport = createRemoteRuntimePtyTransport('env-1', { worktreeId: 'wt-1' })
|
||||
const waitForCreate = () =>
|
||||
vi.waitFor(() => {
|
||||
expect(runtimeCall).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ method: 'terminal.resolvePane' })
|
||||
expect.objectContaining({ method: 'terminal.create' })
|
||||
)
|
||||
})
|
||||
expect(transport.getPtyId()).toBeNull()
|
||||
expect(transport.sendInput('cold-parked\r', 'driving')).toBe(true)
|
||||
expect(latestFrameForOpcode(TerminalStreamOpcode.Input)).toBeUndefined()
|
||||
resolution.resolve()
|
||||
await connecting
|
||||
await vi.waitFor(() => {
|
||||
const frame = latestFrameForOpcode(TerminalStreamOpcode.Input)
|
||||
expect(frame && decodeTerminalStreamText(frame.payload)).toBe('probe-cold-parked\r')
|
||||
return { transport, created, waitForCreate, acceptedSends }
|
||||
}
|
||||
|
||||
describe('remote reattach type-ahead', () => {
|
||||
it('sends typing from a restored screen after the host finishes resolving its original pane', async () => {
|
||||
const resolution = createDeferred<void>()
|
||||
const call = runtimeCall.getMockImplementation()
|
||||
runtimeCall.mockImplementation(async (request: { method: string }) => {
|
||||
if (request.method === 'terminal.resolvePane') {
|
||||
await resolution.promise
|
||||
}
|
||||
return call?.(request)
|
||||
})
|
||||
} finally {
|
||||
resolution.resolve()
|
||||
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'
|
||||
})
|
||||
try {
|
||||
expect(transport.sendInput('probe-', 'driving')).toBe(true)
|
||||
const connecting = transport.connect({ url: '', sessionId: ORIGINAL_ID, callbacks: {} })
|
||||
await vi.waitFor(() => {
|
||||
expect(runtimeCall).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ method: 'terminal.resolvePane' })
|
||||
)
|
||||
})
|
||||
expect(transport.getPtyId()).toBeNull()
|
||||
expect(transport.sendInput('cold-parked\r', 'driving')).toBe(true)
|
||||
expect(latestFrameForOpcode(TerminalStreamOpcode.Input)).toBeUndefined()
|
||||
resolution.resolve()
|
||||
await connecting
|
||||
await completeAttach()
|
||||
await vi.waitFor(() => {
|
||||
const frame = latestFrameForOpcode(TerminalStreamOpcode.Input)
|
||||
expect(frame && decodeTerminalStreamText(frame.payload)).toBe('probe-cold-parked\r')
|
||||
})
|
||||
} finally {
|
||||
resolution.resolve()
|
||||
transport.destroy?.()
|
||||
}
|
||||
})
|
||||
|
||||
it('delivers typing and an acknowledged interrupt from before and during the reattach', async () => {
|
||||
const { transport, created, waitForCreate, acceptedSends } = await createGatedTransport()
|
||||
expect(transport.sendInput('before', 'driving')).toBe(true)
|
||||
const accepted = transport.sendInputAccepted?.('\x03', 'driving')
|
||||
const connecting = transport.connect({ url: '', sessionId: ORIGINAL_ID, callbacks: {} })
|
||||
await waitForCreate()
|
||||
expect(transport.sendInput('after\r', 'driving')).toBe(true)
|
||||
expect(inputFrameTexts()).toEqual([])
|
||||
created.resolve('terminal-1')
|
||||
await connecting
|
||||
await completeAttach()
|
||||
await expect(accepted).resolves.toBe(true)
|
||||
// The interrupt carries the typing queued ahead of it, so order holds across both paths.
|
||||
expect(acceptedSends).toEqual(['before\x03'])
|
||||
await vi.waitFor(() => expect(inputFrameTexts()).toEqual(['after\r']))
|
||||
transport.destroy?.()
|
||||
}
|
||||
})
|
||||
|
||||
it.each([undefined, 'remote:env-2@@terminal-1', 'remote:terminal-1'])(
|
||||
'discards pre-entry typing when the first connect does not restore this pane: %s',
|
||||
async (sessionId) => {
|
||||
const { transport, created, waitForCreate } = await createGatedTransport()
|
||||
const accepted = transport.sendInputAccepted?.('old tail\r', 'driving')
|
||||
const connecting = transport.connect({ url: '', sessionId, callbacks: {} })
|
||||
await expect(accepted).resolves.toBe(false)
|
||||
await waitForCreate()
|
||||
created.resolve('terminal-1')
|
||||
await connecting
|
||||
await flushAsyncTicks()
|
||||
expect(inputFrameTexts()).toEqual([])
|
||||
transport.destroy?.()
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['terminal-replacement', new Error('offline')])(
|
||||
'never forwards retained input when the original terminal is unavailable: %s',
|
||||
async (outcome) => {
|
||||
const { transport, created, waitForCreate } = await createGatedTransport()
|
||||
const connecting = transport.connect({ url: '', sessionId: ORIGINAL_ID, callbacks: {} })
|
||||
await waitForCreate()
|
||||
const accepted = transport.sendInputAccepted?.('dangerous tail\r', 'driving')
|
||||
expect(transport.sendInput('\x1b[I', 'query-reply')).toBe(false)
|
||||
expect(transport.sendInputImmediate('\x1b[1;1R')).toBe(false)
|
||||
created.resolve(outcome)
|
||||
await connecting
|
||||
if (typeof outcome === 'string') {
|
||||
await completeAttach()
|
||||
}
|
||||
await expect(accepted).resolves.toBe(false)
|
||||
await flushAsyncTicks()
|
||||
expect(inputFrameTexts()).toEqual([])
|
||||
transport.destroy?.()
|
||||
}
|
||||
)
|
||||
|
||||
it.each([
|
||||
['terminal-1', 'tail-on-old-pixels\r'],
|
||||
['terminal-replacement', '']
|
||||
])(
|
||||
'keeps typing between binding %s and its first snapshot scoped to the restored terminal',
|
||||
async (boundHandle, delivered) => {
|
||||
const subscribed = createDeferred<void>()
|
||||
const subscribe = runtimeSubscribe.getMockImplementation()
|
||||
runtimeSubscribe.mockImplementation(async (...args: unknown[]) => {
|
||||
await subscribed.promise
|
||||
return subscribe?.(...args)
|
||||
})
|
||||
const { transport, created, waitForCreate } = await createGatedTransport()
|
||||
const connecting = transport.connect({ url: '', sessionId: ORIGINAL_ID, callbacks: {} })
|
||||
await waitForCreate()
|
||||
created.resolve(boundHandle)
|
||||
await vi.waitFor(() => expect(runtimeSubscribe).toHaveBeenCalled())
|
||||
expect(transport.sendInput('tail-on-old-pixels\r', 'driving')).toBe(true)
|
||||
subscribed.resolve()
|
||||
await connecting
|
||||
await completeAttach()
|
||||
// A key typed on the new snapshot proves everything held ahead of it has settled.
|
||||
expect(transport.sendInput('on-new-snapshot', 'driving')).toBe(true)
|
||||
await vi.waitFor(() => expect(inputFrameTexts().join('')).toMatch(/on-new-snapshot$/))
|
||||
expect(inputFrameTexts().join('')).toBe(`${delivered}on-new-snapshot`)
|
||||
transport.destroy?.()
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['attach', 'disconnect', 'detach', 'destroy'] as const)(
|
||||
'drops input and settles accepted writes on %s before a late reattach',
|
||||
async (action) => {
|
||||
const { transport, created, waitForCreate } = await createGatedTransport()
|
||||
const preEntry = transport.sendInputAccepted?.('pre-entry\r', 'driving')
|
||||
if (action === 'attach') {
|
||||
transport.attach({ existingPtyId: ORIGINAL_ID, callbacks: {} })
|
||||
await expect(preEntry).resolves.toBe(false)
|
||||
transport.destroy?.()
|
||||
return
|
||||
}
|
||||
const connecting = transport.connect({ url: '', sessionId: ORIGINAL_ID, callbacks: {} })
|
||||
await waitForCreate()
|
||||
const accepted = transport.sendInputAccepted?.('stale\r', 'driving')
|
||||
transport[action]?.()
|
||||
await expect(preEntry).resolves.toBe(false)
|
||||
await expect(accepted).resolves.toBe(false)
|
||||
created.resolve('terminal-1')
|
||||
await connecting
|
||||
await flushAsyncTicks()
|
||||
expect(inputFrameTexts()).toEqual([])
|
||||
transport.destroy?.()
|
||||
}
|
||||
)
|
||||
})
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import { readTerminalSpawnDeferral } from '../../../../shared/host-spawn-admission'
|
||||
import { remoteRestoreSpawnIntentParams } from '../../runtime/remote-restore-spawn-intent'
|
||||
import { createAgentSessionKeyboardOptions } from '@/runtime/agent-session-keyboard-capability'
|
||||
import { withRemoteReattachInputBuffer } from './remote-reattach-input-buffer'
|
||||
import { createRemoteRuntimeDisconnectedInputGrace } from './remote-runtime-disconnected-input-grace'
|
||||
import {
|
||||
createRemoteRuntimeRecoveryInputHold,
|
||||
@@ -314,7 +313,7 @@ export function createRemoteRuntimePtyTransport(
|
||||
})
|
||||
}
|
||||
|
||||
const recoveryInputHold = createRemoteRuntimeRecoveryInputHold()
|
||||
const recoveryInputHold = createRemoteRuntimeRecoveryInputHold({ holdBeforeFirstConnect: true })
|
||||
const inputJournal = createRemoteRuntimeInputJournal({
|
||||
currentStream: sequencedInputTarget,
|
||||
// Why: a refusal that never clears would drop every keystroke behind a connected-looking pane.
|
||||
@@ -982,7 +981,7 @@ export function createRemoteRuntimePtyTransport(
|
||||
return true
|
||||
}
|
||||
if (lastConnectOptions) {
|
||||
void connectForRecovery(lastConnectOptions)
|
||||
void connectHoldingRestoredInput(lastConnectOptions)
|
||||
return true
|
||||
}
|
||||
return false
|
||||
@@ -1498,9 +1497,12 @@ export function createRemoteRuntimePtyTransport(
|
||||
// Why: a pane binding or auto-recovering a known handle holds typing for it instead of dropping it (#25784).
|
||||
// Why 'disconnected' too: the same handle still reattaches on its own after the window, within the grace.
|
||||
function shouldHoldInput(): boolean {
|
||||
if (destroyed || terminalEnded || handle === null) {
|
||||
if (destroyed || terminalEnded) {
|
||||
return false
|
||||
}
|
||||
if (handle === null) {
|
||||
return recoveryInputHold.isHoldingBeforeBind()
|
||||
}
|
||||
if (recovery.currentPhase === 'disconnected') {
|
||||
return !discardExpiredDisconnectedInput()
|
||||
}
|
||||
@@ -1522,8 +1524,8 @@ export function createRemoteRuntimePtyTransport(
|
||||
|
||||
// Why: these bytes never reached a stream, so holding them for the reattach cannot duplicate them.
|
||||
function holdUnsentInput(text: string): void {
|
||||
if (handle && shouldHoldInput()) {
|
||||
recoveryInputHold.enqueue(heldInputEndpoint(handle), text, 'driving')
|
||||
if (shouldHoldInput()) {
|
||||
recoveryInputHold.enqueue(holdEndpoint(), text, 'driving')
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1549,10 +1551,9 @@ export function createRemoteRuntimePtyTransport(
|
||||
inputBatcher.enqueueAfterValidation(() => {
|
||||
ran = true
|
||||
inputBatcher.flush()
|
||||
const held = handle && shouldHoldInput() ? heldInputEndpoint(handle) : null
|
||||
resolve(
|
||||
held
|
||||
? recoveryInputHold.enqueueAccepted(held, data, inputKind, options)
|
||||
shouldHoldInput()
|
||||
? recoveryInputHold.enqueueAccepted(holdEndpoint(), data, inputKind, options)
|
||||
: sendInputAcceptedToRuntime(data, inputKind, options)
|
||||
)
|
||||
})
|
||||
@@ -1594,15 +1595,26 @@ export function createRemoteRuntimePtyTransport(
|
||||
}
|
||||
}
|
||||
|
||||
// Why null: typing on a restored screen stays scoped to the terminal its connect expects, even
|
||||
// after the host binds a replacement and before the connect settles (#10065).
|
||||
function holdEndpoint(): RemoteRuntimeInputEndpoint | null {
|
||||
return handle && !recoveryInputHold.isHoldingBeforeBind() ? heldInputEndpoint(handle) : null
|
||||
}
|
||||
|
||||
// Why: releases a stale hold first, so the hold check reads the handle that release left bound.
|
||||
function releaseThenHoldEndpoint(
|
||||
inputKind: TerminalInputKind
|
||||
): RemoteRuntimeInputEndpoint | null {
|
||||
if (recoveryInputHold.isHolding()) {
|
||||
function releaseThenShouldHold(inputKind: TerminalInputKind): boolean {
|
||||
if (recoveryInputHold.isHolding() && !recoveryInputHold.isHoldingBeforeBind()) {
|
||||
releaseHeldInput()
|
||||
}
|
||||
return handle && inputKind !== 'query-reply' && shouldHoldInput()
|
||||
? heldInputEndpoint(handle)
|
||||
return inputKind !== 'query-reply' && shouldHoldInput()
|
||||
}
|
||||
|
||||
// Why: only a connect naming this transport's own remote id restores a terminal typing may reach.
|
||||
function restoredInputEndpoint(sessionId: string | undefined): RemoteRuntimeInputEndpoint | null {
|
||||
const restoredHandle = sessionId ? getRemoteRuntimeTerminalHandle(sessionId) : null
|
||||
return restoredHandle &&
|
||||
toRemoteRuntimePtyId(restoredHandle, currentRuntimeEnvironmentId) === sessionId
|
||||
? { handle: restoredHandle, incarnationId: null }
|
||||
: null
|
||||
}
|
||||
|
||||
@@ -2633,275 +2645,284 @@ export function createRemoteRuntimePtyTransport(
|
||||
return sendUnacknowledgedInput(data, true)
|
||||
}
|
||||
|
||||
let connectForRecovery: PtyTransport['connect'] = (options) => transport.connect(options)
|
||||
const transport: PtyTransport = {
|
||||
async connect(options) {
|
||||
cancelTerminalCreateRetryWait()
|
||||
const connectLifecycleEpoch = ++lifecycleEpoch
|
||||
const createEnvironmentId = currentRuntimeEnvironmentId
|
||||
lastConnectOptions = options
|
||||
lastAttachOptions = null
|
||||
storedCallbacks = options.callbacks
|
||||
resetRecoveryReplacementPolicy()
|
||||
resetSameHandleEndReuse()
|
||||
terminalEnded = false
|
||||
connecting = true
|
||||
emitRecoveryState(true)
|
||||
if (destroyed || !worktreeId) {
|
||||
return
|
||||
// Why: typing on a restored screen reaches its original terminal only, never a fresh create.
|
||||
async function connectHoldingRestoredInput(
|
||||
options: Parameters<PtyTransport['connect']>[0]
|
||||
): Promise<PtyConnectResult | undefined> {
|
||||
const expected = recoveryInputHold.expect(restoredInputEndpoint(options.sessionId))
|
||||
try {
|
||||
return await connectPane(options)
|
||||
} finally {
|
||||
recoveryInputHold.settleExpectation(expected, handle !== null && !destroyed)
|
||||
if (handle && recoveryInputHold.isHolding()) {
|
||||
releaseHeldInput()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function connectPane(
|
||||
options: Parameters<PtyTransport['connect']>[0]
|
||||
): Promise<PtyConnectResult | undefined> {
|
||||
cancelTerminalCreateRetryWait()
|
||||
const connectLifecycleEpoch = ++lifecycleEpoch
|
||||
const createEnvironmentId = currentRuntimeEnvironmentId
|
||||
lastConnectOptions = options
|
||||
lastAttachOptions = null
|
||||
storedCallbacks = options.callbacks
|
||||
resetRecoveryReplacementPolicy()
|
||||
resetSameHandleEndReuse()
|
||||
terminalEnded = false
|
||||
connecting = true
|
||||
emitRecoveryState(true)
|
||||
if (destroyed || !worktreeId) {
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
if (isWebTerminalSurfaceTabId(tabId ?? '')) {
|
||||
// Reattach callbacks must not publish the replacement as a fresh
|
||||
// spawn before the reattach result commits tab/layout ownership.
|
||||
return await attachHostSessionMirror(
|
||||
options,
|
||||
!options.sessionId,
|
||||
undefined,
|
||||
connectLifecycleEpoch
|
||||
)
|
||||
}
|
||||
|
||||
try {
|
||||
if (isWebTerminalSurfaceTabId(tabId ?? '')) {
|
||||
// Reattach callbacks must not publish the replacement as a fresh
|
||||
// spawn before the reattach result commits tab/layout ownership.
|
||||
return await attachHostSessionMirror(
|
||||
options,
|
||||
!options.sessionId,
|
||||
undefined,
|
||||
connectLifecycleEpoch
|
||||
)
|
||||
if (options.sessionId && !getRemoteRuntimeTerminalHandle(options.sessionId)) {
|
||||
// Why: a HUB session persists host-native PTY ids; resolve its pane handle without exposing that SSH identity as a client transport id.
|
||||
const resolution = await resolvePersistedHostPane()
|
||||
if (resolution.kind === 'resolved') {
|
||||
return await adoptResolvedHostPane(resolution.terminal, options)
|
||||
}
|
||||
}
|
||||
|
||||
if (options.sessionId && !getRemoteRuntimeTerminalHandle(options.sessionId)) {
|
||||
// Why: a HUB session persists host-native PTY ids; resolve its pane handle without exposing that SSH identity as a client transport id.
|
||||
const resolution = await resolvePersistedHostPane()
|
||||
if (resolution.kind === 'resolved') {
|
||||
return await adoptResolvedHostPane(resolution.terminal, options)
|
||||
}
|
||||
}
|
||||
|
||||
const commandToSend = options.command ?? command
|
||||
const startupCommandDeliveryToSend =
|
||||
options.startupCommandDelivery ?? startupCommandDelivery
|
||||
const envToSend = options.env ?? env
|
||||
const envToDeleteToSend = options.envToDelete ?? envToDelete
|
||||
const launchConfigToSend = options.launchConfig ?? launchConfig
|
||||
const resumeProviderSessionToSend = options.resumeProviderSession ?? resumeProviderSession
|
||||
const launchTokenToSend = options.launchToken ?? launchToken
|
||||
const launchAgentToSend = options.launchAgent ?? launchAgent
|
||||
const restoreSpawnIntent = resumeProviderSessionToSend
|
||||
? options.resumeSpawnIntent
|
||||
: undefined
|
||||
// Why only for a restore: an extra await would reorder every plain create against destroy.
|
||||
const spawnIntentParams = restoreSpawnIntent
|
||||
? await remoteRestoreSpawnIntentParams(createEnvironmentId, restoreSpawnIntent)
|
||||
: {}
|
||||
if (restoreSpawnIntent && (destroyed || lifecycleEpoch !== connectLifecycleEpoch)) {
|
||||
return
|
||||
}
|
||||
const legacyCreateParams = {
|
||||
worktree: toRuntimeTerminalWorktreeSelector(worktreeId),
|
||||
clientMutationId: terminalCreateMutationId,
|
||||
...(commandToSend !== undefined ? { command: commandToSend } : {}),
|
||||
...(startupCommandDeliveryToSend !== undefined
|
||||
? { startupCommandDelivery: startupCommandDeliveryToSend }
|
||||
: {}),
|
||||
...(envToSend !== undefined ? { env: envToSend } : {}),
|
||||
...(envToDeleteToSend !== undefined ? { envToDelete: envToDeleteToSend } : {}),
|
||||
...(launchConfigToSend !== undefined ? { launchConfig: launchConfigToSend } : {}),
|
||||
...(resumeProviderSessionToSend !== undefined
|
||||
? { resumeProviderSession: resumeProviderSessionToSend }
|
||||
: {}),
|
||||
...(launchTokenToSend !== undefined ? { launchToken: launchTokenToSend } : {}),
|
||||
...(launchAgentToSend !== undefined ? { launchAgent: launchAgentToSend } : {}),
|
||||
...spawnIntentParams,
|
||||
...(terminalColorQueryReplies ? { terminalColorQueryReplies } : {}),
|
||||
...(terminalKittyKeyboardProtocol === true
|
||||
? { terminalKittyKeyboardProtocol: true }
|
||||
: {}),
|
||||
tabId,
|
||||
leafId,
|
||||
focus: false,
|
||||
// Why: transport backs an already-mounted pane; activation is local state, not permission for remote UI reveal.
|
||||
presentation: 'background' as const,
|
||||
...(activate === true ? { activate: true } : {})
|
||||
}
|
||||
const legacyCreate = () =>
|
||||
createWithUnknownOutcomeRecovery(
|
||||
'terminal',
|
||||
(timeoutMs, reconcileExisting) =>
|
||||
callRuntimeForEnvironment<{ terminal: RuntimeTerminalCreate }>(
|
||||
createEnvironmentId,
|
||||
'terminal.create',
|
||||
{
|
||||
...legacyCreateParams,
|
||||
...(reconcileExisting ? { reconcileExisting: true } : {})
|
||||
},
|
||||
timeoutMs
|
||||
),
|
||||
createEnvironmentId,
|
||||
connectLifecycleEpoch
|
||||
)
|
||||
const hostAuthorityCreate = async () => {
|
||||
const keyboardOptions = await agentKeyboardOptions(createEnvironmentId)
|
||||
return createWithUnknownOutcomeRecovery(
|
||||
'agent-session',
|
||||
(timeoutMs) =>
|
||||
resumeProviderSessionToSend
|
||||
? callRuntimeForEnvironment<RuntimeEnsureAgentSessionResult>(
|
||||
createEnvironmentId,
|
||||
'terminal.ensureAgentSession',
|
||||
const commandToSend = options.command ?? command
|
||||
const startupCommandDeliveryToSend = options.startupCommandDelivery ?? startupCommandDelivery
|
||||
const envToSend = options.env ?? env
|
||||
const envToDeleteToSend = options.envToDelete ?? envToDelete
|
||||
const launchConfigToSend = options.launchConfig ?? launchConfig
|
||||
const resumeProviderSessionToSend = options.resumeProviderSession ?? resumeProviderSession
|
||||
const launchTokenToSend = options.launchToken ?? launchToken
|
||||
const launchAgentToSend = options.launchAgent ?? launchAgent
|
||||
const restoreSpawnIntent = resumeProviderSessionToSend ? options.resumeSpawnIntent : undefined
|
||||
// Why only for a restore: an extra await would reorder every plain create against destroy.
|
||||
const spawnIntentParams = restoreSpawnIntent
|
||||
? await remoteRestoreSpawnIntentParams(createEnvironmentId, restoreSpawnIntent)
|
||||
: {}
|
||||
if (restoreSpawnIntent && (destroyed || lifecycleEpoch !== connectLifecycleEpoch)) {
|
||||
return
|
||||
}
|
||||
const legacyCreateParams = {
|
||||
worktree: toRuntimeTerminalWorktreeSelector(worktreeId),
|
||||
clientMutationId: terminalCreateMutationId,
|
||||
...(commandToSend !== undefined ? { command: commandToSend } : {}),
|
||||
...(startupCommandDeliveryToSend !== undefined
|
||||
? { startupCommandDelivery: startupCommandDeliveryToSend }
|
||||
: {}),
|
||||
...(envToSend !== undefined ? { env: envToSend } : {}),
|
||||
...(envToDeleteToSend !== undefined ? { envToDelete: envToDeleteToSend } : {}),
|
||||
...(launchConfigToSend !== undefined ? { launchConfig: launchConfigToSend } : {}),
|
||||
...(resumeProviderSessionToSend !== undefined
|
||||
? { resumeProviderSession: resumeProviderSessionToSend }
|
||||
: {}),
|
||||
...(launchTokenToSend !== undefined ? { launchToken: launchTokenToSend } : {}),
|
||||
...(launchAgentToSend !== undefined ? { launchAgent: launchAgentToSend } : {}),
|
||||
...spawnIntentParams,
|
||||
...(terminalColorQueryReplies ? { terminalColorQueryReplies } : {}),
|
||||
...(terminalKittyKeyboardProtocol === true ? { terminalKittyKeyboardProtocol: true } : {}),
|
||||
tabId,
|
||||
leafId,
|
||||
focus: false,
|
||||
// Why: transport backs an already-mounted pane; activation is local state, not permission for remote UI reveal.
|
||||
presentation: 'background' as const,
|
||||
...(activate === true ? { activate: true } : {})
|
||||
}
|
||||
const legacyCreate = () =>
|
||||
createWithUnknownOutcomeRecovery(
|
||||
'terminal',
|
||||
(timeoutMs, reconcileExisting) =>
|
||||
callRuntimeForEnvironment<{ terminal: RuntimeTerminalCreate }>(
|
||||
createEnvironmentId,
|
||||
'terminal.create',
|
||||
{
|
||||
...legacyCreateParams,
|
||||
...(reconcileExisting ? { reconcileExisting: true } : {})
|
||||
},
|
||||
timeoutMs
|
||||
),
|
||||
createEnvironmentId,
|
||||
connectLifecycleEpoch
|
||||
)
|
||||
const hostAuthorityCreate = async () => {
|
||||
const keyboardOptions = await agentKeyboardOptions(createEnvironmentId)
|
||||
return createWithUnknownOutcomeRecovery(
|
||||
'agent-session',
|
||||
(timeoutMs) =>
|
||||
resumeProviderSessionToSend
|
||||
? callRuntimeForEnvironment<RuntimeEnsureAgentSessionResult>(
|
||||
createEnvironmentId,
|
||||
'terminal.ensureAgentSession',
|
||||
{
|
||||
kind: 'explicit',
|
||||
...keyboardOptions,
|
||||
worktree: toRuntimeTerminalWorktreeSelector(worktreeId),
|
||||
agent: launchAgentToSend!,
|
||||
providerSession: resumeProviderSessionToSend,
|
||||
...(launchConfigToSend?.ompResumeFilePath
|
||||
? { ompResumeFilePath: launchConfigToSend.ompResumeFilePath }
|
||||
: {}),
|
||||
...(agentArgsOverride !== undefined ? { agentArgs: agentArgsOverride } : {}),
|
||||
...(agentLaunchPreferences
|
||||
? { launchPreferences: agentLaunchPreferences }
|
||||
: {}),
|
||||
placement: { tabId, leafId },
|
||||
presentation: 'background',
|
||||
...spawnIntentParams
|
||||
},
|
||||
timeoutMs
|
||||
)
|
||||
: callRuntimeForEnvironment<RuntimeCreateAgentSessionResult>(
|
||||
createEnvironmentId,
|
||||
'terminal.createAgentSession',
|
||||
withAgentSessionCreateOperationId(
|
||||
{
|
||||
kind: 'explicit',
|
||||
...keyboardOptions,
|
||||
worktree: toRuntimeTerminalWorktreeSelector(worktreeId),
|
||||
agent: launchAgentToSend!,
|
||||
providerSession: resumeProviderSessionToSend,
|
||||
...(launchConfigToSend?.ompResumeFilePath
|
||||
? { ompResumeFilePath: launchConfigToSend.ompResumeFilePath }
|
||||
: {}),
|
||||
...(agentPrompt ? { prompt: agentPrompt } : {}),
|
||||
...(agentPromptDelivery ? { promptDelivery: agentPromptDelivery } : {}),
|
||||
...(agentArgsOverride !== undefined ? { agentArgs: agentArgsOverride } : {}),
|
||||
...(agentLaunchPreferences
|
||||
? { launchPreferences: agentLaunchPreferences }
|
||||
: {}),
|
||||
placement: { tabId, leafId },
|
||||
presentation: 'background',
|
||||
...spawnIntentParams
|
||||
presentation: 'background'
|
||||
},
|
||||
timeoutMs
|
||||
)
|
||||
: callRuntimeForEnvironment<RuntimeCreateAgentSessionResult>(
|
||||
createEnvironmentId,
|
||||
'terminal.createAgentSession',
|
||||
withAgentSessionCreateOperationId(
|
||||
{
|
||||
...keyboardOptions,
|
||||
worktree: toRuntimeTerminalWorktreeSelector(worktreeId),
|
||||
agent: launchAgentToSend!,
|
||||
...(agentPrompt ? { prompt: agentPrompt } : {}),
|
||||
...(agentPromptDelivery ? { promptDelivery: agentPromptDelivery } : {}),
|
||||
...(agentArgsOverride !== undefined
|
||||
? { agentArgs: agentArgsOverride }
|
||||
: {}),
|
||||
...(agentLaunchPreferences
|
||||
? { launchPreferences: agentLaunchPreferences }
|
||||
: {}),
|
||||
placement: { tabId, leafId },
|
||||
presentation: 'background'
|
||||
},
|
||||
agentCreateOperation.clientOperationId
|
||||
),
|
||||
timeoutMs
|
||||
agentCreateOperation.clientOperationId
|
||||
),
|
||||
createEnvironmentId,
|
||||
connectLifecycleEpoch
|
||||
)
|
||||
timeoutMs
|
||||
),
|
||||
createEnvironmentId,
|
||||
connectLifecycleEpoch
|
||||
)
|
||||
}
|
||||
const resumeHostAuthorityCapability = resumeProviderSessionToSend
|
||||
? agentResumeHostAuthorityCapability(launchAgentToSend)
|
||||
: undefined
|
||||
const created = launchAgentToSend
|
||||
? agentSessionRequiresHostAuthorityReplay
|
||||
? await hostAuthorityCreate()
|
||||
: await runRemoteAgentSessionLaunch<RemoteAgentSessionLaunchResult | null>({
|
||||
environmentId: createEnvironmentId,
|
||||
hostAuthority: hostAuthorityCreate,
|
||||
...(resumeHostAuthorityCapability
|
||||
? { hostAuthorityCapability: resumeHostAuthorityCapability }
|
||||
: {}),
|
||||
legacy: legacyCreate
|
||||
})
|
||||
: await legacyCreate()
|
||||
if (!created) {
|
||||
if (!destroyed && lifecycleEpoch === connectLifecycleEpoch) {
|
||||
connecting = false
|
||||
recovery.markDisconnected()
|
||||
}
|
||||
const resumeHostAuthorityCapability = resumeProviderSessionToSend
|
||||
? agentResumeHostAuthorityCapability(launchAgentToSend)
|
||||
: undefined
|
||||
const created = launchAgentToSend
|
||||
? agentSessionRequiresHostAuthorityReplay
|
||||
? await hostAuthorityCreate()
|
||||
: await runRemoteAgentSessionLaunch<RemoteAgentSessionLaunchResult | null>({
|
||||
environmentId: createEnvironmentId,
|
||||
hostAuthority: hostAuthorityCreate,
|
||||
...(resumeHostAuthorityCapability
|
||||
? { hostAuthorityCapability: resumeHostAuthorityCapability }
|
||||
: {}),
|
||||
legacy: legacyCreate
|
||||
})
|
||||
: await legacyCreate()
|
||||
if (!created) {
|
||||
if (!destroyed && lifecycleEpoch === connectLifecycleEpoch) {
|
||||
connecting = false
|
||||
recovery.markDisconnected()
|
||||
}
|
||||
return
|
||||
}
|
||||
const createdTerminal = created.terminal
|
||||
adoptExecutionMetadata(createdTerminal)
|
||||
if (created.disposition !== undefined && tabId && createdTerminal.tabId) {
|
||||
recordWebAgentSessionHandoff({
|
||||
environmentId: createEnvironmentId,
|
||||
worktreeId,
|
||||
return
|
||||
}
|
||||
const createdTerminal = created.terminal
|
||||
adoptExecutionMetadata(createdTerminal)
|
||||
if (created.disposition !== undefined && tabId && createdTerminal.tabId) {
|
||||
recordWebAgentSessionHandoff({
|
||||
environmentId: createEnvironmentId,
|
||||
worktreeId,
|
||||
provisionalTabId: tabId,
|
||||
hostTabId: createdTerminal.tabId,
|
||||
hostTerminalHandle: createdTerminal.handle
|
||||
})
|
||||
// Snapshot parity must not delay attachment to a terminal the host already created.
|
||||
void refreshWebRuntimeSessionTabsSnapshot(createEnvironmentId, worktreeId, {
|
||||
expectedEnvironmentPairingRevision: currentPaneRevision(),
|
||||
acceptCurrentSnapshot: true,
|
||||
confirmAgentSessionHandoff: {
|
||||
provisionalTabId: tabId,
|
||||
hostTabId: createdTerminal.tabId,
|
||||
hostTerminalHandle: createdTerminal.handle
|
||||
})
|
||||
// Snapshot parity must not delay attachment to a terminal the host already created.
|
||||
void refreshWebRuntimeSessionTabsSnapshot(createEnvironmentId, worktreeId, {
|
||||
expectedEnvironmentPairingRevision: currentPaneRevision(),
|
||||
acceptCurrentSnapshot: true,
|
||||
confirmAgentSessionHandoff: {
|
||||
provisionalTabId: tabId,
|
||||
hostTabId: createdTerminal.tabId,
|
||||
hostTerminalHandle: createdTerminal.handle
|
||||
}
|
||||
})
|
||||
}
|
||||
if (destroyed || lifecycleEpoch !== connectLifecycleEpoch) {
|
||||
if (
|
||||
!hostSnapshotOwnsLaunch(created, createEnvironmentId) &&
|
||||
(createdTerminal.handle !== handle ||
|
||||
createEnvironmentId !== currentRuntimeEnvironmentId)
|
||||
) {
|
||||
await closeRemoteTerminal(createdTerminal.handle, createEnvironmentId)
|
||||
}
|
||||
return
|
||||
}
|
||||
handle = createdTerminal.handle
|
||||
|
||||
if (createdTerminal.isReattach === true) {
|
||||
storedCallbacks.onReattachDetermined?.()
|
||||
}
|
||||
remotePtyId = toRemoteRuntimePtyId(handle, currentRuntimeEnvironmentId)
|
||||
registerShutdownHandlers(remotePtyId)
|
||||
connected = true
|
||||
desiredViewport = {
|
||||
cols: options.cols ?? 80,
|
||||
rows: options.rows ?? 24
|
||||
}
|
||||
if (createdTerminal.isReattach !== true) {
|
||||
onPtySpawn?.(remotePtyId)
|
||||
}
|
||||
emitRecoveryState()
|
||||
|
||||
try {
|
||||
await subscribeToHandle()
|
||||
} catch (error) {
|
||||
if (!recoverAfterSubscribeFailure(error, handle, remotePtyId)) {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
if (destroyed || !connected || !remotePtyId) {
|
||||
return
|
||||
}
|
||||
|
||||
return {
|
||||
id: remotePtyId,
|
||||
replay: '',
|
||||
...(authoritativePtyIncarnationId
|
||||
? { incarnationId: authoritativePtyIncarnationId }
|
||||
: {}),
|
||||
...(createdTerminal.isReattach === true ? { isReattach: true } : {})
|
||||
} satisfies PtyConnectResult
|
||||
} catch (error) {
|
||||
if (!destroyed && lifecycleEpoch === connectLifecycleEpoch) {
|
||||
connecting = false
|
||||
const message = runtimeTerminalErrorMessage(error)
|
||||
const deferral = readTerminalSpawnDeferral(error)
|
||||
if (deferral) {
|
||||
recovery.cancel()
|
||||
emitRecoveryState()
|
||||
storedCallbacks.onSpawnDeferred?.(deferral)
|
||||
} else if (isRemoteTerminalGoneMessage(message)) {
|
||||
recovery.cancel()
|
||||
handleRemoteTerminalError(error)
|
||||
} else if (isRecoverablePaneBindingError(error)) {
|
||||
scheduleConnectRetryAfterRecoverableFailure()
|
||||
} else {
|
||||
recovery.cancel()
|
||||
emitRecoveryState()
|
||||
surfaceErrorMessage(message)
|
||||
}
|
||||
}
|
||||
return undefined
|
||||
})
|
||||
}
|
||||
},
|
||||
if (destroyed || lifecycleEpoch !== connectLifecycleEpoch) {
|
||||
if (
|
||||
!hostSnapshotOwnsLaunch(created, createEnvironmentId) &&
|
||||
(createdTerminal.handle !== handle || createEnvironmentId !== currentRuntimeEnvironmentId)
|
||||
) {
|
||||
await closeRemoteTerminal(createdTerminal.handle, createEnvironmentId)
|
||||
}
|
||||
return
|
||||
}
|
||||
handle = createdTerminal.handle
|
||||
|
||||
if (createdTerminal.isReattach === true) {
|
||||
storedCallbacks.onReattachDetermined?.()
|
||||
}
|
||||
remotePtyId = toRemoteRuntimePtyId(handle, currentRuntimeEnvironmentId)
|
||||
registerShutdownHandlers(remotePtyId)
|
||||
connected = true
|
||||
desiredViewport = {
|
||||
cols: options.cols ?? 80,
|
||||
rows: options.rows ?? 24
|
||||
}
|
||||
if (createdTerminal.isReattach !== true) {
|
||||
onPtySpawn?.(remotePtyId)
|
||||
}
|
||||
emitRecoveryState()
|
||||
|
||||
try {
|
||||
await subscribeToHandle()
|
||||
} catch (error) {
|
||||
if (!recoverAfterSubscribeFailure(error, handle, remotePtyId)) {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
if (destroyed || !connected || !remotePtyId) {
|
||||
return
|
||||
}
|
||||
|
||||
return {
|
||||
id: remotePtyId,
|
||||
replay: '',
|
||||
...(authoritativePtyIncarnationId ? { incarnationId: authoritativePtyIncarnationId } : {}),
|
||||
...(createdTerminal.isReattach === true ? { isReattach: true } : {})
|
||||
} satisfies PtyConnectResult
|
||||
} catch (error) {
|
||||
if (!destroyed && lifecycleEpoch === connectLifecycleEpoch) {
|
||||
connecting = false
|
||||
const message = runtimeTerminalErrorMessage(error)
|
||||
const deferral = readTerminalSpawnDeferral(error)
|
||||
if (deferral) {
|
||||
recovery.cancel()
|
||||
emitRecoveryState()
|
||||
storedCallbacks.onSpawnDeferred?.(deferral)
|
||||
} else if (isRemoteTerminalGoneMessage(message)) {
|
||||
recovery.cancel()
|
||||
handleRemoteTerminalError(error)
|
||||
} else if (isRecoverablePaneBindingError(error)) {
|
||||
scheduleConnectRetryAfterRecoverableFailure()
|
||||
} else {
|
||||
recovery.cancel()
|
||||
emitRecoveryState()
|
||||
surfaceErrorMessage(message)
|
||||
}
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
const transport: PtyTransport = {
|
||||
connect: connectHoldingRestoredInput,
|
||||
|
||||
attach(options) {
|
||||
recoveryInputHold.expect(null)
|
||||
attachPane(options, false)
|
||||
},
|
||||
|
||||
@@ -2968,8 +2989,7 @@ export function createRemoteRuntimePtyTransport(
|
||||
|
||||
// Why: the kind only gates the local hold; terminal.send gets no kind, so the host classifies desktop bytes itself.
|
||||
sendInput(data, inputKind): boolean {
|
||||
const held = releaseThenHoldEndpoint(inputKind)
|
||||
if (held) {
|
||||
if (releaseThenShouldHold(inputKind)) {
|
||||
if (!data) {
|
||||
return true
|
||||
}
|
||||
@@ -2978,7 +2998,7 @@ export function createRemoteRuntimePtyTransport(
|
||||
return inputBatcher.push(data)
|
||||
}
|
||||
inputBatcher.flush()
|
||||
return recoveryInputHold.enqueue(held, data, inputKind)
|
||||
return recoveryInputHold.enqueue(holdEndpoint(), data, inputKind)
|
||||
}
|
||||
return sendInputNow(data)
|
||||
},
|
||||
@@ -2987,13 +3007,12 @@ export function createRemoteRuntimePtyTransport(
|
||||
sendInputImmediate: (data: string): boolean => sendInputImmediateNow(data),
|
||||
|
||||
sendInputAccepted(data, inputKind, options) {
|
||||
const held = releaseThenHoldEndpoint(inputKind)
|
||||
if (held && data) {
|
||||
if (releaseThenShouldHold(inputKind) && data) {
|
||||
if (inputBatcher.hasPendingValidation()) {
|
||||
return holdAcceptedAfterValidation(data, inputKind, options)
|
||||
}
|
||||
inputBatcher.flush()
|
||||
return recoveryInputHold.enqueueAccepted(held, data, inputKind, options)
|
||||
return recoveryInputHold.enqueueAccepted(holdEndpoint(), data, inputKind, options)
|
||||
}
|
||||
return sendInputAcceptedToRuntime(data, inputKind, options)
|
||||
},
|
||||
@@ -3089,7 +3108,7 @@ export function createRemoteRuntimePtyTransport(
|
||||
recovery.currentPhase === 'disconnected'
|
||||
) {
|
||||
recovery.begin()
|
||||
void connectForRecovery(lastConnectOptions)
|
||||
void connectHoldingRestoredInput(lastConnectOptions)
|
||||
return true
|
||||
}
|
||||
// Why: online/resume fires a parked retry; the button must not be weaker than an event (#12684).
|
||||
@@ -3159,9 +3178,6 @@ export function createRemoteRuntimePtyTransport(
|
||||
return stream.serializeBufferOutcome(opts)
|
||||
},
|
||||
|
||||
setConnectForRecovery(connect) {
|
||||
connectForRecovery = connect
|
||||
},
|
||||
destroy() {
|
||||
destroyed = true
|
||||
setAttachmentUnavailable()
|
||||
@@ -3177,5 +3193,5 @@ export function createRemoteRuntimePtyTransport(
|
||||
viewportBatcher.clear()
|
||||
}
|
||||
}
|
||||
return withRemoteReattachInputBuffer(transport)
|
||||
return transport
|
||||
}
|
||||
|
||||
@@ -141,4 +141,20 @@ describe('remote runtime recovery input hold', () => {
|
||||
await expect(accepted).resolves.toBe(false)
|
||||
expect(hold.isHolding()).toBe(false)
|
||||
})
|
||||
|
||||
it('lets only the latest connect settle typing held before the pane bound a terminal', async () => {
|
||||
const hold = createRemoteRuntimeRecoveryInputHold({ holdBeforeFirstConnect: true })
|
||||
const { sent, writer } = createWriter()
|
||||
hold.enqueue(null, 'pre-entry ', 'driving')
|
||||
const superseded = hold.expect(endpoint)
|
||||
const latest = hold.expect(endpoint)
|
||||
hold.enqueue(null, 'typed', 'driving')
|
||||
|
||||
hold.settleExpectation(superseded, false)
|
||||
expect(hold.isHoldingBeforeBind()).toBe(true)
|
||||
hold.settleExpectation(latest, true)
|
||||
hold.release(endpoint, writer)
|
||||
|
||||
await vi.waitFor(() => expect(sent).toEqual(['pre-entry typed']))
|
||||
})
|
||||
})
|
||||
|
||||
@@ -16,19 +16,33 @@ export type RemoteRuntimeInputEndpoint = {
|
||||
type HeldInputWriter = PreconnectInputWriter &
|
||||
Required<Pick<PreconnectInputWriter, 'sendInputAccepted'>>
|
||||
|
||||
// Why null: typing on a restored screen precedes the connect that names its terminal.
|
||||
type HeldEndpoint = RemoteRuntimeInputEndpoint | null
|
||||
|
||||
type HeldInput = {
|
||||
endpoint: HeldEndpoint
|
||||
buffer: PtyPreconnectInputBuffer
|
||||
/** Typed before the pane bound a terminal; settles when the connect that expects it ends. */
|
||||
beforeBind: boolean
|
||||
}
|
||||
|
||||
/** One connect's claim on pre-bind typing; identity decides which connect may settle it. */
|
||||
export type RestoredInputExpectation = { readonly endpoint: HeldEndpoint }
|
||||
|
||||
export type RemoteRuntimeRecoveryInputHold = {
|
||||
isHolding: () => boolean
|
||||
enqueue: (
|
||||
endpoint: RemoteRuntimeInputEndpoint,
|
||||
data: string,
|
||||
inputKind: TerminalInputKind
|
||||
) => boolean
|
||||
isHoldingBeforeBind: () => boolean
|
||||
enqueue: (endpoint: HeldEndpoint, data: string, inputKind: TerminalInputKind) => boolean
|
||||
enqueueAccepted: (
|
||||
endpoint: RemoteRuntimeInputEndpoint,
|
||||
endpoint: HeldEndpoint,
|
||||
data: string,
|
||||
inputKind: TerminalInputKind,
|
||||
options?: AcceptedInputOptions
|
||||
) => Promise<boolean>
|
||||
/** Binds pre-bind typing to the terminal a connect restores; null (a fresh create) drops it. */
|
||||
expect: (endpoint: HeldEndpoint) => RestoredInputExpectation
|
||||
/** Ends the latest connect's expectation: without a bound terminal its typing is dropped. */
|
||||
settleExpectation: (expectation: RestoredInputExpectation, bound: boolean) => void
|
||||
/** Delivers held input to the endpoint it was typed into, or drops it if the pane rebound elsewhere. */
|
||||
release: (bound: RemoteRuntimeInputEndpoint, writer: HeldInputWriter) => void
|
||||
discard: () => void
|
||||
@@ -48,11 +62,17 @@ export function isSameRemoteRuntimeInputEndpoint(
|
||||
}
|
||||
|
||||
/** Keeps keystrokes typed while a remote pane is binding or recovering, scoped to one endpoint. */
|
||||
export function createRemoteRuntimeRecoveryInputHold(): RemoteRuntimeRecoveryInputHold {
|
||||
let held: {
|
||||
endpoint: RemoteRuntimeInputEndpoint
|
||||
buffer: PtyPreconnectInputBuffer
|
||||
} | null = null
|
||||
export function createRemoteRuntimeRecoveryInputHold(
|
||||
options: { holdBeforeFirstConnect?: boolean } = {}
|
||||
): RemoteRuntimeRecoveryInputHold {
|
||||
const newHold = (endpoint: HeldEndpoint, beforeBind: boolean): HeldInput => ({
|
||||
endpoint,
|
||||
buffer: createPtyPreconnectInputBuffer([], { recoveryHold: true }),
|
||||
beforeBind
|
||||
})
|
||||
let held: HeldInput | null = options.holdBeforeFirstConnect ? newHold(null, true) : null
|
||||
// Why: only the latest connect may settle pre-bind typing; a superseded one must not drop it.
|
||||
let latestExpectation: RestoredInputExpectation | null = null
|
||||
|
||||
const discard = (): void => {
|
||||
const current = held
|
||||
@@ -60,31 +80,74 @@ export function createRemoteRuntimeRecoveryInputHold(): RemoteRuntimeRecoveryInp
|
||||
current?.buffer.clear()
|
||||
}
|
||||
|
||||
const bufferFor = (endpoint: RemoteRuntimeInputEndpoint): PtyPreconnectInputBuffer => {
|
||||
if (held && !isSameRemoteRuntimeInputEndpoint(held.endpoint, endpoint)) {
|
||||
const matches = (endpoint: HeldEndpoint): boolean =>
|
||||
endpoint === null
|
||||
? held?.beforeBind === true
|
||||
: held?.endpoint != null && isSameRemoteRuntimeInputEndpoint(held.endpoint, endpoint)
|
||||
|
||||
const bufferFor = (endpoint: HeldEndpoint): PtyPreconnectInputBuffer => {
|
||||
if (held && !matches(endpoint)) {
|
||||
// Why: input typed at one shell must never run in its replacement (#10065).
|
||||
discard()
|
||||
}
|
||||
if (!held?.buffer.isBuffering()) {
|
||||
// Why: a drained buffer refuses input; its release bookkeeping may still be settling.
|
||||
held = { endpoint, buffer: createPtyPreconnectInputBuffer([], { recoveryHold: true }) }
|
||||
held = newHold(endpoint, endpoint === null)
|
||||
}
|
||||
return held.buffer
|
||||
}
|
||||
|
||||
return {
|
||||
isHolding: () => held?.buffer.isBuffering() === true,
|
||||
isHoldingBeforeBind: () => held?.beforeBind === true && held.buffer.isBuffering(),
|
||||
enqueue: (endpoint, data, inputKind) =>
|
||||
bufferFor(endpoint).enqueue(data, 'ordinary', inputKind),
|
||||
enqueueAccepted: (endpoint, data, inputKind, options) =>
|
||||
bufferFor(endpoint).enqueueAccepted(data, inputKind, undefined, options),
|
||||
enqueueAccepted: (endpoint, data, inputKind, acceptedOptions) =>
|
||||
bufferFor(endpoint).enqueueAccepted(data, inputKind, undefined, acceptedOptions),
|
||||
expect(endpoint) {
|
||||
const expectation: RestoredInputExpectation = { endpoint }
|
||||
latestExpectation = endpoint ? expectation : null
|
||||
if (!endpoint) {
|
||||
if (held?.beforeBind) {
|
||||
discard()
|
||||
}
|
||||
return expectation
|
||||
}
|
||||
let expected = held
|
||||
if (!expected?.buffer.isBuffering() || (expected.endpoint && !matches(endpoint))) {
|
||||
discard()
|
||||
expected = newHold(endpoint, true)
|
||||
held = expected
|
||||
}
|
||||
expected.endpoint ??= endpoint
|
||||
expected.beforeBind = true
|
||||
return expectation
|
||||
},
|
||||
settleExpectation(expectation, bound) {
|
||||
if (expectation !== latestExpectation) {
|
||||
return
|
||||
}
|
||||
latestExpectation = null
|
||||
if (!held?.beforeBind) {
|
||||
return
|
||||
}
|
||||
if (bound) {
|
||||
held.beforeBind = false
|
||||
} else {
|
||||
discard()
|
||||
}
|
||||
},
|
||||
release(bound, writer) {
|
||||
const current = held
|
||||
if (!current) {
|
||||
return
|
||||
}
|
||||
if (!isSameRemoteRuntimeInputEndpoint(current.endpoint, bound)) {
|
||||
if (!current.endpoint || !isSameRemoteRuntimeInputEndpoint(current.endpoint, bound)) {
|
||||
discard()
|
||||
if (current.beforeBind) {
|
||||
// Why: the restored screen still shows the original terminal until its connect settles.
|
||||
held = newHold(current.endpoint, true)
|
||||
}
|
||||
return
|
||||
}
|
||||
void current.buffer
|
||||
|
||||
Reference in New Issue
Block a user