mirror of
https://github.com/stablyai/orca.git
synced 2026-10-26 08:02:44 +00:00
Brennan Benson
321e2cf0ba
A mobile terminal.subscribe to a tab the desktop never mounted this session — a workspace the desktop isn't currently showing, a cold-activation-deferred tab (#8597), or a cold-parked tab — has no attached PTY, so the runtime has no headless emulator and the initial snapshot is empty. The mobile terminal then renders blank (repro: on the phone, scroll to an old workspace whose terminals the desktop unmounted; they stay blank). When a mobile subscribe hits an empty snapshot (or resolves no PTY), the runtime now asks the renderer to background-mount that tab so the PTY attaches and the live data stream — already subscribed on that path — delivers its output. A never-mounted workspace has no renderer-graph leaf, so its terminals are surfaced to mobile via synthetic pty:<ptyId> handles that carry no real UI tabId. The runtime passes the ptyId through the mount request and the renderer resolves the owning tab from the persisted tab model (resolveTerminalTabIdForPtyId), falling back to a whole-worktree background mount if it cannot. The request is idempotent for already-mounted tabs and never unmounts a live pane, so the worst case is the same blank as before — no regression. Validated on an iOS simulator with a two-workspace desktop: opening the workspace the desktop was not showing rendered a fully blank terminal without the fix and the live shell prompt with it. Tests: runtime mount-request gating (real-tab / pty-form / no-ptyId / unknown), subscribe-handler empty-snapshot-vs-content, and the ptyId->tabId resolver; all 608 RPC method tests pass.
867 lines
32 KiB
TypeScript
867 lines
32 KiB
TypeScript
import { describe, expect, it, vi } from 'vitest'
|
|
import { RpcDispatcher } from './dispatcher'
|
|
import type { RpcRequest } from './core'
|
|
import type { OrcaRuntimeService } from '../orca-runtime'
|
|
import { TERMINAL_METHODS } from './methods/terminal'
|
|
import type { RuntimeTerminalWait } from '../../../shared/runtime-types'
|
|
import {
|
|
TerminalStreamOpcode,
|
|
decodeTerminalStreamFrame,
|
|
decodeTerminalStreamJson,
|
|
decodeTerminalStreamText
|
|
} from '../../../shared/terminal-stream-protocol'
|
|
|
|
function stubRuntime(overrides: Partial<OrcaRuntimeService> = {}): OrcaRuntimeService {
|
|
return {
|
|
getRuntimeId: () => 'test-runtime',
|
|
// Why: subscribe streams register as remote view subscribers for Phase-5
|
|
// query-authority suppression (terminal-query-authority.md).
|
|
registerRemoteTerminalViewSubscriber: () => () => {},
|
|
// Why: a no-PTY / empty-snapshot mobile subscribe asks the renderer to
|
|
// background-mount the tab (STA-1840). No-op by default.
|
|
requestRendererTerminalTabMount: () => {},
|
|
...overrides
|
|
} as OrcaRuntimeService
|
|
}
|
|
|
|
const makeRequest = (method: string, params?: unknown): RpcRequest => ({
|
|
id: 'req-1',
|
|
authToken: 'tok',
|
|
method,
|
|
params
|
|
})
|
|
|
|
describe('terminal subscribe buffering', () => {
|
|
it('settles mobile subscribe waits when the stream signal aborts before PTY spawn', async () => {
|
|
vi.useFakeTimers()
|
|
try {
|
|
const messages: string[] = []
|
|
const controller = new AbortController()
|
|
const runtime = stubRuntime({
|
|
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }),
|
|
waitForLeafPtyId: vi.fn(
|
|
(_handle: string, _timeoutMs?: number, signal?: AbortSignal) =>
|
|
new Promise<string>((_resolve, reject) => {
|
|
signal?.addEventListener('abort', () => reject(new Error('request_aborted')), {
|
|
once: true
|
|
})
|
|
})
|
|
),
|
|
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false })
|
|
})
|
|
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
|
|
|
const dispatchPromise = dispatcher.dispatchStreaming(
|
|
makeRequest('terminal.subscribe', {
|
|
terminal: 'terminal-1',
|
|
client: { id: 'phone-1', type: 'mobile' },
|
|
capabilities: { terminalBinaryStream: 1 }
|
|
}),
|
|
(msg) => messages.push(msg),
|
|
{
|
|
signal: controller.signal,
|
|
connectionId: 'conn-phone',
|
|
sendBinary: vi.fn(),
|
|
registerBinaryStreamHandler: vi.fn(() => vi.fn())
|
|
}
|
|
)
|
|
|
|
expect(runtime.waitForLeafPtyId).toHaveBeenCalled()
|
|
controller.abort()
|
|
const outcomePromise = Promise.race([
|
|
dispatchPromise.then(() => 'settled'),
|
|
new Promise<'pending'>((resolve) => setTimeout(() => resolve('pending'), 0))
|
|
])
|
|
await vi.advanceTimersByTimeAsync(0)
|
|
|
|
expect(await outcomePromise).toBe('settled')
|
|
expect(runtime.readTerminal).not.toHaveBeenCalled()
|
|
expect(messages).toEqual([])
|
|
} finally {
|
|
vi.useRealTimers()
|
|
}
|
|
})
|
|
|
|
it('captures live queries before awaiting mobile fit and delivers them after the snapshot', async () => {
|
|
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
|
const cleanups = new Map<string, () => void>()
|
|
let dataListener:
|
|
| ((data: string, meta?: { seq?: number; rawLength?: number }) => void)
|
|
| undefined
|
|
let resolveMobileSubscribe: () => void = () => {}
|
|
const registerRemoteTerminalViewSubscriber = vi.fn(() => vi.fn())
|
|
const runtime = stubRuntime({
|
|
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
|
handleMobileSubscribe: vi.fn(
|
|
() =>
|
|
new Promise<boolean>((resolve) => {
|
|
resolveMobileSubscribe = () => resolve(true)
|
|
})
|
|
),
|
|
handleMobileUnsubscribe: vi.fn(),
|
|
subscribeToTerminalData: vi.fn((_ptyId, listener) => {
|
|
dataListener = listener
|
|
return vi.fn()
|
|
}),
|
|
registerRemoteTerminalViewSubscriber,
|
|
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
|
|
serializeTerminalBuffer: vi
|
|
.fn()
|
|
.mockResolvedValue({ data: 'snapshot', cols: 80, rows: 24, seq: 4 }),
|
|
getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }),
|
|
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
|
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
|
isTerminalAlternateScreen: vi.fn().mockReturnValue(false),
|
|
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
|
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
|
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
|
cleanups.set(id, cleanup)
|
|
}),
|
|
cleanupSubscription: vi.fn((id: string) => {
|
|
cleanups.get(id)?.()
|
|
cleanups.delete(id)
|
|
}),
|
|
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {}))
|
|
})
|
|
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
|
|
|
const dispatchPromise = dispatcher.dispatchStreaming(
|
|
makeRequest('terminal.subscribe', {
|
|
terminal: 'terminal-1',
|
|
client: { id: 'phone-1', type: 'mobile' },
|
|
capabilities: { terminalBinaryStream: 1 }
|
|
}),
|
|
vi.fn(),
|
|
{
|
|
connectionId: 'conn-phone',
|
|
sendBinary: (bytes) => {
|
|
binaryFrames.push(bytes)
|
|
},
|
|
registerBinaryStreamHandler: vi.fn(() => vi.fn())
|
|
}
|
|
)
|
|
|
|
await vi.waitFor(() => expect(runtime.handleMobileSubscribe).toHaveBeenCalled())
|
|
expect(dataListener).toBeDefined()
|
|
expect(registerRemoteTerminalViewSubscriber).toHaveBeenCalledWith('pty-1')
|
|
dataListener?.('\x1b[6n', { seq: 4, rawLength: 4 })
|
|
resolveMobileSubscribe()
|
|
|
|
await vi.waitFor(() =>
|
|
expect(
|
|
binaryFrames.some((bytes) => {
|
|
const frame = decodeTerminalStreamFrame(bytes)
|
|
return (
|
|
frame?.opcode === TerminalStreamOpcode.Output &&
|
|
decodeTerminalStreamText(frame.payload) === '\x1b[6n'
|
|
)
|
|
})
|
|
).toBe(true)
|
|
)
|
|
const queryOutputFrames = binaryFrames.filter((bytes) => {
|
|
const frame = decodeTerminalStreamFrame(bytes)
|
|
return (
|
|
frame?.opcode === TerminalStreamOpcode.Output &&
|
|
decodeTerminalStreamText(frame.payload) === '\x1b[6n'
|
|
)
|
|
})
|
|
expect(queryOutputFrames).toHaveLength(1)
|
|
|
|
runtime.cleanupSubscription('terminal-1:phone-1')
|
|
await dispatchPromise
|
|
})
|
|
|
|
it('marks scrollback-only subscribed previews truncated when the uncursored read is limited', async () => {
|
|
const messages: string[] = []
|
|
const runtime = stubRuntime({
|
|
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: null }),
|
|
readTerminal: vi.fn().mockResolvedValue({
|
|
tail: ['line 120'],
|
|
truncated: false,
|
|
limited: true
|
|
})
|
|
})
|
|
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
|
|
|
await dispatcher.dispatchStreaming(
|
|
makeRequest('terminal.subscribe', { terminal: 'terminal-1' }),
|
|
(msg) => messages.push(msg)
|
|
)
|
|
|
|
expect(messages.map((msg) => JSON.parse(msg).result)).toEqual([
|
|
{
|
|
type: 'subscribed',
|
|
streamId: null,
|
|
lines: ['line 120'],
|
|
truncated: true
|
|
},
|
|
{ type: 'end' }
|
|
])
|
|
})
|
|
|
|
it('marks legacy scrollback previews truncated when the uncursored read is limited', async () => {
|
|
const messages: string[] = []
|
|
const cleanups = new Map<string, () => void>()
|
|
const runtime = stubRuntime({
|
|
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
|
readTerminal: vi.fn().mockResolvedValue({
|
|
tail: ['line 120'],
|
|
truncated: false,
|
|
limited: true
|
|
}),
|
|
serializeTerminalBuffer: vi.fn().mockResolvedValue(null),
|
|
getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }),
|
|
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
|
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
|
subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()),
|
|
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
|
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
|
cleanups.set(id, cleanup)
|
|
}),
|
|
cleanupSubscription: vi.fn((id: string) => {
|
|
cleanups.get(id)?.()
|
|
cleanups.delete(id)
|
|
}),
|
|
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {}))
|
|
})
|
|
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)
|
|
)
|
|
|
|
await vi.waitFor(() =>
|
|
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'scrollback')).toBe(true)
|
|
)
|
|
const scrollback = messages
|
|
.map((msg) => JSON.parse(msg).result)
|
|
.find((result) => result?.type === 'scrollback')
|
|
expect(scrollback).toMatchObject({
|
|
type: 'scrollback',
|
|
lines: ['line 120'],
|
|
truncated: true
|
|
})
|
|
|
|
runtime.cleanupSubscription('terminal-1:desktop-1')
|
|
await dispatchPromise
|
|
})
|
|
|
|
it('does not register legacy JSON listeners after the stream signal aborts during snapshot', async () => {
|
|
vi.useFakeTimers()
|
|
try {
|
|
const messages: string[] = []
|
|
const controller = new AbortController()
|
|
let resolveSnapshot: (value: { data: string; cols: number; rows: number }) => void = () => {}
|
|
const runtime = stubRuntime({
|
|
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
|
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
|
|
serializeTerminalBuffer: vi.fn(
|
|
() =>
|
|
new Promise<{ data: string; cols: number; rows: number }>((resolve) => {
|
|
resolveSnapshot = resolve
|
|
})
|
|
),
|
|
getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }),
|
|
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
|
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
|
subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()),
|
|
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
|
registerSubscriptionCleanup: vi.fn(),
|
|
cleanupSubscription: vi.fn(),
|
|
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {}))
|
|
})
|
|
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-legacy-json', signal: controller.signal }
|
|
)
|
|
|
|
await vi.waitFor(() => expect(runtime.serializeTerminalBuffer).toHaveBeenCalled())
|
|
controller.abort()
|
|
resolveSnapshot({ data: '', cols: 80, rows: 24 })
|
|
const outcomePromise = Promise.race([
|
|
dispatchPromise.then(() => 'settled'),
|
|
new Promise<'pending'>((resolve) => setTimeout(() => resolve('pending'), 0))
|
|
])
|
|
await vi.advanceTimersByTimeAsync(0)
|
|
|
|
expect(await outcomePromise).toBe('settled')
|
|
expect(runtime.subscribeToTerminalData).not.toHaveBeenCalled()
|
|
expect(runtime.subscribeToFitOverrideChanges).not.toHaveBeenCalled()
|
|
// Cleanup must exist before snapshot work so an abort cannot orphan a
|
|
// desktop width floor, but the abort must consume it before listeners start.
|
|
expect(runtime.registerSubscriptionCleanup).toHaveBeenCalledWith(
|
|
'terminal-1:desktop-1',
|
|
expect.any(Function),
|
|
'conn-legacy-json'
|
|
)
|
|
expect(runtime.cleanupSubscription).toHaveBeenCalledWith('terminal-1:desktop-1')
|
|
expect(runtime.waitForTerminal).not.toHaveBeenCalled()
|
|
expect(messages).toEqual([])
|
|
} finally {
|
|
vi.useRealTimers()
|
|
}
|
|
})
|
|
|
|
it('marks binary subscribed previews truncated when the uncursored read is limited', async () => {
|
|
const messages: string[] = []
|
|
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
|
const cleanups = new Map<string, () => void>()
|
|
const runtime = stubRuntime({
|
|
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
|
readTerminal: vi.fn().mockResolvedValue({
|
|
tail: ['line 120'],
|
|
truncated: false,
|
|
limited: true
|
|
}),
|
|
serializeTerminalBuffer: vi.fn().mockResolvedValue(null),
|
|
getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }),
|
|
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()),
|
|
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
|
cleanups.set(id, cleanup)
|
|
}),
|
|
cleanupSubscription: vi.fn((id: string) => {
|
|
cleanups.get(id)?.()
|
|
cleanups.delete(id)
|
|
}),
|
|
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
|
|
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
|
|
updateMobileViewport: vi.fn().mockResolvedValue(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' },
|
|
capabilities: { terminalBinaryStream: 1 }
|
|
}),
|
|
(msg) => messages.push(msg),
|
|
{
|
|
connectionId: 'conn-binary-limited',
|
|
sendBinary: (bytes) => {
|
|
binaryFrames.push(bytes)
|
|
}
|
|
}
|
|
)
|
|
|
|
await vi.waitFor(() =>
|
|
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true)
|
|
)
|
|
const subscribed = messages
|
|
.map((msg) => JSON.parse(msg).result)
|
|
.find((result) => result?.type === 'subscribed')
|
|
expect(subscribed).toMatchObject({
|
|
type: 'subscribed',
|
|
lines: ['line 120'],
|
|
truncated: true
|
|
})
|
|
const snapshotStart = binaryFrames
|
|
.map((frame) => decodeTerminalStreamFrame(frame))
|
|
.find((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotStart)
|
|
expect(snapshotStart && decodeTerminalStreamJson(snapshotStart.payload)).toMatchObject({
|
|
truncated: true
|
|
})
|
|
|
|
runtime.cleanupSubscription('terminal-1:desktop-1')
|
|
await dispatchPromise
|
|
})
|
|
|
|
it('does not mark binary snapshot frames truncated from a limited read when serialized data is available', async () => {
|
|
const messages: string[] = []
|
|
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
|
const cleanups = new Map<string, () => void>()
|
|
const runtime = stubRuntime({
|
|
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
|
readTerminal: vi.fn().mockResolvedValue({
|
|
tail: ['line 120'],
|
|
truncated: false,
|
|
limited: true
|
|
}),
|
|
serializeTerminalBuffer: vi.fn().mockResolvedValue({
|
|
data: 'serialized snapshot\r\n',
|
|
cols: 100,
|
|
rows: 30
|
|
}),
|
|
getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }),
|
|
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()),
|
|
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
|
cleanups.set(id, cleanup)
|
|
}),
|
|
cleanupSubscription: vi.fn((id: string) => {
|
|
cleanups.get(id)?.()
|
|
cleanups.delete(id)
|
|
}),
|
|
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
|
|
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
|
|
updateMobileViewport: vi.fn().mockResolvedValue(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' },
|
|
capabilities: { terminalBinaryStream: 1 }
|
|
}),
|
|
(msg) => messages.push(msg),
|
|
{
|
|
connectionId: 'conn-binary-serialized-limited',
|
|
sendBinary: (bytes) => {
|
|
binaryFrames.push(bytes)
|
|
}
|
|
}
|
|
)
|
|
|
|
await vi.waitFor(() =>
|
|
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true)
|
|
)
|
|
const snapshotStart = binaryFrames
|
|
.map((frame) => decodeTerminalStreamFrame(frame))
|
|
.find((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotStart)
|
|
expect(snapshotStart && decodeTerminalStreamJson(snapshotStart.payload)).toMatchObject({
|
|
cols: 100,
|
|
rows: 30,
|
|
truncated: false,
|
|
truncatedByByteBudget: false
|
|
})
|
|
|
|
runtime.cleanupSubscription('terminal-1:desktop-1')
|
|
await dispatchPromise
|
|
})
|
|
|
|
it('recovers binary output overflow queued while the initial snapshot is serializing', async () => {
|
|
vi.useFakeTimers()
|
|
try {
|
|
const messages: string[] = []
|
|
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
|
const cleanups = new Map<string, () => void>()
|
|
const dataListenerRef: {
|
|
current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void
|
|
} = {}
|
|
const snapshotResolvers: ((value: {
|
|
data: string
|
|
cols: number
|
|
rows: number
|
|
seq?: number
|
|
source?: 'headless' | 'renderer'
|
|
}) => void)[] = []
|
|
const runtime = stubRuntime({
|
|
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
|
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
|
|
serializeTerminalBuffer: vi.fn(
|
|
() =>
|
|
new Promise<{
|
|
data: string
|
|
cols: number
|
|
rows: number
|
|
seq?: number
|
|
source?: 'headless' | 'renderer'
|
|
}>((resolve) => {
|
|
snapshotResolvers.push(resolve)
|
|
})
|
|
),
|
|
getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
|
|
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
|
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
|
subscribeToTerminalData: vi.fn((_: string, listener: (data: string) => void) => {
|
|
dataListenerRef.current = listener
|
|
return vi.fn()
|
|
}),
|
|
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
|
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
|
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
|
cleanups.set(id, cleanup)
|
|
}),
|
|
cleanupSubscription: vi.fn((id: string) => {
|
|
const cleanup = cleanups.get(id)
|
|
cleanups.delete(id)
|
|
cleanup?.()
|
|
}),
|
|
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
|
|
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
|
|
updateMobileViewport: vi.fn().mockResolvedValue(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' },
|
|
capabilities: { terminalBinaryStream: 1 }
|
|
}),
|
|
(msg) => messages.push(msg),
|
|
{
|
|
connectionId: 'conn-buffered',
|
|
sendBinary: (bytes) => {
|
|
binaryFrames.push(bytes)
|
|
}
|
|
}
|
|
)
|
|
|
|
await vi.waitFor(() => expect(dataListenerRef.current).toBeDefined())
|
|
const shiftSpy = vi.spyOn(Array.prototype, 'shift')
|
|
let seq = 0
|
|
for (let index = 0; index < 400; index += 1) {
|
|
const data = `${String(index).padStart(3, '0')}${'x'.repeat(1021)}`
|
|
seq += data.length
|
|
dataListenerRef.current?.(data, { seq, rawLength: data.length })
|
|
}
|
|
const shiftCallCount = shiftSpy.mock.calls.length
|
|
shiftSpy.mockRestore()
|
|
await vi.waitFor(() => expect(runtime.serializeTerminalBuffer).toHaveBeenCalled())
|
|
snapshotResolvers[0]?.({ data: '', cols: 120, rows: 40, seq: 0, source: 'headless' })
|
|
await vi.waitFor(() => expect(runtime.serializeTerminalBuffer).toHaveBeenCalledTimes(2))
|
|
snapshotResolvers[1]?.({
|
|
data: 'recovered after overflow\r\n',
|
|
cols: 120,
|
|
rows: 40,
|
|
seq,
|
|
source: 'headless'
|
|
})
|
|
await vi.waitFor(() =>
|
|
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true)
|
|
)
|
|
await vi.runOnlyPendingTimersAsync()
|
|
|
|
const decodedFrames = binaryFrames
|
|
.map((frame) => decodeTerminalStreamFrame(frame))
|
|
.filter((frame): frame is NonNullable<typeof frame> => frame !== null)
|
|
const snapshotStarts = decodedFrames.filter(
|
|
(frame) => frame.opcode === TerminalStreamOpcode.SnapshotStart
|
|
)
|
|
const decodedStarts = snapshotStarts.map((frame) => decodeTerminalStreamJson(frame.payload))
|
|
// Why one snapshot: overflow during the initial serialize is recovered
|
|
// INLINE (drop pending, re-read, re-serialize) before anything is sent,
|
|
// so the client's first scrollback snapshot is already current. The
|
|
// 'resized'/pending-output-overflow follow-up path remains only for
|
|
// overflow that begins after the initial snapshot went out.
|
|
expect(decodedStarts).toEqual([expect.objectContaining({ kind: 'scrollback', seq })])
|
|
const snapshotText = decodedFrames
|
|
.filter((frame) => frame.opcode === TerminalStreamOpcode.SnapshotChunk)
|
|
.map((frame) => decodeTerminalStreamText(frame.payload))
|
|
.join('')
|
|
const output = decodedFrames
|
|
.filter((frame) => frame.opcode === TerminalStreamOpcode.Output)
|
|
.map((frame) => decodeTerminalStreamText(frame.payload))
|
|
.join('')
|
|
expect(output.length).toBeLessThanOrEqual(256 * 1024)
|
|
expect(output).not.toContain('000')
|
|
expect(output).not.toContain('399')
|
|
expect(snapshotText).toContain('recovered after overflow')
|
|
expect(shiftCallCount).toBe(0)
|
|
|
|
runtime.cleanupSubscription('terminal-1:desktop-1')
|
|
await dispatchPromise
|
|
} finally {
|
|
vi.useRealTimers()
|
|
}
|
|
})
|
|
|
|
it('drops stale mobile resize re-stream completions for legacy binary streams', async () => {
|
|
const messages: string[] = []
|
|
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
|
const cleanups = new Map<string, () => void>()
|
|
let resizeListener:
|
|
| ((event: {
|
|
cols: number
|
|
rows: number
|
|
displayMode: string
|
|
reason: string
|
|
seq: number
|
|
}) => void)
|
|
| undefined
|
|
const restreamResolves: ((value: {
|
|
data: string
|
|
cols: number
|
|
rows: number
|
|
oscLinks?: {
|
|
row: number
|
|
startCol: number
|
|
endCol: number
|
|
uri: string
|
|
}[]
|
|
}) => void)[] = []
|
|
const serializeTerminalBuffer = vi
|
|
.fn()
|
|
.mockResolvedValueOnce({ data: 'initial', cols: 80, rows: 24 })
|
|
.mockImplementation(
|
|
() =>
|
|
new Promise<{
|
|
data: string
|
|
cols: number
|
|
rows: number
|
|
oscLinks?: {
|
|
row: number
|
|
startCol: number
|
|
endCol: number
|
|
uri: string
|
|
}[]
|
|
}>((resolve) => {
|
|
restreamResolves.push(resolve)
|
|
})
|
|
)
|
|
const runtime = stubRuntime({
|
|
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
|
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
|
|
serializeTerminalBuffer,
|
|
getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }),
|
|
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
|
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
|
isTerminalAlternateScreen: vi.fn().mockReturnValue(false),
|
|
handleMobileSubscribe: vi.fn().mockResolvedValue(undefined),
|
|
handleMobileUnsubscribe: vi.fn(),
|
|
subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()),
|
|
subscribeToTerminalResize: vi.fn((_, listener) => {
|
|
resizeListener = listener as typeof resizeListener
|
|
return vi.fn()
|
|
}),
|
|
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
|
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
|
cleanups.set(id, cleanup)
|
|
}),
|
|
cleanupSubscription: vi.fn((id: string) => {
|
|
const cleanup = cleanups.get(id)
|
|
cleanups.delete(id)
|
|
cleanup?.()
|
|
}),
|
|
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
|
|
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
|
|
updateMobileViewport: vi.fn().mockResolvedValue({ updated: true, applied: true })
|
|
})
|
|
const dispatcher = new RpcDispatcher({
|
|
runtime,
|
|
methods: TERMINAL_METHODS
|
|
})
|
|
|
|
const dispatchPromise = dispatcher.dispatchStreaming(
|
|
makeRequest('terminal.subscribe', {
|
|
terminal: 'terminal-1',
|
|
client: { id: 'phone-1', type: 'mobile' },
|
|
capabilities: { terminalBinaryStream: 1 }
|
|
}),
|
|
(msg) => messages.push(msg),
|
|
{
|
|
connectionId: 'conn-stale-resize',
|
|
sendBinary: (bytes) => {
|
|
binaryFrames.push(bytes)
|
|
}
|
|
}
|
|
)
|
|
|
|
await vi.waitFor(() => expect(resizeListener).toBeDefined())
|
|
binaryFrames.splice(0)
|
|
|
|
resizeListener?.({
|
|
cols: 90,
|
|
rows: 24,
|
|
displayMode: 'auto',
|
|
reason: 'apply-layout',
|
|
seq: 2
|
|
})
|
|
resizeListener?.({
|
|
cols: 100,
|
|
rows: 24,
|
|
displayMode: 'auto',
|
|
reason: 'apply-layout',
|
|
seq: 3
|
|
})
|
|
await vi.waitFor(() => expect(restreamResolves).toHaveLength(2))
|
|
|
|
const newerOscLinks = [{ row: 0, startCol: 4, endCol: 9, uri: 'https://example.com' }]
|
|
restreamResolves[1]?.({
|
|
data: 'newer',
|
|
cols: 100,
|
|
rows: 24,
|
|
oscLinks: newerOscLinks
|
|
})
|
|
await vi.waitFor(() =>
|
|
expect(
|
|
binaryFrames.some((frame) => {
|
|
const decoded = decodeTerminalStreamFrame(frame)
|
|
return (
|
|
decoded?.opcode === TerminalStreamOpcode.SnapshotChunk &&
|
|
decodeTerminalStreamText(decoded.payload) === 'newer'
|
|
)
|
|
})
|
|
).toBe(true)
|
|
)
|
|
restreamResolves[0]?.({ data: 'older', cols: 90, rows: 24 })
|
|
await Promise.resolve()
|
|
await Promise.resolve()
|
|
|
|
const snapshotData = binaryFrames
|
|
.map((frame) => decodeTerminalStreamFrame(frame))
|
|
.filter((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotChunk)
|
|
.map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : ''))
|
|
expect(snapshotData).toEqual(['newer'])
|
|
const snapshotStart = binaryFrames
|
|
.map((frame) => decodeTerminalStreamFrame(frame))
|
|
.find((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotStart)
|
|
expect(snapshotStart && decodeTerminalStreamJson(snapshotStart.payload)).toMatchObject({
|
|
kind: 'resized',
|
|
oscLinks: newerOscLinks
|
|
})
|
|
|
|
runtime.cleanupSubscription('terminal-1:phone-1')
|
|
await dispatchPromise
|
|
})
|
|
|
|
it('applies inline overflow recovery when the snapshot has no output seq', async () => {
|
|
vi.useFakeTimers()
|
|
try {
|
|
const messages: string[] = []
|
|
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
|
const cleanups = new Map<string, () => void>()
|
|
const dataListenerRef: {
|
|
current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void
|
|
} = {}
|
|
const snapshotResolvers: ((value: {
|
|
data: string
|
|
cols: number
|
|
rows: number
|
|
seq?: number
|
|
source?: 'headless' | 'renderer'
|
|
}) => void)[] = []
|
|
const runtime = stubRuntime({
|
|
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
|
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
|
|
serializeTerminalBuffer: vi.fn(
|
|
() =>
|
|
new Promise<{
|
|
data: string
|
|
cols: number
|
|
rows: number
|
|
seq?: number
|
|
source?: 'headless' | 'renderer'
|
|
}>((resolve) => {
|
|
snapshotResolvers.push(resolve)
|
|
})
|
|
),
|
|
getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
|
|
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
|
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
|
subscribeToTerminalData: vi.fn((_: string, listener: (data: string) => void) => {
|
|
dataListenerRef.current = listener
|
|
return vi.fn()
|
|
}),
|
|
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
|
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
|
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
|
cleanups.set(id, cleanup)
|
|
}),
|
|
cleanupSubscription: vi.fn((id: string) => {
|
|
const cleanup = cleanups.get(id)
|
|
cleanups.delete(id)
|
|
cleanup?.()
|
|
}),
|
|
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
|
|
sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
|
|
updateMobileViewport: vi.fn().mockResolvedValue(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' },
|
|
capabilities: { terminalBinaryStream: 1 }
|
|
}),
|
|
(msg) => messages.push(msg),
|
|
{
|
|
connectionId: 'conn-buffered-no-seq',
|
|
sendBinary: (bytes) => {
|
|
binaryFrames.push(bytes)
|
|
}
|
|
}
|
|
)
|
|
|
|
await vi.waitFor(() => expect(dataListenerRef.current).toBeDefined())
|
|
let seq = 0
|
|
for (let index = 0; index < 400; index += 1) {
|
|
const data = `${String(index).padStart(3, '0')}${'x'.repeat(1021)}`
|
|
seq += data.length
|
|
dataListenerRef.current?.(data, { seq, rawLength: data.length })
|
|
}
|
|
await vi.waitFor(() => expect(runtime.serializeTerminalBuffer).toHaveBeenCalled())
|
|
snapshotResolvers[0]?.({ data: '', cols: 120, rows: 40, seq: 0, source: 'headless' })
|
|
await vi.waitFor(() => expect(runtime.serializeTerminalBuffer).toHaveBeenCalledTimes(2))
|
|
// Why applying is safe without a seq here: inline recovery serialized
|
|
// AFTER dropping the overflowed pending queue, so the snapshot covers
|
|
// those chunks by construction — no seq-based trimming is needed.
|
|
snapshotResolvers[1]?.({
|
|
data: 'renderer fallback snapshot\r\n',
|
|
cols: 120,
|
|
rows: 40,
|
|
source: 'renderer'
|
|
})
|
|
await vi.waitFor(() =>
|
|
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true)
|
|
)
|
|
await vi.runOnlyPendingTimersAsync()
|
|
|
|
const decodedFrames = binaryFrames
|
|
.map((frame) => decodeTerminalStreamFrame(frame))
|
|
.filter((frame): frame is NonNullable<typeof frame> => frame !== null)
|
|
const snapshotStarts = decodedFrames.filter(
|
|
(frame) => frame.opcode === TerminalStreamOpcode.SnapshotStart
|
|
)
|
|
// Why seq 1 (layout seq): a no-output-seq snapshot falls back to the
|
|
// layout seq on the wire; the recovered data still ships as the first
|
|
// and only scrollback snapshot.
|
|
expect(snapshotStarts.map((frame) => decodeTerminalStreamJson(frame.payload))).toEqual([
|
|
expect.objectContaining({ kind: 'scrollback', seq: 1 })
|
|
])
|
|
const snapshotText = decodedFrames
|
|
.filter((frame) => frame.opcode === TerminalStreamOpcode.SnapshotChunk)
|
|
.map((frame) => decodeTerminalStreamText(frame.payload))
|
|
.join('')
|
|
expect(snapshotText).toContain('renderer fallback snapshot')
|
|
const output = decodedFrames
|
|
.filter((frame) => frame.opcode === TerminalStreamOpcode.Output)
|
|
.map((frame) => decodeTerminalStreamText(frame.payload))
|
|
.join('')
|
|
// Why empty: the overflowed pending queue was dropped before the
|
|
// covering snapshot was serialized; nothing needs replay.
|
|
expect(output).toBe('')
|
|
|
|
runtime.cleanupSubscription('terminal-1:desktop-1')
|
|
await dispatchPromise
|
|
} finally {
|
|
vi.useRealTimers()
|
|
}
|
|
})
|
|
})
|