mirror of
https://github.com/stablyai/orca.git
synced 2026-10-18 00:03:20 +00:00
Fix Remote Host downloads and agent status parity (#6436)
* Fix remote host downloads and agent status parity Co-authored-by: Orca <help@stably.ai> * Address remote download review comments Co-authored-by: Orca <help@stably.ai> * Avoid inefficient SSH chunk fallback Co-authored-by: Orca <help@stably.ai> --------- Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
Jinwoo Hong
Orca
@@ -33,6 +33,7 @@ import {
|
||||
import type {
|
||||
RuntimeFileListResult,
|
||||
RuntimeFileOpenResult,
|
||||
RuntimeFileReadChunkResult,
|
||||
RuntimeFilePreviewResult,
|
||||
RuntimeFileReadResult,
|
||||
RuntimeTerminalPathResolution
|
||||
@@ -469,6 +470,45 @@ export class RuntimeFileCommands {
|
||||
return { content: buffer.toString('utf-8'), isBinary: false }
|
||||
}
|
||||
|
||||
async readFileExplorerChunk(
|
||||
worktreeSelector: string,
|
||||
relativePath: string,
|
||||
offset: number,
|
||||
length: number
|
||||
): Promise<RuntimeFileReadChunkResult> {
|
||||
const target = await this.resolveFileExplorerPath(worktreeSelector, relativePath)
|
||||
const provider = target.connectionId ? getSshFilesystemProvider(target.connectionId) : null
|
||||
if (target.connectionId) {
|
||||
if (!provider) {
|
||||
throw new Error(SSH_FILESYSTEM_PROVIDER_UNAVAILABLE_MESSAGE)
|
||||
}
|
||||
const fileStat = await provider.stat(target.path)
|
||||
if (fileStat.type === 'directory') {
|
||||
throw new Error('Cannot download a directory')
|
||||
}
|
||||
throw new Error('SSH runtime chunked download is unavailable; use the SSH download path')
|
||||
}
|
||||
|
||||
const filePath = await resolveAuthorizedPath(target.path, this.host.requireStore())
|
||||
const fileStats = await stat(filePath)
|
||||
if (fileStats.isDirectory()) {
|
||||
throw new Error('Cannot download a directory')
|
||||
}
|
||||
const handle = await open(filePath, 'r')
|
||||
try {
|
||||
const buffer = Buffer.alloc(Math.min(length, Math.max(0, fileStats.size - offset)))
|
||||
const { bytesRead } = await handle.read(buffer, 0, buffer.byteLength, offset)
|
||||
const chunk = buffer.subarray(0, bytesRead)
|
||||
return {
|
||||
contentBase64: chunk.toString('base64'),
|
||||
bytesRead,
|
||||
eof: offset + bytesRead >= fileStats.size
|
||||
}
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
}
|
||||
|
||||
async writeFileExplorerFile(
|
||||
worktreeSelector: string,
|
||||
relativePath: string,
|
||||
|
||||
@@ -4655,6 +4655,8 @@ export class OrcaRuntimeService {
|
||||
this.fileCommands.watchFileExplorer.bind(this.fileCommands)
|
||||
readFileExplorerPreview: RuntimeFileCommands['readFileExplorerPreview'] =
|
||||
this.fileCommands.readFileExplorerPreview.bind(this.fileCommands)
|
||||
readFileExplorerChunk: RuntimeFileCommands['readFileExplorerChunk'] =
|
||||
this.fileCommands.readFileExplorerChunk.bind(this.fileCommands)
|
||||
writeFileExplorerFile: RuntimeFileCommands['writeFileExplorerFile'] =
|
||||
this.fileCommands.writeFileExplorerFile.bind(this.fileCommands)
|
||||
writeFileExplorerFileBase64: RuntimeFileCommands['writeFileExplorerFileBase64'] =
|
||||
|
||||
@@ -338,6 +338,33 @@ describe('file RPC methods', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('reads a file chunk for a selected worktree', async () => {
|
||||
const runtime = {
|
||||
getRuntimeId: () => 'test-runtime',
|
||||
readFileExplorerChunk: vi.fn().mockResolvedValue({
|
||||
contentBase64: 'YWJj',
|
||||
bytesRead: 3,
|
||||
eof: true
|
||||
})
|
||||
} as unknown as OrcaRuntimeService
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: FILE_METHODS })
|
||||
|
||||
const response = await dispatcher.dispatch(
|
||||
makeRequest('files.readChunk', {
|
||||
worktree: 'id:wt-1',
|
||||
relativePath: 'archive.zip',
|
||||
offset: 0,
|
||||
length: 1024
|
||||
})
|
||||
)
|
||||
|
||||
expect(runtime.readFileExplorerChunk).toHaveBeenCalledWith('id:wt-1', 'archive.zip', 0, 1024)
|
||||
expect(response).toMatchObject({
|
||||
ok: true,
|
||||
result: { contentBase64: 'YWJj', bytesRead: 3, eof: true }
|
||||
})
|
||||
})
|
||||
|
||||
it('reads a file explorer directory for a selected worktree', async () => {
|
||||
const runtime = {
|
||||
getRuntimeId: () => 'test-runtime',
|
||||
|
||||
@@ -78,6 +78,15 @@ const FileWriteBase64Chunk = FileWriteBase64.extend({
|
||||
append: z.boolean().optional()
|
||||
})
|
||||
|
||||
const FileReadChunk = FileOpen.extend({
|
||||
offset: z.number().int().nonnegative(),
|
||||
length: z
|
||||
.number()
|
||||
.int()
|
||||
.positive()
|
||||
.max(512 * 1024)
|
||||
})
|
||||
|
||||
const FileRename = WorktreeSelector.extend({
|
||||
oldRelativePath: z
|
||||
.unknown()
|
||||
@@ -175,6 +184,17 @@ export const FILE_METHODS: RpcAnyMethod[] = [
|
||||
handler: async (params, { runtime }) =>
|
||||
runtime.readFileExplorerPreview(params.worktree, params.relativePath)
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'files.readChunk',
|
||||
params: FileReadChunk,
|
||||
handler: async (params, { runtime }) =>
|
||||
runtime.readFileExplorerChunk(
|
||||
params.worktree,
|
||||
params.relativePath,
|
||||
params.offset,
|
||||
params.length
|
||||
)
|
||||
}),
|
||||
defineMethod({
|
||||
name: 'files.readDir',
|
||||
params: FileTreePath,
|
||||
|
||||
@@ -323,6 +323,27 @@ function appendPendingMultiplexOutput(
|
||||
stream.pendingOutputOverflowed ||= trimmed.overflowed
|
||||
}
|
||||
|
||||
function getOutputAfterSnapshotSeq(
|
||||
chunk: TerminalOutputChunk,
|
||||
snapshotSeq: number | undefined
|
||||
): string | null {
|
||||
if (
|
||||
typeof snapshotSeq !== 'number' ||
|
||||
typeof chunk.meta?.seq !== 'number' ||
|
||||
typeof chunk.meta.rawLength !== 'number'
|
||||
) {
|
||||
return chunk.data
|
||||
}
|
||||
if (chunk.meta.seq <= snapshotSeq) {
|
||||
return null
|
||||
}
|
||||
const chunkStartSeq = chunk.meta.seq - chunk.meta.rawLength
|
||||
if (chunkStartSeq >= snapshotSeq) {
|
||||
return chunk.data
|
||||
}
|
||||
return chunk.data.slice(snapshotSeq - chunkStartSeq)
|
||||
}
|
||||
|
||||
function trimPendingOutputToBudget(
|
||||
pendingOutput: TerminalOutputChunk[],
|
||||
pendingOutputBytes: number
|
||||
@@ -1489,7 +1510,8 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
const size = runtime.getTerminalSize(ptyId)
|
||||
const displayMode = runtime.getMobileDisplayMode(ptyId)
|
||||
const layoutSeq = runtime.getLayout(ptyId)?.seq
|
||||
const snapshotSeq = serialized?.seq ?? layoutSeq
|
||||
const snapshotFrameSeq = serialized?.seq ?? layoutSeq
|
||||
const snapshotOutputSeq = serialized?.seq
|
||||
if (!isMobile) {
|
||||
const fitOverride = runtime.getTerminalFitOverride(ptyId)
|
||||
emit({
|
||||
@@ -1520,7 +1542,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
cols: serialized?.cols ?? size?.cols ?? 80,
|
||||
rows: serialized?.rows ?? size?.rows ?? 24,
|
||||
displayMode,
|
||||
seq: snapshotSeq,
|
||||
seq: snapshotFrameSeq,
|
||||
truncated: serialized ? read.truncated : isTerminalReadPayloadIncomplete(read),
|
||||
truncatedByByteBudget: serialized?.truncatedByByteBudget,
|
||||
source: serialized?.source,
|
||||
@@ -1532,7 +1554,10 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
stream.lastResizeCols = serialized?.cols ?? size?.cols
|
||||
stream.buffering = false
|
||||
for (const chunk of stream.pendingOutput.splice(0)) {
|
||||
stream.outputBatcher.push(chunk.data, chunk.meta)
|
||||
const uncoveredData = getOutputAfterSnapshotSeq(chunk, snapshotOutputSeq)
|
||||
if (uncoveredData) {
|
||||
stream.outputBatcher.push(uncoveredData, chunk.meta)
|
||||
}
|
||||
}
|
||||
stream.pendingOutputBytes = 0
|
||||
stream.pendingOutputOverflowed = false
|
||||
@@ -1778,16 +1803,17 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
.catch(() => runtime.cleanupSubscription(subscriptionId))
|
||||
const sendFrame = (
|
||||
opcode: TerminalStreamOpcode,
|
||||
payload: Uint8Array<ArrayBufferLike> = new Uint8Array()
|
||||
payload: Uint8Array<ArrayBufferLike> = new Uint8Array(),
|
||||
frameSeq = cursor++
|
||||
): void => {
|
||||
if (closed || !sendBinary) {
|
||||
return
|
||||
}
|
||||
sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq: cursor++, payload }))
|
||||
sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq: frameSeq, payload }))
|
||||
}
|
||||
outputBatcher = createTerminalOutputBatcher((data) => {
|
||||
for (const chunk of iterateTerminalOutputFrameChunks(data)) {
|
||||
sendFrame(TerminalStreamOpcode.Output, chunk.bytes)
|
||||
outputBatcher = createTerminalOutputBatcher((data, meta) => {
|
||||
for (const chunk of iterateTerminalOutputFrameChunks(data, meta)) {
|
||||
sendFrame(TerminalStreamOpcode.Output, chunk.bytes, chunk.seq)
|
||||
}
|
||||
})
|
||||
unregisterBinaryHandler =
|
||||
@@ -1842,7 +1868,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
return
|
||||
}
|
||||
|
||||
unsubscribeData = runtime.subscribeToTerminalData(ptyId, (data) => {
|
||||
unsubscribeData = runtime.subscribeToTerminalData(ptyId, (data, meta) => {
|
||||
if (closed) {
|
||||
return
|
||||
}
|
||||
@@ -1854,13 +1880,13 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
const measurement = measureTerminalStreamByteLength(data, {
|
||||
stopAfterBytes: remainingBudget
|
||||
})
|
||||
pendingOutput.push({ data, bytes: measurement.byteLength })
|
||||
pendingOutput.push({ data, bytes: measurement.byteLength, meta })
|
||||
pendingOutputBytes += measurement.byteLength
|
||||
const trimmed = trimPendingOutputToBudget(pendingOutput, pendingOutputBytes)
|
||||
pendingOutputBytes = trimmed.bytes
|
||||
return
|
||||
}
|
||||
outputBatcher?.push(data)
|
||||
outputBatcher?.push(data, meta)
|
||||
})
|
||||
|
||||
const read = await runtime.readTerminal(params.terminal)
|
||||
@@ -1874,7 +1900,9 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
// the mobile client's stale-event filter knows the high-water mark.
|
||||
// Undefined when the PTY has never transitioned (filter is fail-open).
|
||||
// See docs/mobile-terminal-layout-state-machine.md.
|
||||
const seq = runtime.getLayout(ptyId)?.seq
|
||||
const layoutSeq = runtime.getLayout(ptyId)?.seq
|
||||
const snapshotFrameSeq = serialized?.seq ?? layoutSeq
|
||||
const snapshotOutputSeq = serialized?.seq
|
||||
emit({
|
||||
type: 'subscribed',
|
||||
streamId,
|
||||
@@ -1883,14 +1911,14 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
cols: serialized?.cols ?? size?.cols,
|
||||
rows: serialized?.rows ?? size?.rows,
|
||||
displayMode,
|
||||
seq
|
||||
seq: layoutSeq
|
||||
})
|
||||
const snapshotStats = sendSnapshotFrames(sendFrame, {
|
||||
kind: 'scrollback',
|
||||
cols: serialized?.cols ?? size?.cols ?? 80,
|
||||
rows: serialized?.rows ?? size?.rows ?? 24,
|
||||
displayMode,
|
||||
seq,
|
||||
seq: snapshotFrameSeq,
|
||||
truncated: serialized ? read.truncated : isTerminalReadPayloadIncomplete(read),
|
||||
truncatedByByteBudget: serialized?.truncatedByByteBudget,
|
||||
oscLinks: serialized?.oscLinks,
|
||||
@@ -1910,7 +1938,10 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
lastResizeCols = serialized?.cols ?? size?.cols
|
||||
buffering = false
|
||||
for (const item of pendingOutput.splice(0)) {
|
||||
outputBatcher.push(item.data, item.meta)
|
||||
const uncoveredData = getOutputAfterSnapshotSeq(item, snapshotOutputSeq)
|
||||
if (uncoveredData) {
|
||||
outputBatcher.push(uncoveredData, item.meta)
|
||||
}
|
||||
}
|
||||
pendingOutputBytes = 0
|
||||
outputBatcher.flush()
|
||||
|
||||
@@ -1026,6 +1026,275 @@ describe('terminal multiplex RPC', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('flushes output buffered during initial multiplex snapshot once', 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
|
||||
} = {}
|
||||
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: 120, rows: 40 }),
|
||||
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
||||
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
||||
subscribeToTerminalData: vi.fn((_: string, listener) => {
|
||||
dataListenerRef.current = listener
|
||||
return vi.fn()
|
||||
}),
|
||||
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
||||
getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
|
||||
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 }),
|
||||
updateDesktopViewport: vi.fn().mockResolvedValue(true)
|
||||
})
|
||||
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-output-on-subscribe',
|
||||
sendBinary: (bytes) => binaryFrames.push(bytes),
|
||||
registerBinaryStreamHandler: vi.fn(() => vi.fn())
|
||||
}
|
||||
)
|
||||
|
||||
await vi.waitFor(() => expect(dataListenerRef.current).toBeDefined())
|
||||
dataListenerRef.current?.('starting shell\r\n', { seq: 16, rawLength: 16 })
|
||||
resolveSnapshot({ data: '', cols: 120, rows: 40 })
|
||||
await vi.waitFor(() =>
|
||||
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true)
|
||||
)
|
||||
await vi.runOnlyPendingTimersAsync()
|
||||
|
||||
const output = binaryFrames
|
||||
.map((frame) => decodeTerminalStreamFrame(frame))
|
||||
.filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
|
||||
.map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : ''))
|
||||
.join('')
|
||||
expect(output).toBe('starting shell\r\n')
|
||||
|
||||
runtime.cleanupSubscription('terminal-1:desktop-1')
|
||||
await dispatchPromise
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('drops buffered multiplex output already covered by the initial snapshot 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
|
||||
} = {}
|
||||
let resolveSnapshot: (value: {
|
||||
data: string
|
||||
cols: number
|
||||
rows: number
|
||||
seq: 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; seq: number }>((resolve) => {
|
||||
resolveSnapshot = 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) => {
|
||||
dataListenerRef.current = listener
|
||||
return vi.fn()
|
||||
}),
|
||||
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
||||
getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
|
||||
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 }),
|
||||
updateDesktopViewport: vi.fn().mockResolvedValue(true)
|
||||
})
|
||||
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-output-covered-by-snapshot',
|
||||
sendBinary: (bytes) => binaryFrames.push(bytes),
|
||||
registerBinaryStreamHandler: vi.fn(() => vi.fn())
|
||||
}
|
||||
)
|
||||
|
||||
await vi.waitFor(() => expect(dataListenerRef.current).toBeDefined())
|
||||
const startupLine = 'starting shell\r\n'
|
||||
dataListenerRef.current?.(startupLine, {
|
||||
seq: startupLine.length,
|
||||
rawLength: startupLine.length
|
||||
})
|
||||
resolveSnapshot({
|
||||
data: startupLine,
|
||||
cols: 120,
|
||||
rows: 40,
|
||||
seq: startupLine.length
|
||||
})
|
||||
await vi.waitFor(() =>
|
||||
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true)
|
||||
)
|
||||
await vi.runOnlyPendingTimersAsync()
|
||||
|
||||
const outputFrames = binaryFrames
|
||||
.map((frame) => decodeTerminalStreamFrame(frame))
|
||||
.filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
|
||||
const snapshotStart = binaryFrames
|
||||
.map((frame) => decodeTerminalStreamFrame(frame))
|
||||
.find((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotStart)
|
||||
expect(snapshotStart && decodeTerminalStreamJson(snapshotStart.payload)).toMatchObject({
|
||||
seq: startupLine.length
|
||||
})
|
||||
expect(outputFrames).toHaveLength(0)
|
||||
|
||||
runtime.cleanupSubscription('terminal-1:desktop-1')
|
||||
await dispatchPromise
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('replays only buffered multiplex output not covered by the initial snapshot 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
|
||||
} = {}
|
||||
let resolveSnapshot: (value: {
|
||||
data: string
|
||||
cols: number
|
||||
rows: number
|
||||
seq: 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; seq: number }>((resolve) => {
|
||||
resolveSnapshot = 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) => {
|
||||
dataListenerRef.current = listener
|
||||
return vi.fn()
|
||||
}),
|
||||
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
||||
getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
|
||||
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 }),
|
||||
updateDesktopViewport: vi.fn().mockResolvedValue(true)
|
||||
})
|
||||
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-output-partially-covered-by-snapshot',
|
||||
sendBinary: (bytes) => binaryFrames.push(bytes),
|
||||
registerBinaryStreamHandler: vi.fn(() => vi.fn())
|
||||
}
|
||||
)
|
||||
|
||||
await vi.waitFor(() => expect(dataListenerRef.current).toBeDefined())
|
||||
const buffered = 'hello world'
|
||||
dataListenerRef.current?.(buffered, {
|
||||
seq: buffered.length,
|
||||
rawLength: buffered.length
|
||||
})
|
||||
resolveSnapshot({
|
||||
data: 'hello',
|
||||
cols: 120,
|
||||
rows: 40,
|
||||
seq: 'hello'.length
|
||||
})
|
||||
await vi.waitFor(() =>
|
||||
expect(messages.some((msg) => JSON.parse(msg).result?.type === 'subscribed')).toBe(true)
|
||||
)
|
||||
await vi.runOnlyPendingTimersAsync()
|
||||
|
||||
const output = binaryFrames
|
||||
.map((frame) => decodeTerminalStreamFrame(frame))
|
||||
.filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
|
||||
.map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : ''))
|
||||
.join('')
|
||||
expect(output).toBe(' world')
|
||||
|
||||
runtime.cleanupSubscription('terminal-1:desktop-1')
|
||||
await dispatchPromise
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('settles mobile multiplex PTY waits when the stream signal aborts before PTY spawn', async () => {
|
||||
const messages: string[] = []
|
||||
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
||||
|
||||
@@ -152,6 +152,7 @@ const MOBILE_RPC_METHOD_ALLOWLIST = new Set([
|
||||
'files.open',
|
||||
'files.openDiff',
|
||||
'files.read',
|
||||
'files.readChunk',
|
||||
'files.readPreview',
|
||||
'files.resolveTerminalPath',
|
||||
'git.abortMerge',
|
||||
|
||||
Reference in New Issue
Block a user