mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 08:03:28 +00:00
Deliver subscribe overflow recovery as resized snapshot
Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -1771,13 +1771,24 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
if (!recovery) {
|
||||
break
|
||||
}
|
||||
// Why: without an output seq (renderer-source fallback) covered
|
||||
// chunks cannot be trimmed exactly, and the renderer view may lag
|
||||
// the queued chunks under backpressure. Keep the bounded replay
|
||||
// instead of applying an unverifiable snapshot.
|
||||
if (typeof recovery.seq !== 'number') {
|
||||
break
|
||||
}
|
||||
// Why: shipped mobile clients drop a second scrollback snapshot for
|
||||
// an initialized handle but apply a resized snapshot inline by
|
||||
// re-initializing xterm with fresh scrollback. Omit seq on the wire
|
||||
// so the client's layout-seq staleness filter is not polluted with
|
||||
// output-byte sequences.
|
||||
const recoveryStats = sendSnapshotFrames(sendFrame, {
|
||||
kind: 'scrollback',
|
||||
kind: 'resized',
|
||||
cols: recovery.cols,
|
||||
rows: recovery.rows,
|
||||
displayMode,
|
||||
reason: 'pending-output-overflow',
|
||||
seq: recovery.seq,
|
||||
source: recovery.source,
|
||||
truncated: false,
|
||||
truncatedByByteBudget: recovery.truncatedByByteBudget,
|
||||
|
||||
@@ -429,14 +429,20 @@ describe('terminal subscribe buffering', () => {
|
||||
const snapshotStarts = decodedFrames.filter(
|
||||
(frame) => frame.opcode === TerminalStreamOpcode.SnapshotStart
|
||||
)
|
||||
expect(snapshotStarts.map((frame) => decodeTerminalStreamJson(frame.payload))).toEqual([
|
||||
const decodedStarts = snapshotStarts.map((frame) => decodeTerminalStreamJson(frame.payload))
|
||||
// Why: shipped mobile clients only apply a mid-session snapshot when it
|
||||
// arrives as a resized frame; a second scrollback frame is dropped.
|
||||
expect(decodedStarts).toEqual([
|
||||
expect.objectContaining({ kind: 'scrollback', seq: 1 }),
|
||||
expect.objectContaining({
|
||||
kind: 'resized',
|
||||
reason: 'pending-output-overflow',
|
||||
seq,
|
||||
source: 'headless'
|
||||
})
|
||||
])
|
||||
// Why: output-byte sequences must not pollute the client layout-seq
|
||||
// staleness filter.
|
||||
expect(decodedStarts[1]).not.toHaveProperty('seq')
|
||||
const snapshotText = decodedFrames
|
||||
.filter((frame) => frame.opcode === TerminalStreamOpcode.SnapshotChunk)
|
||||
.map((frame) => decodeTerminalStreamText(frame.payload))
|
||||
@@ -457,4 +463,118 @@ describe('terminal subscribe buffering', () => {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('keeps bounded replay when overflow recovery 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: renderer-source snapshots carry no output seq, so covered chunks
|
||||
// cannot be trimmed and the recovery snapshot must not be applied.
|
||||
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
|
||||
)
|
||||
expect(snapshotStarts.map((frame) => decodeTerminalStreamJson(frame.payload))).toEqual([
|
||||
expect.objectContaining({ kind: 'scrollback', seq: 1 })
|
||||
])
|
||||
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).toContain('399')
|
||||
|
||||
runtime.cleanupSubscription('terminal-1:desktop-1')
|
||||
await dispatchPromise
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user