fix(terminal): paint paired remote terminals on first subscribe (#8954)
* fix(terminal): paint paired remote terminals on first subscribe * test(terminal): cover desktop legacy subscriber PTY wait The widened terminal.subscribe gate (isMobile -> params.client) had no regression test: reverting it to the mobile-only gate left the suite green. Add a desktop legacy-subscriber case asserting the mount request and late-PTY wait fire before the scrollback-only fallback. * fix(terminal): cancel abandoned multiplex PTY waits Desktop subscribers can close while the new late-PTY wait is pending. Register provisional slot cancellation and abort all pending waits on unsubscribe, connection abort, or multiplex teardown so a late PTY cannot resurrect a ghost output stream. --------- Co-authored-by: Brennan Benson <79079362+brennanb2025@users.noreply.github.com>
This commit is contained in:
Rod Boev
GitHub
parent
5907816457
commit
72a5c6f199
|
|
@ -1580,6 +1580,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
let closed = false
|
||||
let cursor = 0
|
||||
const streams = new Map<number, TerminalMultiplexStream>()
|
||||
const pendingPtyWaitControllers = new Map<number, Set<AbortController>>()
|
||||
let ackTotalInFlightBytes = 0
|
||||
let resolveMultiplex = (): void => {}
|
||||
const multiplexClosed = new Promise<void>((resolve) => {
|
||||
|
|
@ -1799,11 +1800,28 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
emit({ type: 'end', streamId })
|
||||
}
|
||||
}
|
||||
const cancelPendingPtyWaits = (streamId: number): void => {
|
||||
const controllers = pendingPtyWaitControllers.get(streamId)
|
||||
if (!controllers) {
|
||||
return
|
||||
}
|
||||
pendingPtyWaitControllers.delete(streamId)
|
||||
for (const controller of controllers) {
|
||||
controller.abort()
|
||||
}
|
||||
}
|
||||
const cancelAllPendingPtyWaits = (): void => {
|
||||
for (const streamId of Array.from(pendingPtyWaitControllers.keys())) {
|
||||
cancelPendingPtyWaits(streamId)
|
||||
}
|
||||
}
|
||||
const closeMultiplex = (): void => {
|
||||
if (closed) {
|
||||
return
|
||||
}
|
||||
closed = true
|
||||
signal?.removeEventListener('abort', cancelAllPendingPtyWaits)
|
||||
cancelAllPendingPtyWaits()
|
||||
const remoteDesktopKeysByPty = new Map<string, string[]>()
|
||||
for (const streamId of Array.from(streams.keys())) {
|
||||
const stream = streams.get(streamId)
|
||||
|
|
@ -1830,6 +1848,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.Unsubscribe) {
|
||||
cancelPendingPtyWaits(stream.streamId)
|
||||
detachStream(stream.streamId, false)
|
||||
return
|
||||
}
|
||||
|
|
@ -2081,15 +2100,47 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
emit({ type: 'end', streamId: request.streamId })
|
||||
return
|
||||
}
|
||||
if (!leaf?.ptyId && isMobile) {
|
||||
if (!leaf?.ptyId && request.client) {
|
||||
// Why: a never-mounted tab has no graph leaf to await; mounting the
|
||||
// exact tab lets its PTY attach without activating the worktree.
|
||||
runtime.requestRendererTerminalTabMount(request.terminal)
|
||||
const waitController = new AbortController()
|
||||
const pendingControllers = pendingPtyWaitControllers.get(request.streamId) ?? new Set()
|
||||
pendingControllers.add(waitController)
|
||||
pendingPtyWaitControllers.set(request.streamId, pendingControllers)
|
||||
if (signal?.aborted) {
|
||||
waitController.abort()
|
||||
}
|
||||
// Why: the live slot handler does not exist until the PTY attaches;
|
||||
// retain cancellation ownership while the pane is still pending.
|
||||
const unregisterPendingHandler = registerBinaryStreamHandler(
|
||||
request.streamId,
|
||||
(frame) => {
|
||||
if (frame.opcode === TerminalStreamOpcode.Unsubscribe) {
|
||||
cancelPendingPtyWaits(request.streamId)
|
||||
detachStream(request.streamId, false)
|
||||
}
|
||||
}
|
||||
)
|
||||
try {
|
||||
const ptyId = await runtime.waitForLeafPtyId(request.terminal, 10_000, signal)
|
||||
const ptyId = await runtime.waitForLeafPtyId(
|
||||
request.terminal,
|
||||
10_000,
|
||||
waitController.signal
|
||||
)
|
||||
leaf = { ptyId }
|
||||
} catch {
|
||||
if (closed || signal?.aborted) {
|
||||
if (closed || signal?.aborted || waitController.signal.aborted) {
|
||||
return
|
||||
}
|
||||
// Fall through to the explicit no_connected_pty error below.
|
||||
} finally {
|
||||
const currentControllers = pendingPtyWaitControllers.get(request.streamId)
|
||||
currentControllers?.delete(waitController)
|
||||
if (currentControllers?.size === 0) {
|
||||
pendingPtyWaitControllers.delete(request.streamId)
|
||||
}
|
||||
unregisterPendingHandler()
|
||||
}
|
||||
}
|
||||
if (!leaf?.ptyId) {
|
||||
|
|
@ -2213,30 +2264,6 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
return
|
||||
}
|
||||
|
||||
if (!isMobile) {
|
||||
stream.unsubscribeFit = runtime.subscribeToFitOverrideChanges(ptyId, (event) => {
|
||||
const mode =
|
||||
event.mode === 'mobile-fit'
|
||||
? event.mode
|
||||
: (runtime.getRemoteDesktopFitHold?.(ptyId, stream.remoteDesktopSubscriptionKey)
|
||||
.mode ?? 'desktop-fit')
|
||||
emit({
|
||||
type: 'fit-override-changed',
|
||||
streamId: request.streamId,
|
||||
mode,
|
||||
cols: event.cols,
|
||||
rows: event.rows
|
||||
})
|
||||
})
|
||||
stream.unsubscribeDriver = runtime.subscribeToDriverChanges(ptyId, (driver) => {
|
||||
emit({
|
||||
type: 'driver-changed',
|
||||
streamId: request.streamId,
|
||||
driver
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
let read = await runtime.readTerminal(request.terminal)
|
||||
let serialized = await serializeBudgetedMobileSnapshot(runtime, ptyId, isMobile)
|
||||
if (closed || streams.get(request.streamId) !== stream) {
|
||||
|
|
@ -2264,25 +2291,6 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
const layoutSeq = runtime.getLayout(ptyId)?.seq
|
||||
const snapshotFrameSeq = serialized?.seq ?? layoutSeq
|
||||
const snapshotOutputSeq = serialized?.seq
|
||||
if (!isMobile) {
|
||||
const fitOverride = runtime.getTerminalFitOverride(ptyId)
|
||||
const desktopHold = runtime.getRemoteDesktopFitHold?.(
|
||||
ptyId,
|
||||
stream.remoteDesktopSubscriptionKey
|
||||
) ?? { mode: 'desktop-fit' as const, cols: size?.cols ?? 0, rows: size?.rows ?? 0 }
|
||||
emit({
|
||||
type: 'fit-override-changed',
|
||||
streamId: request.streamId,
|
||||
mode: fitOverride?.mode ?? desktopHold.mode,
|
||||
cols: fitOverride?.cols ?? desktopHold.cols,
|
||||
rows: fitOverride?.rows ?? desktopHold.rows
|
||||
})
|
||||
emit({
|
||||
type: 'driver-changed',
|
||||
streamId: request.streamId,
|
||||
driver: runtime.getDriver(ptyId)
|
||||
})
|
||||
}
|
||||
emit({
|
||||
type: 'subscribed',
|
||||
streamId: request.streamId,
|
||||
|
|
@ -2327,6 +2335,46 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
stream.pendingOutputBytes = 0
|
||||
stream.pendingOutputOverflowed = false
|
||||
stream.outputBatcher.flush()
|
||||
if (!isMobile) {
|
||||
stream.unsubscribeFit = runtime.subscribeToFitOverrideChanges(ptyId, (event) => {
|
||||
const mode =
|
||||
event.mode === 'mobile-fit'
|
||||
? event.mode
|
||||
: (runtime.getRemoteDesktopFitHold?.(ptyId, stream.remoteDesktopSubscriptionKey)
|
||||
.mode ?? 'desktop-fit')
|
||||
emit({
|
||||
type: 'fit-override-changed',
|
||||
streamId: request.streamId,
|
||||
mode,
|
||||
cols: event.cols,
|
||||
rows: event.rows
|
||||
})
|
||||
})
|
||||
stream.unsubscribeDriver = runtime.subscribeToDriverChanges(ptyId, (driver) => {
|
||||
emit({
|
||||
type: 'driver-changed',
|
||||
streamId: request.streamId,
|
||||
driver
|
||||
})
|
||||
})
|
||||
const fitOverride = runtime.getTerminalFitOverride(ptyId)
|
||||
const desktopHold = runtime.getRemoteDesktopFitHold?.(
|
||||
ptyId,
|
||||
stream.remoteDesktopSubscriptionKey
|
||||
) ?? { mode: 'desktop-fit' as const, cols: size?.cols ?? 0, rows: size?.rows ?? 0 }
|
||||
emit({
|
||||
type: 'fit-override-changed',
|
||||
streamId: request.streamId,
|
||||
mode: fitOverride?.mode ?? desktopHold.mode,
|
||||
cols: fitOverride?.cols ?? desktopHold.cols,
|
||||
rows: fitOverride?.rows ?? desktopHold.rows
|
||||
})
|
||||
emit({
|
||||
type: 'driver-changed',
|
||||
streamId: request.streamId,
|
||||
driver: runtime.getDriver(ptyId)
|
||||
})
|
||||
}
|
||||
stream.unsubscribeResize = runtime.subscribeToTerminalResize(ptyId, (event) => {
|
||||
stream.outputBatcher.flush()
|
||||
const resizeGeneration = stream.resizeGeneration + 1
|
||||
|
|
@ -2430,6 +2478,8 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
}
|
||||
})
|
||||
|
||||
signal?.addEventListener('abort', cancelAllPendingPtyWaits, { once: true })
|
||||
|
||||
runtime.registerSubscriptionCleanup(
|
||||
`terminal-multiplex:${connectionId}`,
|
||||
closeMultiplex,
|
||||
|
|
@ -2464,10 +2514,10 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
}
|
||||
|
||||
// Why: the left pane's PTY spawns asynchronously after the tab is created.
|
||||
// Mobile clients that subscribe before the PTY is ready would get a bare
|
||||
// Clients that subscribe before the PTY is ready would get a bare
|
||||
// scrollback+end with no live stream or phone-fit. Wait for the PTY so
|
||||
// the subscribe can proceed normally.
|
||||
if (!leaf?.ptyId && isMobile) {
|
||||
if (!leaf?.ptyId && params.client) {
|
||||
// Why: a never-mounted tab has no graph leaf to await; mounting the
|
||||
// exact tab lets its PTY attach without activating the worktree.
|
||||
rendererMountRequestedBeforePty = runtime.requestRendererTerminalTabMount(params.terminal)
|
||||
|
|
|
|||
|
|
@ -26,6 +26,7 @@ function stubRuntime(overrides: Partial<OrcaRuntimeService> = {}): OrcaRuntimeSe
|
|||
// only stub the legacy resolveLeafForHandle still bind; tests that need a
|
||||
// null/stale leaf override this explicitly.
|
||||
resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
||||
requestRendererTerminalTabMount: vi.fn().mockReturnValue(true),
|
||||
updateRemoteDesktopViewer: vi.fn().mockResolvedValue(true),
|
||||
unregisterRemoteDesktopViewer: vi.fn().mockResolvedValue(true),
|
||||
unregisterRemoteDesktopViewers: vi.fn().mockResolvedValue(true),
|
||||
|
|
@ -40,6 +41,92 @@ function makeRequest(method: string, params?: unknown): RpcRequest {
|
|||
return { id: 'req-1', authToken: 'tok', method, params }
|
||||
}
|
||||
|
||||
function startDesktopMultiplexSubscribe(
|
||||
overrides: Partial<OrcaRuntimeService> = {},
|
||||
trace?: string[]
|
||||
) {
|
||||
const messages: string[] = []
|
||||
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
||||
const handlers = new Map<
|
||||
number,
|
||||
(frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void
|
||||
>()
|
||||
const cleanups = new Map<string, () => void>()
|
||||
const runtime = stubRuntime({
|
||||
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
|
||||
serializeTerminalBuffer: vi.fn().mockResolvedValue({ data: 'snapshot', cols: 120, rows: 40 }),
|
||||
getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
|
||||
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
||||
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
||||
subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()),
|
||||
getTerminalFitOverride: vi.fn().mockReturnValue(null),
|
||||
getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
|
||||
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
||||
cleanups.set(id, cleanup)
|
||||
}),
|
||||
cleanupSubscription: vi.fn((id: string) => {
|
||||
cleanups.get(id)?.()
|
||||
}),
|
||||
...overrides,
|
||||
waitForTerminal:
|
||||
overrides.waitForTerminal ?? vi.fn(() => new Promise<RuntimeTerminalWait>(() => {}))
|
||||
})
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
||||
const dispatchPromise = dispatcher.dispatchStreaming(
|
||||
makeRequest('terminal.multiplex', {}),
|
||||
(msg) => {
|
||||
messages.push(msg)
|
||||
const type = JSON.parse(msg).result?.type
|
||||
if (type) {
|
||||
trace?.push(type)
|
||||
}
|
||||
},
|
||||
{
|
||||
connectionId: 'conn-desktop-first-paint',
|
||||
sendBinary: (bytes) => {
|
||||
binaryFrames.push(bytes)
|
||||
const opcode = decodeTerminalStreamFrame(bytes)?.opcode
|
||||
if (
|
||||
opcode === TerminalStreamOpcode.SnapshotStart ||
|
||||
opcode === TerminalStreamOpcode.SnapshotChunk ||
|
||||
opcode === TerminalStreamOpcode.SnapshotEnd
|
||||
) {
|
||||
trace?.push('snapshot')
|
||||
}
|
||||
},
|
||||
registerBinaryStreamHandler: (streamId, handler) => {
|
||||
handlers.set(streamId, handler)
|
||||
return () => handlers.delete(streamId)
|
||||
}
|
||||
}
|
||||
)
|
||||
return { messages, binaryFrames, handlers, cleanups, runtime, dispatchPromise }
|
||||
}
|
||||
|
||||
function sendDesktopMultiplexSubscribe(
|
||||
handlers: Map<number, (frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void>
|
||||
) {
|
||||
handlers.get(0)?.(
|
||||
decodeTerminalStreamFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Subscribe,
|
||||
streamId: 0,
|
||||
seq: 1,
|
||||
payload: encodeTerminalStreamJson({
|
||||
streamId: 7,
|
||||
terminal: 'terminal-1',
|
||||
client: { id: 'desktop-1', type: 'desktop' },
|
||||
capabilities: { ackOutput: 1, desktopViewportClaims: 1 },
|
||||
viewport: { cols: 120, rows: 40 }
|
||||
})
|
||||
})
|
||||
)!
|
||||
)
|
||||
}
|
||||
|
||||
describe('terminal multiplex RPC', () => {
|
||||
it('multiplexes terminal streams and routes desktop resize to the source PTY', async () => {
|
||||
vi.useFakeTimers()
|
||||
|
|
@ -2676,11 +2763,15 @@ describe('terminal multiplex RPC', () => {
|
|||
)
|
||||
|
||||
await vi.waitFor(() => expect(runtime.waitForLeafPtyId).toHaveBeenCalled())
|
||||
expect(runtime.waitForLeafPtyId).toHaveBeenCalledWith('terminal-1', 10_000, controller.signal)
|
||||
const pendingWaitSignal = vi.mocked(runtime.waitForLeafPtyId).mock.calls[0]?.[2]
|
||||
expect(runtime.waitForLeafPtyId).toHaveBeenCalledWith(
|
||||
'terminal-1',
|
||||
10_000,
|
||||
expect.any(AbortSignal)
|
||||
)
|
||||
|
||||
controller.abort()
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
await vi.waitFor(() => expect(pendingWaitSignal?.aborted).toBe(true))
|
||||
|
||||
expect(runtime.readTerminal).not.toHaveBeenCalled()
|
||||
expect(
|
||||
|
|
@ -2694,6 +2785,285 @@ describe('terminal multiplex RPC', () => {
|
|||
await dispatchPromise
|
||||
})
|
||||
|
||||
it("waits for a desktop multiplex subscriber's PTY before retiring the terminal", async () => {
|
||||
let resolvePty: (ptyId: string) => void = () => {}
|
||||
const runtime = stubRuntime({
|
||||
resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }),
|
||||
requestRendererTerminalTabMount: vi.fn(),
|
||||
waitForLeafPtyId: vi.fn(
|
||||
() =>
|
||||
new Promise<string>((resolve) => {
|
||||
resolvePty = resolve
|
||||
})
|
||||
)
|
||||
})
|
||||
const harness = startDesktopMultiplexSubscribe(runtime)
|
||||
await vi.waitFor(() =>
|
||||
expect(harness.messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true)
|
||||
)
|
||||
sendDesktopMultiplexSubscribe(harness.handlers)
|
||||
await vi.waitFor(() => expect(runtime.waitForLeafPtyId).toHaveBeenCalled())
|
||||
expect(runtime.requestRendererTerminalTabMount).toHaveBeenCalledWith('terminal-1')
|
||||
expect(
|
||||
harness.binaryFrames
|
||||
.map((frame) => decodeTerminalStreamFrame(frame))
|
||||
.filter((frame) => frame?.opcode === TerminalStreamOpcode.Error)
|
||||
.map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : ''))
|
||||
).toEqual([])
|
||||
resolvePty('pty-1')
|
||||
await vi.waitFor(() =>
|
||||
expect(
|
||||
harness.binaryFrames.map((frame) => decodeTerminalStreamFrame(frame)?.opcode)
|
||||
).toContain(TerminalStreamOpcode.SnapshotChunk)
|
||||
)
|
||||
expect(
|
||||
harness.binaryFrames
|
||||
.map((frame) => decodeTerminalStreamFrame(frame))
|
||||
.filter((frame) => frame?.opcode === TerminalStreamOpcode.Error)
|
||||
.map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : ''))
|
||||
).toEqual([])
|
||||
harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.()
|
||||
await harness.dispatchPromise
|
||||
})
|
||||
|
||||
it('cancels a pending desktop PTY wait when its multiplex slot unsubscribes', async () => {
|
||||
let resolvePty: (ptyId: string) => void = () => {}
|
||||
let waitSignal: AbortSignal | undefined
|
||||
const readTerminal = vi.fn().mockResolvedValue({ tail: [], truncated: false })
|
||||
const subscribeToTerminalData = vi.fn().mockReturnValue(vi.fn())
|
||||
const registerRemoteTerminalViewSubscriber = vi.fn().mockReturnValue(vi.fn())
|
||||
const waitForLeafPtyId = vi.fn(
|
||||
(_handle: string, _timeoutMs?: number, signal?: AbortSignal) =>
|
||||
new Promise<string>((resolve, reject) => {
|
||||
resolvePty = resolve
|
||||
waitSignal = signal
|
||||
signal?.addEventListener('abort', () => reject(new Error('request_aborted')), {
|
||||
once: true
|
||||
})
|
||||
})
|
||||
)
|
||||
const harness = startDesktopMultiplexSubscribe({
|
||||
resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }),
|
||||
waitForLeafPtyId,
|
||||
readTerminal,
|
||||
subscribeToTerminalData,
|
||||
registerRemoteTerminalViewSubscriber
|
||||
})
|
||||
await vi.waitFor(() =>
|
||||
expect(harness.messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true)
|
||||
)
|
||||
sendDesktopMultiplexSubscribe(harness.handlers)
|
||||
await vi.waitFor(() => expect(waitForLeafPtyId).toHaveBeenCalled())
|
||||
|
||||
harness.handlers.get(7)?.(
|
||||
decodeTerminalStreamFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Unsubscribe,
|
||||
streamId: 7,
|
||||
seq: 2,
|
||||
payload: new Uint8Array()
|
||||
})
|
||||
)!
|
||||
)
|
||||
resolvePty('pty-1')
|
||||
|
||||
await vi.waitFor(() =>
|
||||
expect(waitSignal?.aborted || readTerminal.mock.calls.length > 0).toBe(true)
|
||||
)
|
||||
// Why: a closed pane must not become a hidden live-output consumer when its late PTY appears.
|
||||
expect(waitSignal?.aborted).toBe(true)
|
||||
expect(readTerminal).not.toHaveBeenCalled()
|
||||
expect(subscribeToTerminalData).not.toHaveBeenCalled()
|
||||
expect(registerRemoteTerminalViewSubscriber).not.toHaveBeenCalled()
|
||||
expect(harness.handlers.has(7)).toBe(false)
|
||||
|
||||
harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.()
|
||||
await harness.dispatchPromise
|
||||
})
|
||||
|
||||
it("still reports no_connected_pty when a desktop multiplex subscriber's PTY never appears", async () => {
|
||||
const runtime = stubRuntime({
|
||||
resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }),
|
||||
waitForLeafPtyId: vi.fn().mockRejectedValue(new Error('timeout'))
|
||||
})
|
||||
const harness = startDesktopMultiplexSubscribe(runtime)
|
||||
await vi.waitFor(() =>
|
||||
expect(harness.messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true)
|
||||
)
|
||||
sendDesktopMultiplexSubscribe(harness.handlers)
|
||||
await vi.waitFor(() =>
|
||||
expect(
|
||||
harness.binaryFrames.map((frame) => decodeTerminalStreamFrame(frame)?.opcode)
|
||||
).toContain(TerminalStreamOpcode.Error)
|
||||
)
|
||||
const errorFrame = harness.binaryFrames
|
||||
.map((frame) => decodeTerminalStreamFrame(frame))
|
||||
.find((frame) => frame?.opcode === TerminalStreamOpcode.Error)
|
||||
expect(errorFrame && decodeTerminalStreamText(errorFrame.payload)).toBe('no_connected_pty')
|
||||
harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.()
|
||||
await harness.dispatchPromise
|
||||
})
|
||||
|
||||
it('emits initial desktop fit events after the first multiplex snapshot', async () => {
|
||||
const trace: string[] = []
|
||||
let fitListener: ((event: { mode: string; cols: number; rows: number }) => void) | undefined
|
||||
let driverListener: ((driver: unknown) => void) | undefined
|
||||
const harness = startDesktopMultiplexSubscribe(
|
||||
{
|
||||
readTerminal: vi.fn(async () => {
|
||||
fitListener?.({ mode: 'desktop-fit', cols: 100, rows: 30 })
|
||||
driverListener?.({ kind: 'transition-during-snapshot' })
|
||||
return { tail: [], truncated: false } as unknown as Awaited<
|
||||
ReturnType<OrcaRuntimeService['readTerminal']>
|
||||
>
|
||||
}),
|
||||
subscribeToFitOverrideChanges: vi.fn((_ptyId, listener) => {
|
||||
fitListener = listener
|
||||
return vi.fn()
|
||||
}),
|
||||
subscribeToDriverChanges: vi.fn((_ptyId, listener) => {
|
||||
driverListener = listener
|
||||
return vi.fn()
|
||||
})
|
||||
},
|
||||
trace
|
||||
)
|
||||
await vi.waitFor(() =>
|
||||
expect(harness.messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true)
|
||||
)
|
||||
sendDesktopMultiplexSubscribe(harness.handlers)
|
||||
await vi.waitFor(() => expect(trace).toContain('driver-changed'))
|
||||
expect(trace.lastIndexOf('snapshot')).toBeLessThan(trace.indexOf('fit-override-changed'))
|
||||
expect(trace.lastIndexOf('snapshot')).toBeLessThan(trace.indexOf('driver-changed'))
|
||||
harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.()
|
||||
await harness.dispatchPromise
|
||||
})
|
||||
|
||||
it('does not wait when a desktop multiplex subscriber already has a PTY', async () => {
|
||||
const runtime = stubRuntime({
|
||||
requestRendererTerminalTabMount: vi.fn(),
|
||||
waitForLeafPtyId: vi.fn()
|
||||
})
|
||||
const harness = startDesktopMultiplexSubscribe(runtime)
|
||||
await vi.waitFor(() =>
|
||||
expect(harness.messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true)
|
||||
)
|
||||
sendDesktopMultiplexSubscribe(harness.handlers)
|
||||
await vi.waitFor(() =>
|
||||
expect(harness.messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(
|
||||
true
|
||||
)
|
||||
)
|
||||
expect(runtime.waitForLeafPtyId).not.toHaveBeenCalled()
|
||||
expect(runtime.requestRendererTerminalTabMount).not.toHaveBeenCalled()
|
||||
harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.()
|
||||
await harness.dispatchPromise
|
||||
})
|
||||
|
||||
it('preserves clientless multiplex subscriptions without a PTY wait', async () => {
|
||||
const messages: string[] = []
|
||||
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
||||
const handlers = new Map<
|
||||
number,
|
||||
(frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void
|
||||
>()
|
||||
let cleanup: () => void = () => {}
|
||||
const runtime = stubRuntime({
|
||||
resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }),
|
||||
requestRendererTerminalTabMount: vi.fn(),
|
||||
waitForLeafPtyId: vi.fn(),
|
||||
registerSubscriptionCleanup: vi.fn((_id: string, callback: () => void) => {
|
||||
cleanup = callback
|
||||
})
|
||||
})
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
||||
const dispatchPromise = dispatcher.dispatchStreaming(
|
||||
makeRequest('terminal.multiplex', {}),
|
||||
(msg) => messages.push(msg),
|
||||
{
|
||||
connectionId: 'conn-clientless-multiplex',
|
||||
sendBinary: (bytes) => {
|
||||
binaryFrames.push(bytes)
|
||||
},
|
||||
registerBinaryStreamHandler: (streamId, handler) => {
|
||||
handlers.set(streamId, handler)
|
||||
return () => handlers.delete(streamId)
|
||||
}
|
||||
}
|
||||
)
|
||||
await vi.waitFor(() =>
|
||||
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'ready')).toBe(true)
|
||||
)
|
||||
handlers.get(0)?.(
|
||||
decodeTerminalStreamFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Subscribe,
|
||||
streamId: 0,
|
||||
seq: 1,
|
||||
payload: encodeTerminalStreamJson({ streamId: 7, terminal: 'terminal-1' })
|
||||
})
|
||||
)!
|
||||
)
|
||||
await vi.waitFor(() =>
|
||||
expect(binaryFrames.map((frame) => decodeTerminalStreamFrame(frame)?.opcode)).toContain(
|
||||
TerminalStreamOpcode.Error
|
||||
)
|
||||
)
|
||||
expect(runtime.waitForLeafPtyId).not.toHaveBeenCalled()
|
||||
expect(runtime.requestRendererTerminalTabMount).not.toHaveBeenCalled()
|
||||
const errorFrame = binaryFrames
|
||||
.map((frame) => decodeTerminalStreamFrame(frame))
|
||||
.find((frame) => frame?.opcode === TerminalStreamOpcode.Error)
|
||||
expect(errorFrame && decodeTerminalStreamText(errorFrame.payload)).toBe('no_connected_pty')
|
||||
cleanup()
|
||||
await dispatchPromise
|
||||
})
|
||||
|
||||
it('preserves clientless legacy subscriptions without a PTY wait or mount', async () => {
|
||||
const messages: string[] = []
|
||||
const runtime = stubRuntime({
|
||||
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }),
|
||||
waitForLeafPtyId: vi.fn(),
|
||||
requestRendererTerminalTabMount: vi.fn(),
|
||||
readTerminal: vi.fn().mockResolvedValue({ tail: ['scrollback'], truncated: false })
|
||||
})
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
||||
const dispatchPromise = dispatcher.dispatchStreaming(
|
||||
makeRequest('terminal.subscribe', { terminal: 'terminal-1' }),
|
||||
(msg) => messages.push(msg),
|
||||
{ connectionId: 'conn-clientless-legacy' }
|
||||
)
|
||||
await dispatchPromise
|
||||
expect(runtime.waitForLeafPtyId).not.toHaveBeenCalled()
|
||||
expect(runtime.requestRendererTerminalTabMount).not.toHaveBeenCalled()
|
||||
expect(messages.map((msg) => JSON.parse(msg).result?.type)).toEqual(['subscribed', 'end'])
|
||||
})
|
||||
|
||||
it('waits for a desktop legacy subscriber PTY before the scrollback-only fallback', async () => {
|
||||
const messages: string[] = []
|
||||
const runtime = stubRuntime({
|
||||
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }),
|
||||
waitForLeafPtyId: vi.fn().mockRejectedValue(new Error('timeout')),
|
||||
requestRendererTerminalTabMount: vi.fn().mockReturnValue(true),
|
||||
readTerminal: vi.fn().mockResolvedValue({ tail: ['scrollback'], truncated: false })
|
||||
})
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
||||
const dispatchPromise = dispatcher.dispatchStreaming(
|
||||
makeRequest('terminal.subscribe', {
|
||||
terminal: 'terminal-1',
|
||||
client: { id: 'desktop-1', type: 'desktop' }
|
||||
}),
|
||||
(msg) => messages.push(msg),
|
||||
{ connectionId: 'conn-desktop-legacy' }
|
||||
)
|
||||
await dispatchPromise
|
||||
// Widened gate: a desktop client must mount + await its late PTY, not skip
|
||||
// straight to the bare scrollback path the way it did under the mobile-only gate.
|
||||
expect(runtime.requestRendererTerminalTabMount).toHaveBeenCalledWith('terminal-1')
|
||||
expect(runtime.waitForLeafPtyId).toHaveBeenCalledWith('terminal-1', 10_000, undefined)
|
||||
expect(messages.map((msg) => JSON.parse(msg).result?.type)).toEqual(['subscribed', 'end'])
|
||||
})
|
||||
|
||||
it('keeps view-subscriber releases balanced when a same-streamId subscribe overwrites a blocked one', async () => {
|
||||
const messages: string[] = []
|
||||
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
||||
|
|
|
|||
Loading…
Reference in New Issue