feat(runtime): stream file uploads instead of buffering whole files (#16106)

* feat(runtime): stream file uploads instead of buffering whole files

Staging read each dropped file whole with readFile(), base64-encoded it
(a 4/3 expansion), and passed the string through IPC to the renderer,
which re-chunked it. Peak memory was ~2.3x the file size before a byte
moved, so a 25 MB per-file cap existed to protect the heap.

Staging now records identity only. The byte pump moves into main, where
the file handle and the runtime socket both live: 384 KiB slices (512 KiB
once base64-encoded, matching the chunk size the renderer used) appended
through the existing files.writeBase64Chunk RPC. Peak memory is one slice
regardless of file size, so the ceilings become user-safety limits on an
unattended transfer — 2 GB per file, 8 GB per drop — and over-limit errors
name both the size and the limit.

Because staging and streaming are separate calls, the staged entry carries
size, inode, device and mtime, and the streamer re-checks all four against
the pre-open lstat and against the handle it actually reads. A source
replaced or rewritten at the same size between the two calls is refused
rather than uploaded under the original name. The post-read check compares
mtime as well as size, so an in-place rewrite mid-transfer aborts before
commitUpload renames anything into place.

O_NOFOLLOW, realpath containment and stat identity are preserved, and the
pairing revision plus the runtime id ride every chunk, so a re-pair or a
replacement runtime aborts instead of appending the rest of the file to a
different host.

No wire change: files.writeBase64Chunk and its params are untouched, so
old and new hosts behave identically. The SSH import path is separate and
unchanged. The web client has no local filesystem to stream from and says
so instead of failing obscurely.

* fix(runtime): close the empty-upload and per-drop budget holes

Two gaps the first pass left open.

A zero-byte source returned before the post-transfer identity check, so a
file that gained content during the empty write's round trip committed as
an empty file at the user's chosen name. The empty chunk now falls through
to the same final check the slice loop uses.

Each staged source also started its own byte counter, so the 8 GB ceiling
capped one source rather than the drop: five 2 GB files staged cleanly at
10 GB total. The IPC handler now carries one budget across sourcePaths and
adds only what each source actually staged. The per-file ceiling is still
re-enforced where the bytes move; the drop total holds at staging because
identity enforcement means each file streams exactly the bytes measured.

* docs(runtime): name the invariants the upload helpers carry

* fix(runtime): name the source in errors and stop uploads with their window

Three problems an independent review turned up.

A dropped file's relative path is '', so the over-limit error read "'' is
3 GB, over the 2 GB per-file remote import limit" — the message this change
exists to fix, naming nothing. Errors now fall back to the file's own name;
the staged entry keeps '' so the destination path is unaffected. The
streamer had the same shape, falling back to the hidden .orca-upload-<nonce>
temp destination, a path the user never chose.

The byte loop used to live in the renderer and died with it. Moving it into
main meant closing or reloading the window left the rest of a multi-GB
transfer running, with the renderer's temp cleanup never reaching its
finally. An AbortSignal now rides the caller's lifetime and every chunk, is
re-checked per slice, and main sweeps the abandoned temp path itself when
the renderer is no longer there to do it.

Upload failures also reached the import result wrapped in Electron's
"Error invoking remote method '...'" prefix, because the throw crossed IPC
instead of happening in-renderer; extractIpcErrorMessage unwraps it.

An existing staging test asserted the empty-name message, so it encoded the
bug rather than catching it; it now asserts the file name.

* test(runtime): cover the containment check and the per-chunk host guards

The "escapes the dropped root" test only reached the lstat symlink guard,
so assertEntryInsideRoot had no coverage at all. The shape that actually
needs it is a regular file under a symlinked intermediate directory: lstat
sees a plain file, and realpath containment is the only thing that refuses
it. Disabling the guard now fails this test and nothing else.

Nothing asserted that the SSH target, connection generation and execution
host reach the writeBase64Chunk params either — the renderer tests stop at
the IPC boundary, so the streamer's half of that contract was untested.

* fix(runtime): survive a straggling append when sweeping an aborted upload

Aborting rejects the in-flight chunk locally, but the host may still apply
that append, and appends open with flag 'a' — which recreates the file the
sweep just deleted. The delete and the straggler also race: they are
separate calls on a queue that is not ordered between them.

Slices are strictly sequential, so at most one append can be outstanding.
A second pass after it has had time to land is therefore sufficient, not
merely a heuristic. The sweep moves out of filesystem-mutations.ts into its
own module so the behaviour is testable directly.

Found by an independent review pass, which also pointed out that the
"escapes the dropped root" test only reached the lstat symlink guard.

* fix(runtime): abort uploads only when the document commits, and honour manual disconnect per chunk

did-start-navigation fires before will-navigate blocks an external link or a
stray file drop, and the renderer survives those (verified against Electron 43
with a hidden window). Aborting there killed a healthy upload with a misleading
'window went away' error. did-navigate fires only once a new document has
replaced the caller.

The renderer's per-chunk calls used to go through the IPC handler that refuses
a manually disconnected environment; the loop in main made no such check, so a
disconnect mid-upload kept pushing the rest of the file. The handler now
resolves the selector to an environment id and the streamer checks it per slice.

Adds slice-boundary coverage against the real chunk schema and host write
flags, staging-to-stream on a real filesystem, and handler-level lifetime tests.

---------

Co-authored-by: Neil <neil@stably.ai>
This commit is contained in:
mmarabel
2026-09-14 14:08:31 -07:00
committed by GitHub
co-authored by Neil
parent f21f81dcfc
commit 68f0b2e835
26 changed files with 2190 additions and 689 deletions
+9 -22
View File
@@ -1,3 +1,8 @@
import type {
StagedRuntimeUploadEntry,
StagedRuntimeUploadSource
} from '../../shared/runtime-upload-staging-contract'
export type ImportSkipReason = 'missing' | 'symlink' | 'permission-denied' | 'unsupported'
export type ResolveDroppedPathsResult = {
@@ -27,25 +32,7 @@ export type ImportItemResult =
reason: string
}
export type StagedExternalImportSource =
| {
sourcePath: string
status: 'staged'
name: string
kind: 'file' | 'directory'
entries: StagedExternalImportEntry[]
}
| {
sourcePath: string
status: 'skipped'
reason: ImportSkipReason
}
| {
sourcePath: string
status: 'failed'
reason: string
}
export type StagedExternalImportEntry =
| { relativePath: string; kind: 'directory' }
| { relativePath: string; kind: 'file'; contentBase64: string }
// Why: staging crosses IPC to the renderer and back into the streamer, so the
// shape lives in shared and every layer names the same type.
export type StagedExternalImportSource = StagedRuntimeUploadSource
export type StagedExternalImportEntry = StagedRuntimeUploadEntry
+42 -11
View File
@@ -73,6 +73,7 @@ describe('fs:importExternalPaths', () => {
size: 12,
ino: 1,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => true,
isDirectory: () => false,
isSymbolicLink: () => false
@@ -94,6 +95,7 @@ describe('fs:importExternalPaths', () => {
size: entry.isDir ? 0 : 12,
ino: entry.isDir ? 2 : 3,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => !entry.isDir,
isDirectory: () => entry.isDir,
isSymbolicLink: () => false
@@ -142,6 +144,7 @@ describe('fs:importExternalPaths', () => {
size: content.byteLength,
ino: 1,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => true
}),
createReadStream: () => Readable.from([content]),
@@ -216,6 +219,7 @@ describe('fs:importExternalPaths', () => {
size: 12,
ino: 1,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => true
}),
createReadStream: () => Readable.from([Buffer.from('file-content')]),
@@ -484,6 +488,7 @@ describe('fs:importExternalPaths', () => {
size: 4,
ino: 1,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => true,
isDirectory: () => false,
isSymbolicLink: () => false
@@ -498,6 +503,7 @@ describe('fs:importExternalPaths', () => {
size: 4,
ino: 1,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => true
}),
readFile: readFileHandleMock,
@@ -514,11 +520,21 @@ describe('fs:importExternalPaths', () => {
status: 'staged',
name: 'logo.png',
kind: 'file',
entries: [{ relativePath: '', kind: 'file', contentBase64: 'cG5n' }]
entries: [
{
relativePath: '',
kind: 'file',
byteLength: 4,
inode: 1,
deviceId: 1,
modifiedAtMs: 1700000000000
}
]
}
])
expect(copyFileMock).not.toHaveBeenCalled()
expect(readFileHandleMock).toHaveBeenCalled()
// Why: bodies stream at upload time, so staging must never read the file.
expect(readFileHandleMock).not.toHaveBeenCalled()
expect(closeMock).toHaveBeenCalled()
})
@@ -533,6 +549,7 @@ describe('fs:importExternalPaths', () => {
size: 0,
ino: 1,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => false,
isDirectory: () => true,
isSymbolicLink: () => false
@@ -543,6 +560,7 @@ describe('fs:importExternalPaths', () => {
size: 4,
ino: 2,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => true,
isDirectory: () => false,
isSymbolicLink: () => false
@@ -578,6 +596,7 @@ describe('fs:importExternalPaths', () => {
size: 4,
ino: 2,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => true
}),
readFile: vi.fn().mockResolvedValue(Buffer.from('icon')),
@@ -597,7 +616,14 @@ describe('fs:importExternalPaths', () => {
entries: [
{ relativePath: '', kind: 'directory' },
{ relativePath: '..assets', kind: 'directory' },
{ relativePath: '..assets/icon.txt', kind: 'file', contentBase64: 'aWNvbg==' }
{
relativePath: '..assets/icon.txt',
kind: 'file',
byteLength: 4,
inode: 2,
deviceId: 1,
modifiedAtMs: 1700000000000
}
]
}
])
@@ -612,6 +638,7 @@ describe('fs:importExternalPaths', () => {
size: 0,
ino: 1,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => false,
isDirectory: () => true,
isSymbolicLink: () => false
@@ -637,14 +664,15 @@ describe('fs:importExternalPaths', () => {
expect(openMock).not.toHaveBeenCalled()
})
it('checks runtime upload directory byte budget before reading a file that exceeds the total cap', async () => {
it('checks runtime upload directory byte budget before opening a file that exceeds the total cap', async () => {
const sourcePath = '/tmp/dropped/project'
const resolvedPath = path.resolve(sourcePath)
const filePaths = ['one.bin', 'two.bin', 'three.bin', 'four.bin', 'overflow.bin'].map((name) =>
path.join(resolvedPath, name)
)
const mib = 1024 * 1024
const regularSize = 25 * mib
// Four files exactly fill the 8 GB total ceiling; the fifth pushes past it.
const regularSize = 2 * 1024 * mib
const overflowSize = Number(mib)
const readFileMock = vi.fn().mockResolvedValue(Buffer.from('chunk'))
@@ -654,6 +682,7 @@ describe('fs:importExternalPaths', () => {
size: 0,
ino: 1,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => false,
isDirectory: () => true,
isSymbolicLink: () => false
@@ -666,6 +695,7 @@ describe('fs:importExternalPaths', () => {
size,
ino: fileIndex + 2,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => true,
isDirectory: () => false,
isSymbolicLink: () => false
@@ -689,6 +719,7 @@ describe('fs:importExternalPaths', () => {
size: regularSize,
ino: fileIndex + 2,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => true
}),
readFile: readFileMock,
@@ -702,11 +733,9 @@ describe('fs:importExternalPaths', () => {
sourcePaths: [sourcePath]
})) as { sources: { status: string; reason?: string }[] }
expect(result.sources[0]).toMatchObject({
status: 'failed',
reason: 'Remote import is too large'
})
expect(readFileMock).toHaveBeenCalledTimes(4)
expect(result.sources[0]).toMatchObject({ status: 'failed' })
expect(result.sources[0]?.reason).toContain('total remote import limit')
expect(readFileMock).not.toHaveBeenCalled()
expect(openMock).not.toHaveBeenCalledWith(filePaths.at(-1), expect.anything())
})
@@ -719,6 +748,7 @@ describe('fs:importExternalPaths', () => {
size: 4,
ino: 1,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => true,
isDirectory: () => false,
isSymbolicLink: () => false
@@ -732,6 +762,7 @@ describe('fs:importExternalPaths', () => {
size: 4,
ino: 2,
dev: 1,
mtimeMs: 1700000000000,
isFile: () => true
}),
readFile: readFileHandleMock,
@@ -744,7 +775,7 @@ describe('fs:importExternalPaths', () => {
expect(result.sources[0]).toMatchObject({
status: 'failed',
reason: "File changed during upload staging: ''"
reason: "File changed during upload staging: 'logo.png'"
})
expect(readFileHandleMock).not.toHaveBeenCalled()
})
@@ -0,0 +1,176 @@
import { EventEmitter } from 'node:events'
import { beforeEach, describe, expect, it, vi } from 'vitest'
const handlers = new Map<string, (event: unknown, args: unknown) => Promise<unknown>>()
const { handleMock, streamMock, sweepMock } = vi.hoisted(() => ({
handleMock: vi.fn(),
streamMock: vi.fn(),
sweepMock: vi.fn()
}))
vi.mock('electron', () => ({
ipcMain: { handle: handleMock },
app: { getPath: () => '/user-data' }
}))
vi.mock('./runtime-upload-file-stream', () => ({
streamExternalFileToRuntime: streamMock
}))
vi.mock('./runtime-upload-temp-sweep', () => ({
sweepAbandonedRuntimeUploadTempPath: sweepMock
}))
vi.mock('../../shared/runtime-environment-store', () => ({
resolveEnvironment: (_userDataPath: string, selector: string) => ({
id: selector === 'env-alias' ? 'env-1' : selector
})
}))
import { registerFilesystemMutationHandlers } from './filesystem-mutations'
import { RENDERER_GONE_MESSAGE } from './renderer-lifetime-abort'
const request = {
environmentId: 'env-1',
sourceRootPath: '/drop/file.bin',
entryRelativePath: '',
expected: { byteLength: 1, inode: 1, deviceId: 1, modifiedAtMs: 1 },
worktree: 'wt-1',
relativePath: '.file.bin.orca-upload-x',
expectedEnvironmentPairingRevision: 3,
expectedEnvironmentRuntimeId: 'rt-1'
}
function fakeSender(): EventEmitter {
return new EventEmitter()
}
function listenerCount(sender: EventEmitter): number {
return ['destroyed', 'render-process-gone', 'did-navigate'].reduce(
(total, name) => total + sender.listenerCount(name),
0
)
}
beforeEach(() => {
handlers.clear()
handleMock.mockReset()
streamMock.mockReset()
sweepMock.mockReset()
sweepMock.mockResolvedValue(undefined)
handleMock.mockImplementation((channel: string, handler: never) => {
handlers.set(channel, handler)
})
registerFilesystemMutationHandlers(
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the upload handler under test never reads the store; registration only needs a Store-shaped value.
{ getRepos: () => [], getSettings: () => ({ workspaceDir: '/workspace' }) } as never
)
})
function invoke(sender: EventEmitter): Promise<unknown> {
return handlers.get('fs:uploadExternalFileToRuntime')!({ sender }, request)
}
describe('fs:uploadExternalFileToRuntime', () => {
it('streams with the user data path and a live signal, and leaves no listeners behind', async () => {
const sender = fakeSender()
streamMock.mockImplementation(async (args: { userDataPath: string; signal: AbortSignal }) => {
expect(args.userDataPath).toBe('/user-data')
expect(args.signal.aborted).toBe(false)
expect(listenerCount(sender)).toBe(3)
return { byteLength: 42 }
})
await expect(invoke(sender)).resolves.toEqual({ byteLength: 42 })
expect(streamMock).toHaveBeenCalledWith(expect.objectContaining(request))
expect(sweepMock).not.toHaveBeenCalled()
expect(listenerCount(sender)).toBe(0)
})
it('resolves the selector to the environment id before streaming and sweeping', async () => {
const sender = fakeSender()
streamMock.mockImplementation(
({ signal }: { signal: AbortSignal }) =>
new Promise((_resolve, reject) => {
signal.addEventListener('abort', () => reject(signal.reason), { once: true })
sender.emit('destroyed')
})
)
await expect(
handlers.get('fs:uploadExternalFileToRuntime')!(
{ sender },
{ ...request, environmentId: 'env-alias' }
)
).rejects.toThrow(RENDERER_GONE_MESSAGE)
expect(streamMock).toHaveBeenCalledWith(expect.objectContaining({ environmentId: 'env-1' }))
expect(sweepMock).toHaveBeenCalledWith('/user-data', { ...request, environmentId: 'env-1' })
})
it('aborts, sweeps the temp path, and rethrows when the renderer is destroyed mid-stream', async () => {
const sender = fakeSender()
streamMock.mockImplementation(
({ signal }: { signal: AbortSignal }) =>
new Promise((_resolve, reject) => {
signal.addEventListener('abort', () => reject(signal.reason), { once: true })
sender.emit('destroyed')
})
)
await expect(invoke(sender)).rejects.toThrow(RENDERER_GONE_MESSAGE)
expect(sweepMock).toHaveBeenCalledTimes(1)
expect(sweepMock).toHaveBeenCalledWith('/user-data', request)
expect(listenerCount(sender)).toBe(0)
})
it('aborts once a reload commits, not on a blocked navigation or an in-app route change', async () => {
const sender = fakeSender()
let observed: AbortSignal | undefined
streamMock.mockImplementation(
({ signal }: { signal: AbortSignal }) =>
new Promise((resolve, reject) => {
observed = signal
signal.addEventListener('abort', () => reject(signal.reason), { once: true })
sender.emit('did-start-navigation', { isMainFrame: true, isSameDocument: true })
sender.emit('did-start-navigation', { isMainFrame: true, isSameDocument: false })
sender.emit('will-navigate', { defaultPrevented: true }, 'https://example.invalid/')
queueMicrotask(() => {
expect(signal.aborted).toBe(false)
sender.emit('did-navigate', 'file:///app/index.html', 200, 'OK')
resolve({ byteLength: 0 })
})
})
)
await expect(invoke(sender)).rejects.toThrow(RENDERER_GONE_MESSAGE)
expect(observed?.aborted).toBe(true)
expect(sweepMock).toHaveBeenCalledTimes(1)
})
it('does not sweep when the stream fails while the renderer is still alive', async () => {
const sender = fakeSender()
streamMock.mockRejectedValue(new Error("File changed since it was staged: 'file.bin'"))
await expect(invoke(sender)).rejects.toThrow("File changed since it was staged: 'file.bin'")
expect(sweepMock).not.toHaveBeenCalled()
expect(listenerCount(sender)).toBe(0)
})
it('still rethrows the stream error if the sweep itself throws', async () => {
const sender = fakeSender()
sweepMock.mockRejectedValue(new Error('sweep exploded'))
streamMock.mockImplementation(
({ signal }: { signal: AbortSignal }) =>
new Promise((_resolve, reject) => {
signal.addEventListener('abort', () => reject(signal.reason), { once: true })
sender.emit('render-process-gone')
})
)
// Why: the sweep contract is "never rejects"; if it ever did, this documents
// that the handler would surface the sweep error instead of the upload's.
await expect(invoke(sender)).rejects.toThrow('sweep exploded')
expect(listenerCount(sender)).toBe(0)
})
})
+52 -3
View File
@@ -1,4 +1,4 @@
import { ipcMain } from 'electron'
import { app, ipcMain } from 'electron'
import { constants } from 'node:fs'
import { copyFile, mkdir, writeFile } from 'node:fs/promises'
import { basename, dirname } from 'node:path'
@@ -18,7 +18,15 @@ import type {
StagedExternalImportSource
} from './filesystem-import-result-types'
import { importOneSource } from './filesystem-import-local'
import { stageOneSourceForRuntimeUpload } from './filesystem-runtime-upload-staging'
import {
stagedRuntimeUploadByteLength,
stageOneSourceForRuntimeUpload
} from './filesystem-runtime-upload-staging'
import { streamExternalFileToRuntime } from './runtime-upload-file-stream'
import { abortWhenRendererGone } from './renderer-lifetime-abort'
import { sweepAbandonedRuntimeUploadTempPath } from './runtime-upload-temp-sweep'
import type { RuntimeUploadFileStreamRequest } from '../../shared/runtime-upload-staging-contract'
import { resolveEnvironment } from '../../shared/runtime-environment-store'
/**
* IPC handlers for file/folder creation and renaming.
@@ -196,13 +204,54 @@ export function registerFilesystemMutationHandlers(store: Store): void {
args: { sourcePaths: string[] }
): Promise<{ sources: StagedExternalImportSource[] }> => {
const sources: StagedExternalImportSource[] = []
// Why: one budget for the whole drop — per-source counters would let five
// 2 GB files through a ceiling meant to cap the drop.
let totalBytes = 0
for (const sourcePath of args.sourcePaths) {
sources.push(await stageOneSourceForRuntimeUpload(sourcePath))
const source = await stageOneSourceForRuntimeUpload(sourcePath, totalBytes)
totalBytes += stagedRuntimeUploadByteLength(source)
sources.push(source)
}
return { sources }
}
)
// Why: the file handle and the runtime socket both live in main, so the byte
// pump runs here. The renderer keeps deconflict/commit/rollback orchestration
// and never sees file contents.
ipcMain.handle(
'fs:uploadExternalFileToRuntime',
async (event, args: RuntimeUploadFileStreamRequest): Promise<{ byteLength: number }> => {
const userDataPath = app.getPath('userData')
// Why: the streamer's manual-disconnect check keys on the environment id,
// and the renderer may pass any selector the store resolves.
const request = {
...args,
environmentId: resolveEnvironment(userDataPath, args.environmentId).id
}
// Why: the renderer's own loop died with its window. Now that the bytes
// move in main, a reload or close has to stop the transfer explicitly,
// or a multi-GB upload outlives the window that asked for it.
const lifetime = abortWhenRendererGone(event.sender)
try {
return await streamExternalFileToRuntime({
...request,
userDataPath,
signal: lifetime.signal
})
} catch (error) {
if (lifetime.signal.aborted) {
// Why: the renderer owns temp cleanup, and it is gone — so the
// abandoned temp path is only collectable from here.
await sweepAbandonedRuntimeUploadTempPath(userDataPath, request)
}
throw error
} finally {
lifetime.dispose()
}
}
)
// Why: terminal drag-and-drop resolver. Local worktrees pass paths through
// unchanged (reference-in-place; preserves zero-latency drop). SSH worktrees
// upload each path into `${worktreePath}/.orca/drops/` and return remote
@@ -0,0 +1,166 @@
import { lstat, mkdtemp, mkdir, rm, symlink, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import type * as RuntimeImportLimits from './runtime-import-limits'
type RuntimeImportLimitsModule = typeof RuntimeImportLimits
vi.mock('./filesystem-auth', () => ({ authorizeExternalPath: () => {} }))
// Why: real ceilings are gigabytes, and truncate() is not sparse on NTFS, so a
// literal over-limit fixture would allocate that much on Windows CI.
vi.mock('./runtime-import-limits', async (importOriginal) => ({
...(await importOriginal<RuntimeImportLimitsModule>()),
REMOTE_IMPORT_MAX_FILE_BYTES: 4 * 1024,
REMOTE_IMPORT_MAX_TOTAL_BYTES: 16 * 1024
}))
const { stagedRuntimeUploadByteLength, stageOneSourceForRuntimeUpload } =
await import('./filesystem-runtime-upload-staging')
let workDir: string
beforeEach(async () => {
workDir = await mkdtemp(join(tmpdir(), 'orca-upload-staging-'))
})
afterEach(async () => {
await rm(workDir, { force: true, recursive: true })
})
describe('stageOneSourceForRuntimeUpload', () => {
it('records size instead of file contents so staging never holds the body', async () => {
const filePath = join(workDir, 'note.txt')
await writeFile(filePath, 'hello world')
const staged = await stageOneSourceForRuntimeUpload(filePath)
expect(staged).toMatchObject({
status: 'staged',
kind: 'file',
name: 'note.txt',
entries: [{ relativePath: '', kind: 'file', byteLength: 11 }]
})
expect(JSON.stringify(staged)).not.toContain('contentBase64')
})
it('records the identity the uploader re-checks, not just the size', async () => {
const filePath = join(workDir, 'note.txt')
await writeFile(filePath, 'hello world')
const stat = await lstat(filePath)
const staged = await stageOneSourceForRuntimeUpload(filePath)
expect(staged).toMatchObject({
status: 'staged',
entries: [
{
byteLength: 11,
inode: stat.ino,
deviceId: stat.dev,
modifiedAtMs: stat.mtimeMs
}
]
})
})
it('stages a file with no cap error, where the old buffering path refused', async () => {
const filePath = join(workDir, 'big.bin')
await writeFile(filePath, Buffer.alloc(3 * 1024))
await expect(stageOneSourceForRuntimeUpload(filePath)).resolves.toMatchObject({
status: 'staged',
entries: [{ kind: 'file', byteLength: 3 * 1024 }]
})
})
it('names the file, the actual size and the limit when a file is over the ceiling', async () => {
const filePath = join(workDir, 'clip.mp4')
await writeFile(filePath, Buffer.alloc(6 * 1024))
const staged = await stageOneSourceForRuntimeUpload(filePath)
expect(staged).toMatchObject({ status: 'failed' })
// Why: a dropped file's relative path is '', so this is the regression that
// would otherwise report "'' is 6 KB, over the 4 KB ... limit".
expect(staged.status === 'failed' && staged.reason).toBe(
"'clip.mp4' is 6 KB, over the 4 KB per-file remote import limit"
)
})
it('names the offending entry by its path inside a dropped directory', async () => {
const rootPath = join(workDir, 'media')
await mkdir(join(rootPath, 'clips'), { recursive: true })
await writeFile(join(rootPath, 'clips', 'big.mp4'), Buffer.alloc(6 * 1024))
const staged = await stageOneSourceForRuntimeUpload(rootPath)
expect(staged.status === 'failed' && staged.reason).toContain("'clips/big.mp4'")
})
it('counts earlier sources in the drop against the total ceiling', async () => {
const filePath = join(workDir, 'second.bin')
await writeFile(filePath, Buffer.alloc(3 * 1024))
// Alone it fits; after 14 KB of earlier sources the 16 KB drop ceiling is gone.
await expect(stageOneSourceForRuntimeUpload(filePath, 0)).resolves.toMatchObject({
status: 'staged'
})
const overBudget = await stageOneSourceForRuntimeUpload(filePath, 14 * 1024)
expect(overBudget).toMatchObject({ status: 'failed' })
expect(overBudget.status === 'failed' && overBudget.reason).toContain(
'total remote import limit'
)
})
it('reports the bytes a source contributes to the drop budget', async () => {
const rootPath = join(workDir, 'tree')
await mkdir(join(rootPath, 'nested'), { recursive: true })
await writeFile(join(rootPath, 'a.txt'), 'aa')
await writeFile(join(rootPath, 'nested', 'b.txt'), 'bbb')
const staged = await stageOneSourceForRuntimeUpload(rootPath)
expect(stagedRuntimeUploadByteLength(staged)).toBe(5)
expect(
stagedRuntimeUploadByteLength({
sourcePath: '/missing',
status: 'skipped',
reason: 'missing'
})
).toBe(0)
})
// symlink() needs privileges or Developer Mode on Windows.
it.skipIf(process.platform === 'win32')('keeps rejecting symlinked sources', async () => {
const targetPath = join(workDir, 'target.txt')
await writeFile(targetPath, 'data')
const linkPath = join(workDir, 'link.txt')
await symlink(targetPath, linkPath)
await expect(stageOneSourceForRuntimeUpload(linkPath)).resolves.toMatchObject({
status: 'skipped',
reason: 'symlink'
})
})
it('stages directory trees as metadata for every entry', async () => {
const rootPath = join(workDir, 'assets')
await mkdir(join(rootPath, 'nested'), { recursive: true })
await writeFile(join(rootPath, 'a.txt'), 'aa')
await writeFile(join(rootPath, 'nested', 'b.txt'), 'bbb')
const staged = await stageOneSourceForRuntimeUpload(rootPath)
expect(staged.status).toBe('staged')
const entries = staged.status === 'staged' ? staged.entries : []
expect(entries).toEqual(
expect.arrayContaining([
{ relativePath: '', kind: 'directory' },
expect.objectContaining({ relativePath: 'a.txt', kind: 'file', byteLength: 2 }),
{ relativePath: 'nested', kind: 'directory' },
expect.objectContaining({ relativePath: 'nested/b.txt', kind: 'file', byteLength: 3 })
])
)
})
})
@@ -1,3 +1,8 @@
import {
formatByteCeiling,
REMOTE_IMPORT_MAX_FILE_BYTES,
REMOTE_IMPORT_MAX_TOTAL_BYTES
} from './runtime-import-limits'
import { constants } from 'node:fs'
import { lstat, open, readdir, realpath } from 'node:fs/promises'
import { basename, isAbsolute, join, relative, resolve, sep } from 'node:path'
@@ -8,18 +13,31 @@ import type {
StagedExternalImportSource
} from './filesystem-import-result-types'
const REMOTE_IMPORT_MAX_FILE_BYTES = 25 * 1024 * 1024
const REMOTE_IMPORT_MAX_TOTAL_BYTES = 100 * 1024 * 1024
class RuntimeUploadSymlinkError extends Error {}
/** Bytes this source contributes to the drop budget; 0 unless it staged. */
export function stagedRuntimeUploadByteLength(source: StagedExternalImportSource): number {
if (source.status !== 'staged') {
return 0
}
return source.entries.reduce(
(total, entry) => (entry.kind === 'file' ? total + entry.byteLength : total),
0
)
}
/**
* @param totalBytesBefore Bytes already staged by earlier sources in the same drop,
* so the total ceiling covers the whole drop rather than each source alone.
*/
export async function stageOneSourceForRuntimeUpload(
sourcePath: string
sourcePath: string,
totalBytesBefore = 0
): Promise<StagedExternalImportSource> {
const resolvedSource = resolve(sourcePath)
// Why: runtime uploads read client-local paths in the client main process;
// authorize before lstat/readFile just like local copy imports.
// authorize before lstat just like local copy imports.
authorizeExternalPath(resolvedSource)
let sourceStat: Awaited<ReturnType<typeof lstat>>
@@ -52,8 +70,8 @@ export async function stageOneSourceForRuntimeUpload(
}
try {
const entries = sourceStat.isDirectory()
? await stageDirectoryEntries(resolvedSource)
: [(await stageFileEntry(resolvedSource, '')).entry]
? await stageDirectoryEntries(resolvedSource, totalBytesBefore)
: [(await stageFileEntry(resolvedSource, '', { totalBytesBefore })).entry]
return {
sourcePath,
status: 'staged',
@@ -73,9 +91,12 @@ export async function stageOneSourceForRuntimeUpload(
}
}
async function stageDirectoryEntries(rootPath: string): Promise<StagedExternalImportEntry[]> {
async function stageDirectoryEntries(
rootPath: string,
totalBytesBefore: number
): Promise<StagedExternalImportEntry[]> {
const entries: StagedExternalImportEntry[] = [{ relativePath: '', kind: 'directory' }]
let totalBytes = 0
let totalBytes = totalBytesBefore
const rootRealPath = await realpath(rootPath)
async function visit(dirPath: string): Promise<void> {
@@ -126,52 +147,52 @@ async function stageDirectoryEntries(rootPath: string): Promise<StagedExternalIm
async function stageFileEntry(
filePath: string,
relativePath: string,
options?: { rootRealPath?: string; totalBytesBefore?: number }
options: { rootRealPath?: string; totalBytesBefore: number }
): Promise<{ entry: StagedExternalImportEntry; byteLength: number }> {
const statResult = await lstat(filePath)
const displayPath = normalizeRelativeUploadPath(relativePath)
// Why: a dropped file's relative path is '', so errors would name nothing.
// The entry keeps '' — only the message falls back to the file's own name.
const displayName = displayPath || basename(filePath)
if (statResult.isSymbolicLink()) {
throw new RuntimeUploadSymlinkError(`Symlink not allowed in '${displayPath}'`)
throw new RuntimeUploadSymlinkError(`Symlink not allowed in '${displayName}'`)
}
if (!statResult.isFile()) {
throw new Error(`Unsupported file type in '${displayPath}'`)
throw new Error(`Unsupported file type in '${displayName}'`)
}
if (options?.rootRealPath) {
await assertRealPathInsideRoot(options.rootRealPath, filePath, displayPath)
if (options.rootRealPath) {
await assertRealPathInsideRoot(options.rootRealPath, filePath, displayName)
}
const initialTotalBytes =
options?.totalBytesBefore === undefined
? statResult.size
: options.totalBytesBefore + statResult.size
assertRemoteUploadBudget(relativePath, statResult.size, initialTotalBytes)
assertRemoteUploadBudget(displayName, statResult.size, options.totalBytesBefore + statResult.size)
const fileHandle = await open(filePath, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0))
try {
const openedStat = await fileHandle.stat()
if (!openedStat.isFile()) {
throw new Error(`Unsupported file type in '${displayPath}'`)
throw new Error(`Unsupported file type in '${displayName}'`)
}
if (
openedStat.size !== statResult.size ||
(statResult.ino !== 0 && openedStat.ino !== 0 && openedStat.ino !== statResult.ino) ||
(statResult.dev !== 0 && openedStat.dev !== 0 && openedStat.dev !== statResult.dev)
) {
throw new Error(`File changed during upload staging: '${displayPath}'`)
}
const totalBytes =
options?.totalBytesBefore === undefined
? openedStat.size
: options.totalBytesBefore + openedStat.size
assertRemoteUploadBudget(relativePath, openedStat.size, totalBytes)
const buffer = await fileHandle.readFile()
const afterReadStat = await fileHandle.stat()
if (afterReadStat.size !== openedStat.size) {
throw new Error(`File changed during upload staging: '${displayPath}'`)
throw new Error(`File changed during upload staging: '${displayName}'`)
}
assertRemoteUploadBudget(
displayName,
openedStat.size,
options.totalBytesBefore + openedStat.size
)
// Why: bytes are read slice-by-slice at upload time, so staging records the
// identity the streamer re-checks rather than the body itself. Size alone
// would let a same-size replacement slip through between the two calls.
return {
entry: {
relativePath: displayPath,
kind: 'file',
contentBase64: buffer.toString('base64')
byteLength: openedStat.size,
inode: openedStat.ino,
deviceId: openedStat.dev,
modifiedAtMs: openedStat.mtimeMs
},
byteLength: openedStat.size
}
@@ -197,15 +218,21 @@ async function assertRealPathInsideRoot(
}
function assertRemoteUploadBudget(
relativePath: string,
displayName: string,
fileBytes: number,
totalBytes: number
): void {
if (fileBytes > REMOTE_IMPORT_MAX_FILE_BYTES) {
throw new Error(`'${relativePath}' is too large for remote import`)
throw new Error(
`'${displayName}' is ${formatByteCeiling(fileBytes)}, over the ` +
`${formatByteCeiling(REMOTE_IMPORT_MAX_FILE_BYTES)} per-file remote import limit`
)
}
if (totalBytes > REMOTE_IMPORT_MAX_TOTAL_BYTES) {
throw new Error('Remote import is too large')
throw new Error(
`This import is ${formatByteCeiling(totalBytes)}, over the ` +
`${formatByteCeiling(REMOTE_IMPORT_MAX_TOTAL_BYTES)} total remote import limit`
)
}
}
@@ -0,0 +1,91 @@
import { EventEmitter } from 'node:events'
import { describe, expect, it } from 'vitest'
import {
abortWhenRendererGone,
RENDERER_GONE_MESSAGE,
type RendererLifetimeSender
} from './renderer-lifetime-abort'
function fakeSender(): RendererLifetimeSender & EventEmitter {
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: EventEmitter implements the once/on/removeListener surface this helper uses, and those three are all it calls; WebContents' overloaded signatures cannot be satisfied structurally.
return new EventEmitter() as RendererLifetimeSender & EventEmitter
}
describe('abortWhenRendererGone', () => {
it('aborts when the renderer is destroyed', () => {
const sender = fakeSender()
const { signal } = abortWhenRendererGone(sender)
expect(signal.aborted).toBe(false)
sender.emit('destroyed')
expect(signal.aborted).toBe(true)
expect(String(signal.reason)).toContain(RENDERER_GONE_MESSAGE)
})
it('aborts when the render process is gone', () => {
const sender = fakeSender()
const { signal } = abortWhenRendererGone(sender)
sender.emit('render-process-gone')
expect(signal.aborted).toBe(true)
})
it('aborts once a reload has replaced the document, not on in-app route changes', () => {
const sender = fakeSender()
const { signal } = abortWhenRendererGone(sender)
sender.emit('did-start-navigation', {
isMainFrame: true,
isSameDocument: true,
url: 'file:///app#x'
})
sender.emit('did-navigate-in-page', 'file:///app#x')
expect(signal.aborted).toBe(false)
sender.emit('did-start-navigation', {
isMainFrame: true,
isSameDocument: false,
url: 'file:///app'
})
sender.emit('did-navigate', 'file:///app', 200, 'OK')
expect(signal.aborted).toBe(true)
})
it('ignores a main-frame navigation that starts but is blocked before it commits', () => {
// Why: Electron emits did-start-navigation before will-navigate gets to
// preventDefault() an external link or a stray file drop; the renderer
// document survives those, so the upload must too.
const sender = fakeSender()
const { signal } = abortWhenRendererGone(sender)
sender.emit('did-start-navigation', {
isMainFrame: true,
isSameDocument: false,
url: 'https://example.invalid/'
})
sender.emit('will-navigate', { defaultPrevented: true }, 'https://example.invalid/')
sender.emit('did-start-navigation', {
isMainFrame: true,
isSameDocument: false,
url: 'file:///Users/me/dropped.png'
})
sender.emit('will-navigate', { defaultPrevented: true }, 'file:///Users/me/dropped.png')
expect(signal.aborted).toBe(false)
})
it('leaves no listeners on a long-lived renderer once disposed', () => {
const sender = fakeSender()
const { dispose } = abortWhenRendererGone(sender)
expect(sender.listenerCount('destroyed')).toBe(1)
dispose()
dispose()
expect(sender.listenerCount('destroyed')).toBe(0)
expect(sender.listenerCount('render-process-gone')).toBe(0)
expect(sender.listenerCount('did-navigate')).toBe(0)
})
})
+45
View File
@@ -0,0 +1,45 @@
import type { WebContents } from 'electron'
export type RendererLifetimeSender = Pick<WebContents, 'once' | 'removeListener'>
export const RENDERER_GONE_MESSAGE = 'The window that started this upload went away'
/**
* Abort signal that fires when the calling renderer goes away.
*
* Work the renderer used to do itself died with it. Once it moves into main,
* nothing stops a long transfer from outliving the window that asked for it,
* so the caller's lifetime has to be wired up explicitly.
*
* Always `dispose()` in a finally — otherwise every call leaks a listener on a
* long-lived WebContents.
*/
export function abortWhenRendererGone(sender: RendererLifetimeSender): {
signal: AbortSignal
dispose: () => void
} {
const controller = new AbortController()
const abort = (): void => controller.abort(new Error(RENDERER_GONE_MESSAGE))
let disposed = false
sender.once('destroyed', abort)
sender.once('render-process-gone', abort)
// Why: did-start-navigation also fires for navigations that will-navigate then
// blocks — an external link, a stray file drop — and the renderer survives
// those. did-navigate fires only once a new document has replaced the caller,
// and never for same-document route changes inside the live app.
sender.once('did-navigate', abort)
return {
signal: controller.signal,
dispose: () => {
if (disposed) {
return
}
disposed = true
sender.removeListener('destroyed', abort)
sender.removeListener('render-process-gone', abort)
sender.removeListener('did-navigate', abort)
}
}
}
@@ -27,7 +27,8 @@ import {
import {
clearRuntimeEnvironmentManualDisconnect,
isRuntimeEnvironmentManuallyDisconnected,
markRuntimeEnvironmentManuallyDisconnected
markRuntimeEnvironmentManuallyDisconnected,
RUNTIME_MANUALLY_DISCONNECTED_MESSAGE
} from './runtime-environment-manual-disconnect'
import {
callRuntimeEnvironment,
@@ -42,7 +43,7 @@ function manuallyDisconnectedResponse(
ok: false,
error: {
code: 'runtime_manually_disconnected',
message: 'Runtime environment is manually disconnected.'
message: RUNTIME_MANUALLY_DISCONNECTED_MESSAGE
},
_meta: { runtimeId: environment.runtimeId }
}
@@ -1,5 +1,7 @@
const manuallyDisconnectedEnvironmentIds = new Set<string>()
export const RUNTIME_MANUALLY_DISCONNECTED_MESSAGE = 'Runtime environment is manually disconnected.'
export function markRuntimeEnvironmentManuallyDisconnected(environmentId: string): void {
manuallyDisconnectedEnvironmentIds.add(environmentId)
}
@@ -0,0 +1,33 @@
import { describe, expect, it } from 'vitest'
import {
formatByteCeiling,
REMOTE_IMPORT_MAX_FILE_BYTES,
REMOTE_IMPORT_MAX_TOTAL_BYTES
} from './runtime-import-limits'
describe('formatByteCeiling', () => {
it('renders a size one byte over a ceiling as larger than the ceiling', () => {
// "is 2 GB, over the 2 GB limit" reads like a broken check, not a big file.
expect(formatByteCeiling(REMOTE_IMPORT_MAX_FILE_BYTES)).toBe('2 GB')
expect(formatByteCeiling(REMOTE_IMPORT_MAX_FILE_BYTES + 1)).toBe('2.1 GB')
})
it('leaves an exact ceiling as a whole number', () => {
expect(formatByteCeiling(REMOTE_IMPORT_MAX_TOTAL_BYTES)).toBe('8 GB')
expect(formatByteCeiling(1024)).toBe('1 KB')
})
it('scales through the units', () => {
expect(formatByteCeiling(512)).toBe('512 B')
expect(formatByteCeiling(1024 * 1024)).toBe('1 MB')
expect(formatByteCeiling(1024 ** 4)).toBe('1 TB')
})
it('rounds up rather than to nearest', () => {
expect(formatByteCeiling(1024 * 1024 + 1)).toBe('1.1 MB')
})
it('does not crash on zero', () => {
expect(formatByteCeiling(0)).toBe('0 B')
})
})
+18
View File
@@ -0,0 +1,18 @@
// Why: staging streams slices at upload time and never holds a whole file, so
// these are user-safety ceilings on an unattended transfer, not memory guards.
// They stay until the drop UI can show progress and cancel a running upload.
export const REMOTE_IMPORT_MAX_FILE_BYTES = 2 * 1024 * 1024 * 1024
export const REMOTE_IMPORT_MAX_TOTAL_BYTES = 8 * 1024 * 1024 * 1024
/** Rounds up, so a size over a ceiling never renders as the ceiling itself. */
export function formatByteCeiling(bytes: number): string {
const units = ['B', 'KB', 'MB', 'GB', 'TB']
let value = bytes
let unit = 0
while (value >= 1024 && unit < units.length - 1) {
value /= 1024
unit += 1
}
const rounded = Math.ceil(value * 10) / 10
return `${Number.isInteger(rounded) ? rounded : rounded.toFixed(1)} ${units[unit]}`
}
@@ -0,0 +1,438 @@
import { lstat, mkdtemp, mkdir, rename, rm, symlink, utimes, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import type { StagedRuntimeUploadFileIdentity } from '../../shared/runtime-upload-staging-contract'
import type * as RuntimeImportLimits from './runtime-import-limits'
type RuntimeImportLimitsModule = typeof RuntimeImportLimits
type ChunkCall = {
relativePath: string
contentBase64: string
append: boolean
expectedSshTargetId?: string
expectedSshConnectionGeneration?: number
expectedExecutionHostId?: string
}
type RuntimeCallOptions = { expectedEnvironmentRuntimeId?: string; signal?: AbortSignal }
const callRuntimeEnvironment =
vi.fn<
(
userDataPath: string,
environmentId: string,
method: string,
params: ChunkCall,
timeoutMs?: number,
expectedEnvironmentPairingRevision?: number,
envelope?: unknown,
options?: RuntimeCallOptions
) => unknown
>()
vi.mock('./runtime-environment-transport-routing', () => ({
callRuntimeEnvironment: (...args: Parameters<typeof callRuntimeEnvironment>) =>
callRuntimeEnvironment(...args)
}))
vi.mock('./filesystem-auth', () => ({ authorizeExternalPath: () => {} }))
// Why: see filesystem-runtime-upload-staging.test.ts — a real over-limit fixture
// would allocate gigabytes on Windows.
vi.mock('./runtime-import-limits', async (importOriginal) => ({
...(await importOriginal<RuntimeImportLimitsModule>()),
REMOTE_IMPORT_MAX_FILE_BYTES: 2 * 1024 * 1024
}))
const { RUNTIME_UPLOAD_SLICE_BYTES, streamExternalFileToRuntime } =
await import('./runtime-upload-file-stream')
const {
clearRuntimeEnvironmentManualDisconnect,
markRuntimeEnvironmentManuallyDisconnected,
RUNTIME_MANUALLY_DISCONNECTED_MESSAGE
} = await import('./runtime-environment-manual-disconnect')
let workDir: string
function chunkCalls(): ChunkCall[] {
return callRuntimeEnvironment.mock.calls
.filter(([, , method]) => method === 'files.writeBase64Chunk')
.map(([, , , params]) => params)
}
function uploadedBytes(): Buffer {
return Buffer.concat(chunkCalls().map((call) => Buffer.from(call.contentBase64, 'base64')))
}
/** Mirrors what staging records, so tests exercise the real identity contract. */
async function stagedIdentity(filePath: string): Promise<StagedRuntimeUploadFileIdentity> {
const stat = await lstat(filePath)
return {
byteLength: stat.size,
inode: stat.ino,
deviceId: stat.dev,
modifiedAtMs: stat.mtimeMs
}
}
async function baseArgs(sourceRootPath: string, entryPath?: string) {
return {
userDataPath: '/user-data',
environmentId: 'env-1',
sourceRootPath,
entryRelativePath: entryPath ?? '',
expected: await stagedIdentity(entryPath ? join(sourceRootPath, entryPath) : sourceRootPath),
worktree: 'wt-1',
relativePath: '.upload.tmp'
}
}
/** A path whose identity was never measured; every field is deliberately absent. */
function unstagedArgs(sourceRootPath: string, entryPath?: string) {
return {
userDataPath: '/user-data',
environmentId: 'env-1',
sourceRootPath,
entryRelativePath: entryPath ?? '',
expected: { byteLength: 0, inode: 0, deviceId: 0, modifiedAtMs: 0 },
worktree: 'wt-1',
relativePath: '.upload.tmp'
}
}
beforeEach(async () => {
workDir = await mkdtemp(join(tmpdir(), 'orca-upload-stream-'))
callRuntimeEnvironment.mockReset()
callRuntimeEnvironment.mockResolvedValue({ id: 'x', ok: true, result: {}, _meta: {} })
})
afterEach(async () => {
await rm(workDir, { force: true, recursive: true })
})
describe('streamExternalFileToRuntime', () => {
it('sends a file larger than the old 25 MB cap as ordered append-only slices', async () => {
const size = RUNTIME_UPLOAD_SLICE_BYTES * 2 + 1234
const contents = Buffer.alloc(size)
for (let index = 0; index < size; index += 1) {
contents[index] = index % 251
}
const filePath = join(workDir, 'big.bin')
await writeFile(filePath, contents)
await expect(streamExternalFileToRuntime(await baseArgs(filePath))).resolves.toEqual({
byteLength: size
})
const calls = chunkCalls()
expect(calls).toHaveLength(3)
expect(calls.map((call) => call.append)).toEqual([false, true, true])
expect(uploadedBytes().equals(contents)).toBe(true)
})
it('refuses a source whose size no longer matches what staging measured', async () => {
const filePath = join(workDir, 'grown.bin')
await writeFile(filePath, Buffer.alloc(1024))
const staged = await stagedIdentity(filePath)
await writeFile(filePath, Buffer.alloc(2048))
await expect(
streamExternalFileToRuntime({ ...(await baseArgs(filePath)), expected: staged })
).rejects.toThrow("File changed since it was staged: 'grown.bin'")
expect(chunkCalls()).toHaveLength(0)
})
it('refuses a source swapped for a different file of the same size', async () => {
const filePath = join(workDir, 'swapped.bin')
await writeFile(filePath, Buffer.alloc(2048, 0x41))
const staged = await stagedIdentity(filePath)
// A rename-into-place keeps the size and changes the inode.
const decoyPath = join(workDir, 'decoy.bin')
await writeFile(decoyPath, Buffer.alloc(2048, 0x42))
await rename(decoyPath, filePath)
await expect(
streamExternalFileToRuntime({ ...(await baseArgs(filePath)), expected: staged })
).rejects.toThrow('File changed since it was staged')
expect(chunkCalls()).toHaveLength(0)
})
it('refuses a source rewritten in place at the same size after staging', async () => {
const filePath = join(workDir, 'rewritten.bin')
await writeFile(filePath, Buffer.alloc(2048, 0x41))
const staged = await stagedIdentity(filePath)
// Same inode and size; only the modification time moves.
await writeFile(filePath, Buffer.alloc(2048, 0x42))
const bumped = new Date(staged.modifiedAtMs + 5_000)
await utimes(filePath, bumped, bumped)
await expect(
streamExternalFileToRuntime({ ...(await baseArgs(filePath)), expected: staged })
).rejects.toThrow('File changed since it was staged')
expect(chunkCalls()).toHaveLength(0)
})
it('aborts when the source is rewritten at the same size mid-transfer', async () => {
const filePath = join(workDir, 'racing.bin')
const size = RUNTIME_UPLOAD_SLICE_BYTES * 2
await writeFile(filePath, Buffer.alloc(size, 0x41))
const args = await baseArgs(filePath)
let rewritten = false
callRuntimeEnvironment.mockImplementation(async () => {
if (!rewritten) {
rewritten = true
await writeFile(filePath, Buffer.alloc(size, 0x42))
const bumped = new Date(args.expected.modifiedAtMs + 5_000)
await utimes(filePath, bumped, bumped)
}
return { id: 'x', ok: true, result: {}, _meta: {} }
})
await expect(streamExternalFileToRuntime(args)).rejects.toThrow('File changed during upload')
})
it('accepts a source that still matches its staged identity', async () => {
const filePath = join(workDir, 'same.bin')
await writeFile(filePath, Buffer.alloc(2048))
await expect(streamExternalFileToRuntime(await baseArgs(filePath))).resolves.toEqual({
byteLength: 2048
})
})
it('refuses a file over the ceiling and names the source, not the temp path', async () => {
const filePath = join(workDir, 'clip.mp4')
await writeFile(filePath, Buffer.alloc(3 * 1024 * 1024))
// Why: relativePath here is '.upload.tmp', a path the user never chose.
await expect(streamExternalFileToRuntime(await baseArgs(filePath))).rejects.toThrow(
"'clip.mp4' is 3 MB, over the 2 MB per-file remote import limit"
)
expect(chunkCalls()).toHaveLength(0)
})
it('never buffers more than one slice per chunk', async () => {
const filePath = join(workDir, 'sliced.bin')
await writeFile(filePath, Buffer.alloc(RUNTIME_UPLOAD_SLICE_BYTES * 2))
await streamExternalFileToRuntime(await baseArgs(filePath))
for (const call of chunkCalls()) {
expect(Buffer.from(call.contentBase64, 'base64').byteLength).toBeLessThanOrEqual(
RUNTIME_UPLOAD_SLICE_BYTES
)
}
})
it('creates an empty destination for a zero-byte source', async () => {
const filePath = join(workDir, 'empty.txt')
await writeFile(filePath, '')
await expect(streamExternalFileToRuntime(await baseArgs(filePath))).resolves.toEqual({
byteLength: 0
})
expect(chunkCalls()).toEqual([expect.objectContaining({ append: false, contentBase64: '' })])
})
it('refuses to finish a zero-byte upload whose source gained content mid-write', async () => {
const filePath = join(workDir, 'grows.txt')
await writeFile(filePath, '')
const args = await baseArgs(filePath)
callRuntimeEnvironment.mockImplementation(async () => {
await writeFile(filePath, 'content arrived during the empty write')
return { id: 'x', ok: true, result: {}, _meta: {} }
})
await expect(streamExternalFileToRuntime(args)).rejects.toThrow('File changed during upload')
})
it('carries the pairing revision and runtime id on every chunk', async () => {
const filePath = join(workDir, 'guarded.bin')
await writeFile(filePath, Buffer.alloc(RUNTIME_UPLOAD_SLICE_BYTES + 10))
await streamExternalFileToRuntime({
...(await baseArgs(filePath)),
expectedEnvironmentPairingRevision: 41,
expectedEnvironmentRuntimeId: 'runtime-7'
})
const guards = callRuntimeEnvironment.mock.calls
.filter(([, , method]) => method === 'files.writeBase64Chunk')
.map(([, , , , , revision, , options]) => ({
revision,
runtimeId: options?.expectedEnvironmentRuntimeId
}))
expect(guards).toEqual([
{ revision: 41, runtimeId: 'runtime-7' },
{ revision: 41, runtimeId: 'runtime-7' }
])
})
it('stops mid-transfer when the caller aborts instead of streaming the rest', async () => {
const filePath = join(workDir, 'abandoned.bin')
await writeFile(filePath, Buffer.alloc(RUNTIME_UPLOAD_SLICE_BYTES * 4))
const controller = new AbortController()
callRuntimeEnvironment.mockImplementation(async () => {
controller.abort(new Error('window closed'))
return { id: 'x', ok: true, result: {}, _meta: {} }
})
await expect(
streamExternalFileToRuntime({ ...(await baseArgs(filePath)), signal: controller.signal })
).rejects.toThrow('window closed')
// One slice went out before the abort; the other three never do.
expect(chunkCalls()).toHaveLength(1)
})
it('refuses to start once the caller has already aborted', async () => {
const filePath = join(workDir, 'never.bin')
await writeFile(filePath, Buffer.alloc(1024))
const controller = new AbortController()
controller.abort(new Error('window closed'))
await expect(
streamExternalFileToRuntime({ ...(await baseArgs(filePath)), signal: controller.signal })
).rejects.toThrow('window closed')
expect(chunkCalls()).toHaveLength(0)
})
it('passes the abort signal to every chunk so an in-flight request is cancelled', async () => {
const filePath = join(workDir, 'signalled.bin')
await writeFile(filePath, Buffer.alloc(RUNTIME_UPLOAD_SLICE_BYTES + 10))
const controller = new AbortController()
await streamExternalFileToRuntime({
...(await baseArgs(filePath)),
signal: controller.signal
})
const signals = callRuntimeEnvironment.mock.calls
.filter(([, , method]) => method === 'files.writeBase64Chunk')
.map(([, , , , , , , options]) => options?.signal)
expect(signals).toEqual([controller.signal, controller.signal])
})
it('stops at the failing chunk instead of sending the rest of the file', async () => {
const filePath = join(workDir, 'fails.bin')
await writeFile(filePath, Buffer.alloc(RUNTIME_UPLOAD_SLICE_BYTES * 3))
callRuntimeEnvironment.mockResolvedValueOnce({ id: 'x', ok: true, result: {}, _meta: {} })
callRuntimeEnvironment.mockResolvedValueOnce({
id: 'x',
ok: false,
error: { code: 'write_failed', message: 'disk full' }
})
await expect(streamExternalFileToRuntime(await baseArgs(filePath))).rejects.toThrow('disk full')
expect(chunkCalls()).toHaveLength(2)
})
// symlink() needs privileges or Developer Mode on Windows.
it.skipIf(process.platform === 'win32')('refuses a symlinked source', async () => {
const targetPath = join(workDir, 'secret.txt')
await writeFile(targetPath, 'secret')
const linkPath = join(workDir, 'link.txt')
await symlink(targetPath, linkPath)
await expect(streamExternalFileToRuntime(unstagedArgs(linkPath))).rejects.toThrow(
'Symlink not allowed'
)
expect(chunkCalls()).toHaveLength(0)
})
it.skipIf(process.platform === 'win32')(
'refuses a regular file reached through a symlinked directory inside the root',
async () => {
// Why: the symlink guard only lstats the entry itself, which sees a plain
// file here — realpath containment is the only thing that catches this.
const outsideDir = join(workDir, 'outside')
await mkdir(outsideDir)
await writeFile(join(outsideDir, 'secret.txt'), 'secret')
const rootPath = join(workDir, 'root')
await mkdir(rootPath)
await symlink(outsideDir, join(rootPath, 'sub'))
await expect(
streamExternalFileToRuntime(unstagedArgs(rootPath, 'sub/secret.txt'))
).rejects.toThrow('Path escaped upload root during upload')
expect(chunkCalls()).toHaveLength(0)
}
)
it('forwards the host ownership expectations into every chunk', async () => {
const filePath = join(workDir, 'owned.bin')
await writeFile(filePath, Buffer.alloc(RUNTIME_UPLOAD_SLICE_BYTES + 10))
await streamExternalFileToRuntime({
...(await baseArgs(filePath)),
expectedSshTargetId: 'ssh-1',
expectedSshConnectionGeneration: 5,
expectedExecutionHostId: 'ssh:ssh-1'
})
const calls = callRuntimeEnvironment.mock.calls
.filter(([, , method]) => method === 'files.writeBase64Chunk')
.map(([, , , params]) => params)
expect(calls).toHaveLength(2)
for (const params of calls) {
expect(params).toMatchObject({
expectedSshTargetId: 'ssh-1',
expectedSshConnectionGeneration: 5,
expectedExecutionHostId: 'ssh:ssh-1'
})
}
})
it.skipIf(process.platform === 'win32')(
'refuses a symlinked directory entry before it reaches the containment check',
async () => {
const outsidePath = join(workDir, 'outside.txt')
await writeFile(outsidePath, 'outside')
const rootPath = join(workDir, 'root')
await mkdir(rootPath)
await symlink(outsidePath, join(rootPath, 'escape.txt'))
await expect(
streamExternalFileToRuntime(unstagedArgs(rootPath, 'escape.txt'))
).rejects.toThrow('Symlink not allowed')
expect(chunkCalls()).toHaveLength(0)
}
)
})
describe('manual disconnect during a transfer', () => {
afterEach(() => {
clearRuntimeEnvironmentManualDisconnect('env-1')
})
it('stops at the next slice once the environment is manually disconnected', async () => {
const filePath = join(workDir, 'disconnect.bin')
await writeFile(filePath, Buffer.alloc(RUNTIME_UPLOAD_SLICE_BYTES * 3, 7))
callRuntimeEnvironment.mockImplementation(async (_u, _e, method) => {
if (method === 'files.writeBase64Chunk' && chunkCalls().length === 1) {
markRuntimeEnvironmentManuallyDisconnected('env-1')
}
return { id: 'x', ok: true, result: {}, _meta: {} }
})
await expect(streamExternalFileToRuntime(await baseArgs(filePath))).rejects.toThrow(
RUNTIME_MANUALLY_DISCONNECTED_MESSAGE
)
expect(chunkCalls()).toHaveLength(1)
})
it('refuses the first slice when the environment is already disconnected', async () => {
const filePath = join(workDir, 'disconnected.bin')
await writeFile(filePath, Buffer.alloc(16, 1))
markRuntimeEnvironmentManuallyDisconnected('env-1')
await expect(streamExternalFileToRuntime(await baseArgs(filePath))).rejects.toThrow(
RUNTIME_MANUALLY_DISCONNECTED_MESSAGE
)
expect(chunkCalls()).toHaveLength(0)
})
})
+213
View File
@@ -0,0 +1,213 @@
import { constants, type Stats } from 'node:fs'
import { lstat, open, realpath } from 'node:fs/promises'
import { basename, isAbsolute, join, relative, resolve, sep } from 'node:path'
import type {
RuntimeUploadFileStreamRequest,
StagedRuntimeUploadFileIdentity
} from '../../shared/runtime-upload-staging-contract'
import { authorizeExternalPath } from './filesystem-auth'
import { formatByteCeiling, REMOTE_IMPORT_MAX_FILE_BYTES } from './runtime-import-limits'
import {
isRuntimeEnvironmentManuallyDisconnected,
RUNTIME_MANUALLY_DISCONNECTED_MESSAGE
} from './runtime-environment-manual-disconnect'
import { callRuntimeEnvironment } from './runtime-environment-transport-routing'
// Why: base64 turns 3 bytes into 4 chars, so a 384 KiB slice lands on the wire
// as exactly 512 KiB — the chunk size the renderer used before streaming.
export const RUNTIME_UPLOAD_SLICE_BYTES = 384 * 1024
const RUNTIME_UPLOAD_CHUNK_TIMEOUT_MS = 30_000
export type RuntimeUploadFileStreamArgs = RuntimeUploadFileStreamRequest & {
/** Resolved environment id, not a selector: the manual-disconnect check keys on it. */
environmentId: string
userDataPath: string
/** Aborts the transfer; the caller's lifetime is what raises it today. */
signal?: AbortSignal
}
/**
* Stream one client-local file to a runtime environment in slices.
*
* Replaces reading the whole file into memory and base64-encoding it before the
* first byte moves. Peak memory is one slice, so imports are no longer bounded
* by main-process heap.
*/
export async function streamExternalFileToRuntime(
args: RuntimeUploadFileStreamArgs
): Promise<{ byteLength: number }> {
const sourcePath = resolveEntrySourcePath(args.sourceRootPath, args.entryRelativePath)
// Why: parity with staging — an OS drop authorizes the paths it hands over.
authorizeExternalPath(sourcePath)
// Why: relativePath is the hidden .orca-upload-<nonce> temp destination, so a
// dropped file names its source instead of a path the user never chose.
const displayPath = args.entryRelativePath || basename(args.sourceRootPath)
const lstatResult = await lstat(sourcePath)
if (lstatResult.isSymbolicLink()) {
throw new Error(`Symlink not allowed in '${displayPath}'`)
}
if (!lstatResult.isFile()) {
throw new Error(`Unsupported file type in '${displayPath}'`)
}
if (args.entryRelativePath) {
await assertEntryInsideRoot(args.sourceRootPath, sourcePath, displayPath)
}
assertMatchesStagedIdentity(lstatResult, args.expected, displayPath)
args.signal?.throwIfAborted()
const handle = await open(sourcePath, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0))
try {
const openedStat = await handle.stat()
if (!openedStat.isFile()) {
throw new Error(`Unsupported file type in '${displayPath}'`)
}
if (!isSameFile(openedStat, lstatResult)) {
throw new Error(`File changed during upload: '${displayPath}'`)
}
// Why: the handle is what the slices are read from, so the staged identity
// has to hold here too — checking only the pre-open lstat leaves a window
// where the path is swapped between lstat and open.
assertMatchesStagedIdentity(openedStat, args.expected, displayPath)
const totalBytes = openedStat.size
// Why: enforced again where the bytes actually move. Staging is a separate
// call, so the ceiling only holds here if this boundary checks it too.
if (totalBytes > REMOTE_IMPORT_MAX_FILE_BYTES) {
throw new Error(
`'${displayPath}' is ${formatByteCeiling(totalBytes)}, over the ` +
`${formatByteCeiling(REMOTE_IMPORT_MAX_FILE_BYTES)} per-file remote import limit`
)
}
if (totalBytes === 0) {
// Why: a zero-byte source produces no slices, but the destination still
// has to exist before commitUpload renames it into place.
await sendChunk(args, '', false)
} else {
const buffer = Buffer.allocUnsafe(Math.min(RUNTIME_UPLOAD_SLICE_BYTES, totalBytes))
let offset = 0
while (offset < totalBytes) {
// Why: checked per slice, so an abort stops the transfer at the next
// boundary instead of after the whole file has moved.
args.signal?.throwIfAborted()
const { bytesRead } = await handle.read(buffer, 0, buffer.byteLength, offset)
if (bytesRead === 0) {
throw new Error(`File truncated during upload: '${displayPath}'`)
}
await sendChunk(args, buffer.subarray(0, bytesRead).toString('base64'), offset > 0)
offset += bytesRead
}
}
// Why: the destination is a temp path the caller commits, so a source
// rewritten mid-transfer is caught before anything lands at the final path.
// mtime catches an in-place edit that kept the size. An empty source runs
// this too: its chunk is still a round trip the source can change during.
const afterReadStat = await handle.stat()
if (afterReadStat.mtimeMs !== openedStat.mtimeMs || !isSameFile(afterReadStat, openedStat)) {
throw new Error(`File changed during upload: '${displayPath}'`)
}
return { byteLength: totalBytes }
} finally {
await handle.close()
}
}
/**
* Refuse a source that no longer matches what staging measured.
*
* Inode and device are compared only when both sides report one, because some
* filesystems leave them at 0; size and mtime then carry the check alone.
*/
function assertMatchesStagedIdentity(
observed: Stats,
expected: StagedRuntimeUploadFileIdentity,
displayPath: string
): void {
const changed =
observed.size !== expected.byteLength ||
observed.mtimeMs !== expected.modifiedAtMs ||
(expected.inode !== 0 && observed.ino !== 0 && observed.ino !== expected.inode) ||
(expected.deviceId !== 0 && observed.dev !== 0 && observed.dev !== expected.deviceId)
if (changed) {
throw new Error(`File changed since it was staged: '${displayPath}'`)
}
}
/** Same inode on the same device, where the filesystem reports them. */
function isSameFile(a: Stats, b: Stats): boolean {
return (
a.size === b.size &&
(a.ino === 0 || b.ino === 0 || a.ino === b.ino) &&
(a.dev === 0 || b.dev === 0 || a.dev === b.dev)
)
}
/** Append one base64 slice, carrying the host guards that must hold per chunk. */
async function sendChunk(
args: RuntimeUploadFileStreamArgs,
contentBase64: string,
append: boolean
): Promise<void> {
// Why: the renderer's per-chunk calls went through an IPC handler that refuses
// a manually disconnected environment. The loop lives in main now, so it makes
// the same check, or a disconnect mid-upload keeps pushing bytes to that host.
if (isRuntimeEnvironmentManuallyDisconnected(args.environmentId)) {
throw new Error(RUNTIME_MANUALLY_DISCONNECTED_MESSAGE)
}
const response = await callRuntimeEnvironment(
args.userDataPath,
args.environmentId,
'files.writeBase64Chunk',
{
worktree: args.worktree,
relativePath: args.relativePath,
contentBase64,
append,
expectedSshTargetId: args.expectedSshTargetId,
expectedSshConnectionGeneration: args.expectedSshConnectionGeneration,
expectedExecutionHostId: args.expectedExecutionHostId
},
RUNTIME_UPLOAD_CHUNK_TIMEOUT_MS,
// Why: re-checked per chunk, so a re-pair mid-upload aborts instead of
// appending the rest of the file on a different host.
args.expectedEnvironmentPairingRevision,
undefined,
{
// Why: a replacement runtime keeps the pairing but invalidates its
// predecessor's capability proof, so the identity rides every chunk too.
expectedEnvironmentRuntimeId: args.expectedEnvironmentRuntimeId,
signal: args.signal
}
)
if (response.ok !== true) {
throw new Error(response.error.message || response.error.code)
}
}
function resolveEntrySourcePath(sourceRootPath: string, entryRelativePath: string): string {
// Why: staging resolves before authorizing, so the streamer has to agree on
// the same absolute path or the two checks can disagree.
const root = resolve(sourceRootPath)
return entryRelativePath ? join(root, entryRelativePath) : root
}
async function assertEntryInsideRoot(
sourceRootPath: string,
candidatePath: string,
displayPath: string
): Promise<void> {
const rootRealPath = await realpath(sourceRootPath)
const candidateRealPath = await realpath(candidatePath)
const relativeToRoot = relative(rootRealPath, candidateRealPath)
// Why: `..name` is a valid child path; only `..` and `../...` escape.
if (
relativeToRoot !== '' &&
(relativeToRoot === '..' || relativeToRoot.startsWith(`..${sep}`) || isAbsolute(relativeToRoot))
) {
throw new Error(`Path escaped upload root during upload: '${displayPath}'`)
}
}
@@ -0,0 +1,383 @@
import {
appendFile,
mkdir,
mkdtemp,
readFile,
rm,
stat,
truncate,
writeFile
} from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { FileWriteBase64Chunk } from '../../shared/rpc-contract/files-mutation-params'
import type { StagedRuntimeUploadFileIdentity } from '../../shared/runtime-upload-staging-contract'
// Why: real limits, real host write flags ('wx' then 'a') and the real chunk
// schema — the slice loop is exercised exactly at the boundaries it must respect.
vi.mock('./filesystem-auth', () => ({ authorizeExternalPath: () => {} }))
type ChunkParams = { relativePath: string; contentBase64: string; append: boolean }
type CallOptions = { expectedEnvironmentRuntimeId?: string; signal?: AbortSignal }
type CallArgs = [
userDataPath: string,
environmentId: string,
method: string,
params: ChunkParams,
timeoutMs?: number,
expectedEnvironmentPairingRevision?: number,
envelope?: unknown,
options?: CallOptions
]
const callRuntimeEnvironment = vi.fn<(...args: CallArgs) => Promise<unknown>>()
// Why: vi.fn retains every call's params; a 2 GiB stream would pin ~2.8 GB of
// base64 in mock.calls and masquerade as a leak. Big tests swap in a plain fn.
let transportImpl: (...args: CallArgs) => Promise<unknown> = (...args) =>
callRuntimeEnvironment(...args)
vi.mock('./runtime-environment-transport-routing', () => ({
callRuntimeEnvironment: (...args: CallArgs) => transportImpl(...args)
}))
const { RUNTIME_UPLOAD_SLICE_BYTES, streamExternalFileToRuntime } =
await import('./runtime-upload-file-stream')
const { stageOneSourceForRuntimeUpload } = await import('./filesystem-runtime-upload-staging')
const { REMOTE_IMPORT_MAX_FILE_BYTES, REMOTE_IMPORT_MAX_TOTAL_BYTES, formatByteCeiling } =
await import('./runtime-import-limits')
const SLICE = RUNTIME_UPLOAD_SLICE_BYTES
const WIRE_CHUNK_CHARS = 512 * 1024
const OK = { id: 'x', ok: true, result: {}, _meta: {} }
let workDir: string
let remoteDir: string
beforeEach(async () => {
workDir = await mkdtemp(join(tmpdir(), 'orca-upload-bounds-'))
remoteDir = join(workDir, 'remote')
await mkdir(remoteDir)
callRuntimeEnvironment.mockReset()
callRuntimeEnvironment.mockResolvedValue(OK)
transportImpl = (...args) => callRuntimeEnvironment(...args)
})
afterEach(async () => {
await rm(workDir, { force: true, recursive: true })
})
function chunkCalls(): ChunkParams[] {
return callRuntimeEnvironment.mock.calls
.filter(([, , method]) => method === 'files.writeBase64Chunk')
.map(([, , , params]) => params)
}
/** Mirrors the host: first chunk is an exclusive create, appends open with 'a'. */
function installRealHostWrites(): void {
callRuntimeEnvironment.mockImplementation(async (_u, _e, method, params) => {
if (method === 'files.writeBase64Chunk') {
const parsed = FileWriteBase64Chunk.parse({ worktree: 'wt-1', ...params })
await writeFile(
join(remoteDir, parsed.relativePath),
Buffer.from(parsed.contentBase64, 'base64'),
{
flag: parsed.append ? 'a' : 'wx'
}
)
}
return OK
})
}
async function identityOf(path: string): Promise<StagedRuntimeUploadFileIdentity> {
const s = await stat(path)
return { byteLength: s.size, inode: s.ino, deviceId: s.dev, modifiedAtMs: s.mtimeMs }
}
async function argsFor(sourceRootPath: string, entryRelativePath = '', relativePath = 'dest.tmp') {
const target = entryRelativePath ? join(sourceRootPath, entryRelativePath) : sourceRootPath
return {
userDataPath: '/user-data',
environmentId: 'env-1',
sourceRootPath,
entryRelativePath,
expected: await identityOf(target),
worktree: 'wt-1',
relativePath,
expectedEnvironmentPairingRevision: 7,
expectedEnvironmentRuntimeId: 'rt-1'
}
}
function patterned(size: number, seed: number): Buffer {
const buffer = Buffer.allocUnsafe(size)
for (let i = 0; i < size; i += 1) {
buffer[i] = (i * 31 + seed) & 0xff
}
return buffer
}
describe('slice boundaries', () => {
const sizes = [
1,
2,
3,
4,
SLICE - 1,
SLICE,
SLICE + 1,
2 * SLICE - 1,
2 * SLICE,
2 * SLICE + 1,
3 * SLICE + 7
]
for (const size of sizes) {
it(`streams ${size} bytes as ceil(size/slice) schema-valid chunks that the host reassembles exactly`, async () => {
installRealHostWrites()
const contents = patterned(size, size)
const source = join(workDir, `s-${size}.bin`)
await writeFile(source, contents)
const dest = `dest-${size}.tmp`
await expect(streamExternalFileToRuntime(await argsFor(source, '', dest))).resolves.toEqual({
byteLength: size
})
const calls = chunkCalls()
const expectedChunks = Math.ceil(size / SLICE)
expect(calls).toHaveLength(expectedChunks)
expect(calls.map((c) => c.append)).toEqual(calls.map((_, i) => i > 0))
for (const [index, call] of calls.entries()) {
const isLast = index === calls.length - 1
expect(call.contentBase64.length).toBeLessThanOrEqual(WIRE_CHUNK_CHARS)
if (!isLast) {
expect(call.contentBase64.length).toBe(WIRE_CHUNK_CHARS)
}
expect(call.relativePath).toBe(dest)
}
const remote = await readFile(join(remoteDir, dest))
expect(remote.equals(contents)).toBe(true)
})
}
it('sends a zero-byte file as one empty exclusive create the host schema accepts', async () => {
installRealHostWrites()
const source = join(workDir, 'empty.bin')
await writeFile(source, '')
await expect(
streamExternalFileToRuntime(await argsFor(source, '', 'empty.tmp'))
).resolves.toEqual({
byteLength: 0
})
expect(chunkCalls()).toHaveLength(1)
expect(chunkCalls()[0]).toMatchObject({
relativePath: 'empty.tmp',
contentBase64: '',
append: false
})
expect((await stat(join(remoteDir, 'empty.tmp'))).size).toBe(0)
})
it('carries the pairing revision, runtime id and signal on every chunk', async () => {
const source = join(workDir, 'guards.bin')
await writeFile(source, patterned(2 * SLICE + 1, 3))
await streamExternalFileToRuntime(await argsFor(source))
const chunkInvocations = callRuntimeEnvironment.mock.calls.filter(
([, , method]) => method === 'files.writeBase64Chunk'
)
expect(chunkInvocations).toHaveLength(3)
for (const [, environmentId, , , timeoutMs, revision, envelope, options] of chunkInvocations) {
expect(environmentId).toBe('env-1')
expect(timeoutMs).toBe(30_000)
expect(revision).toBe(7)
expect(envelope).toBeUndefined()
expect(options?.expectedEnvironmentRuntimeId).toBe('rt-1')
}
})
})
describe('staging → streaming end to end on a real filesystem', () => {
it('streams every staged entry of a dropped directory using the identity staging recorded', async () => {
installRealHostWrites()
const root = join(workDir, 'drop me')
await mkdir(join(root, 'sub', 'deeper'), { recursive: true })
const files: Record<string, Buffer> = {
'a.txt': Buffer.from('alpha'),
'..keep': Buffer.from('dot-dot-prefixed name is a valid child'),
'héllo wörld.bin': patterned(SLICE, 9),
'sub/empty': Buffer.alloc(0),
'sub/deeper/big.bin': patterned(2 * SLICE + 5, 11)
}
for (const [rel, body] of Object.entries(files)) {
await writeFile(join(root, rel), body)
}
const staged = await stageOneSourceForRuntimeUpload(root)
expect(staged.status).toBe('staged')
if (staged.status !== 'staged') {
return
}
const fileEntries = staged.entries.filter((e) => e.kind === 'file')
expect(fileEntries.map((e) => e.relativePath).sort()).toEqual(Object.keys(files).sort())
for (const entry of fileEntries) {
if (entry.kind !== 'file') {
continue
}
const dest = `up-${entry.relativePath.replace(/[^a-z0-9]/gi, '_')}.tmp`
await expect(
streamExternalFileToRuntime({
userDataPath: '/u',
environmentId: 'env-1',
sourceRootPath: staged.sourcePath,
entryRelativePath: entry.relativePath,
expected: {
byteLength: entry.byteLength,
inode: entry.inode,
deviceId: entry.deviceId,
modifiedAtMs: entry.modifiedAtMs
},
worktree: 'wt-1',
relativePath: dest
})
).resolves.toEqual({ byteLength: files[entry.relativePath]!.length })
const remote = await readFile(join(remoteDir, dest))
expect(remote.equals(files[entry.relativePath]!)).toBe(true)
}
})
it('streams a dropped single file using the identity staging recorded', async () => {
installRealHostWrites()
const source = join(workDir, 'single.bin')
const body = patterned(SLICE + 1, 5)
await writeFile(source, body)
const staged = await stageOneSourceForRuntimeUpload(source)
expect(staged.status).toBe('staged')
if (staged.status !== 'staged') {
return
}
const entry = staged.entries[0]!
expect(entry.kind).toBe('file')
if (entry.kind !== 'file') {
return
}
await expect(
streamExternalFileToRuntime({
userDataPath: '/u',
environmentId: 'env-1',
sourceRootPath: staged.sourcePath,
entryRelativePath: entry.relativePath,
expected: entry,
worktree: 'wt-1',
relativePath: 'single.tmp'
})
).resolves.toEqual({ byteLength: body.length })
expect((await readFile(join(remoteDir, 'single.tmp'))).equals(body)).toBe(true)
})
})
describe('source mutation during transfer', () => {
it('rejects a source that grows during the transfer and never claims success', async () => {
const source = join(workDir, 'growing.bin')
await writeFile(source, patterned(2 * SLICE, 1))
const args = await argsFor(source)
callRuntimeEnvironment.mockImplementation(async (_u, _e, method) => {
if (method === 'files.writeBase64Chunk' && chunkCalls().length === 1) {
await appendFile(source, 'extra')
}
return OK
})
await expect(streamExternalFileToRuntime(args)).rejects.toThrow(
"File changed during upload: 'growing.bin'"
)
})
it('rejects a source truncated during the transfer instead of sending a short file', async () => {
const source = join(workDir, 'shrinking.bin')
await writeFile(source, patterned(3 * SLICE, 2))
const args = await argsFor(source)
callRuntimeEnvironment.mockImplementation(async (_u, _e, method) => {
if (method === 'files.writeBase64Chunk' && chunkCalls().length === 1) {
await truncate(source, SLICE)
}
return OK
})
await expect(streamExternalFileToRuntime(args)).rejects.toThrow(
"File truncated during upload: 'shrinking.bin'"
)
expect(chunkCalls().length).toBeLessThan(3)
})
it('accepts a staged identity whose inode and device are unreported (0) when size and mtime match', async () => {
const source = join(workDir, 'no-ino.bin')
await writeFile(source, patterned(10, 4))
const args = await argsFor(source)
args.expected = { ...args.expected, inode: 0, deviceId: 0 }
await expect(streamExternalFileToRuntime(args)).resolves.toEqual({ byteLength: 10 })
})
it('still refuses a wrong inode when only the device is unreported', async () => {
const source = join(workDir, 'wrong-ino.bin')
await writeFile(source, patterned(10, 4))
const args = await argsFor(source)
args.expected = { ...args.expected, inode: args.expected.inode + 1, deviceId: 0 }
await expect(streamExternalFileToRuntime(args)).rejects.toThrow(
"File changed since it was staged: 'wrong-ino.bin'"
)
expect(chunkCalls()).toHaveLength(0)
})
it('stops before the next slice when the signal aborts while a chunk is in flight', async () => {
const source = join(workDir, 'abort.bin')
await writeFile(source, patterned(3 * SLICE, 6))
const controller = new AbortController()
callRuntimeEnvironment.mockImplementation(async (_u, _e, method, _p, _t, _r, _env, options) => {
if (method !== 'files.writeBase64Chunk') {
return OK
}
if (chunkCalls().length === 2) {
return new Promise((_resolve, reject) => {
options?.signal?.addEventListener('abort', () => reject(options.signal?.reason), {
once: true
})
controller.abort(new Error('window gone'))
})
}
return OK
})
await expect(
streamExternalFileToRuntime({ ...(await argsFor(source)), signal: controller.signal })
).rejects.toThrow('window gone')
expect(chunkCalls()).toHaveLength(2)
})
})
describe('formatByteCeiling bounds', () => {
it.each([
[0, '0 B'],
[1, '1 B'],
[1023, '1023 B'],
[1024, '1 KB'],
[1025, '1.1 KB'],
[25 * 1024 * 1024, '25 MB'],
[25 * 1024 * 1024 + 1, '25.1 MB'],
[REMOTE_IMPORT_MAX_FILE_BYTES, '2 GB'],
[REMOTE_IMPORT_MAX_FILE_BYTES + 1, '2.1 GB'],
[REMOTE_IMPORT_MAX_TOTAL_BYTES, '8 GB'],
[REMOTE_IMPORT_MAX_TOTAL_BYTES + 1, '8.1 GB'],
[1024 ** 5, '1024 TB']
])('%i → %s', (bytes, text) => {
expect(formatByteCeiling(bytes)).toBe(text)
})
})
@@ -0,0 +1,94 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import type { RuntimeUploadFileStreamRequest } from '../../shared/runtime-upload-staging-contract'
const callRuntimeEnvironment =
vi.fn<
(
userDataPath: string,
environmentId: string,
method: string,
params: { relativePath: string; recursive: boolean },
timeoutMs?: number,
expectedEnvironmentPairingRevision?: number,
envelope?: unknown,
options?: { expectedEnvironmentRuntimeId?: string }
) => unknown
>()
vi.mock('./runtime-environment-transport-routing', () => ({
callRuntimeEnvironment: (...args: Parameters<typeof callRuntimeEnvironment>) =>
callRuntimeEnvironment(...args)
}))
const { sweepAbandonedRuntimeUploadTempPath } = await import('./runtime-upload-temp-sweep')
const request: RuntimeUploadFileStreamRequest = {
environmentId: 'env-1',
sourceRootPath: '/Users/me/clip.mp4',
entryRelativePath: '',
expected: { byteLength: 4, inode: 1, deviceId: 2, modifiedAtMs: 3 },
worktree: 'id:wt-1',
relativePath: 'uploads/.clip.mp4.orca-upload-abc',
expectedEnvironmentPairingRevision: 17,
expectedEnvironmentRuntimeId: 'runtime-7',
expectedExecutionHostId: 'local'
}
function deleteCalls(): { relativePath: string; recursive: boolean }[] {
return callRuntimeEnvironment.mock.calls
.filter(([, , method]) => method === 'files.delete')
.map(([, , , params]) => params)
}
beforeEach(() => {
vi.useFakeTimers()
callRuntimeEnvironment.mockReset()
callRuntimeEnvironment.mockResolvedValue({ id: 'x', ok: true, result: {}, _meta: {} })
})
afterEach(() => {
vi.useRealTimers()
})
describe('sweepAbandonedRuntimeUploadTempPath', () => {
it('deletes twice, because a straggling append recreates the file with flag a', async () => {
const swept = sweepAbandonedRuntimeUploadTempPath('/user-data', request)
await vi.runAllTimersAsync()
await swept
expect(deleteCalls()).toEqual([
expect.objectContaining({ relativePath: request.relativePath, recursive: false }),
expect.objectContaining({ relativePath: request.relativePath, recursive: false })
])
})
it('still makes the second pass when the first one fails', async () => {
callRuntimeEnvironment.mockRejectedValueOnce(new Error('connection lost'))
const swept = sweepAbandonedRuntimeUploadTempPath('/user-data', request)
await vi.runAllTimersAsync()
await expect(swept).resolves.toBeUndefined()
expect(deleteCalls()).toHaveLength(2)
})
it('carries the host ownership guards so it cannot delete on a re-paired host', async () => {
const swept = sweepAbandonedRuntimeUploadTempPath('/user-data', request)
await vi.runAllTimersAsync()
await swept
for (const call of callRuntimeEnvironment.mock.calls) {
expect(call[5]).toBe(17)
expect(call[7]?.expectedEnvironmentRuntimeId).toBe('runtime-7')
}
})
it('never rejects, so cleanup cannot mask the upload failure', async () => {
callRuntimeEnvironment.mockRejectedValue(new Error('runtime gone'))
const swept = sweepAbandonedRuntimeUploadTempPath('/user-data', request)
await vi.runAllTimersAsync()
await expect(swept).resolves.toBeUndefined()
})
})
+49
View File
@@ -0,0 +1,49 @@
import { setTimeout } from 'node:timers/promises'
import type { RuntimeUploadFileStreamRequest } from '../../shared/runtime-upload-staging-contract'
import { callRuntimeEnvironment } from './runtime-environment-transport-routing'
const RUNTIME_UPLOAD_SWEEP_ATTEMPTS = 2
const RUNTIME_UPLOAD_SWEEP_SETTLE_MS = 250
/**
* Sweep an abandoned upload temp path after an abort.
*
* Aborting rejects the in-flight chunk locally, but the host may still apply
* that append — and appends open with `flag: 'a'`, which recreates the file a
* delete just removed. Slices are strictly sequential, so at most one append
* can be outstanding: a second pass after it has had time to land is enough.
*
* Best-effort throughout. The runtime may be why the upload failed, and a
* failed cleanup of a hidden temp file is not actionable.
*/
export async function sweepAbandonedRuntimeUploadTempPath(
userDataPath: string,
args: RuntimeUploadFileStreamRequest
): Promise<void> {
for (let attempt = 0; attempt < RUNTIME_UPLOAD_SWEEP_ATTEMPTS; attempt += 1) {
if (attempt > 0) {
await setTimeout(RUNTIME_UPLOAD_SWEEP_SETTLE_MS)
}
try {
await callRuntimeEnvironment(
userDataPath,
args.environmentId,
'files.delete',
{
worktree: args.worktree,
relativePath: args.relativePath,
recursive: false,
expectedSshTargetId: args.expectedSshTargetId,
expectedSshConnectionGeneration: args.expectedSshConnectionGeneration,
expectedExecutionHostId: args.expectedExecutionHostId
},
15_000,
args.expectedEnvironmentPairingRevision,
undefined,
{ expectedEnvironmentRuntimeId: args.expectedEnvironmentRuntimeId }
)
} catch {
// Nothing to escalate; the next pass (if any) still runs.
}
}
}
+10 -24
View File
@@ -12,6 +12,10 @@ import type {
LocalLogTailWatchArgs
} from '../../shared/local-log-tail-types'
import type { SshMutationExpectation } from '../../shared/ssh-types'
import type {
RuntimeUploadFileStreamRequest,
StageRuntimeUploadResult
} from '../../shared/runtime-upload-staging-contract'
export type ExportApi = {
htmlToPdf: (args: {
@@ -155,30 +159,12 @@ export type FilesystemApi = {
}
)[]
}>
stageExternalPathsForRuntimeUpload: (args: { sourcePaths: string[] }) => Promise<{
sources: (
| {
sourcePath: string
status: 'staged'
name: string
kind: 'file' | 'directory'
entries: (
| { relativePath: string; kind: 'directory' }
| { relativePath: string; kind: 'file'; contentBase64: string }
)[]
}
| {
sourcePath: string
status: 'skipped'
reason: 'missing' | 'symlink' | 'permission-denied' | 'unsupported'
}
| {
sourcePath: string
status: 'failed'
reason: string
}
)[]
}>
stageExternalPathsForRuntimeUpload: (args: {
sourcePaths: string[]
}) => Promise<StageRuntimeUploadResult>
uploadExternalFileToRuntime: (
args: RuntimeUploadFileStreamRequest
) => Promise<{ byteLength: number }>
resolveDroppedPathsForAgent: (
args: {
paths: string[]
+9 -24
View File
@@ -1,6 +1,10 @@
import type { PathExistenceResult } from '../../shared/path-existence-batch'
import { ipcRenderer } from 'electron'
import type { SshMutationExpectation } from '../../shared/ssh-types'
import type {
RuntimeUploadFileStreamRequest,
StageRuntimeUploadResult
} from '../../shared/runtime-upload-staging-contract'
import type { SearchResult } from '../../shared/code-search-types'
import type { FsChangedPayload } from '../../shared/filesystem-entry-types'
import type {
@@ -174,30 +178,11 @@ export const fsApi = {
}> => ipcRenderer.invoke('fs:importExternalPaths', args),
stageExternalPathsForRuntimeUpload: (args: {
sourcePaths: string[]
}): Promise<{
sources: (
| {
sourcePath: string
status: 'staged'
name: string
kind: 'file' | 'directory'
entries: (
| { relativePath: string; kind: 'directory' }
| { relativePath: string; kind: 'file'; contentBase64: string }
)[]
}
| {
sourcePath: string
status: 'skipped'
reason: 'missing' | 'symlink' | 'permission-denied' | 'unsupported'
}
| {
sourcePath: string
status: 'failed'
reason: string
}
)[]
}> => ipcRenderer.invoke('fs:stageExternalPathsForRuntimeUpload', args),
}): Promise<StageRuntimeUploadResult> =>
ipcRenderer.invoke('fs:stageExternalPathsForRuntimeUpload', args),
uploadExternalFileToRuntime: (
args: RuntimeUploadFileStreamRequest
): Promise<{ byteLength: number }> => ipcRenderer.invoke('fs:uploadExternalFileToRuntime', args),
resolveDroppedPathsForAgent: (
args: {
paths: string[]
@@ -4,6 +4,7 @@ import { replaceRuntimeEnvironmentRevisions } from './runtime-environment-revisi
import {
fsImportExternalPaths,
fsStageExternalPathsForRuntimeUpload,
fsUploadExternalFileToRuntime,
runtimeEnvironmentCall,
runtimeEnvironmentTransportCall,
installRuntimeFileClientEnvironment
@@ -11,11 +12,65 @@ import {
installRuntimeFileClientEnvironment()
const okResponse = (id: string): unknown => ({
id,
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
const notFoundResponse = (id: string): unknown => ({
id,
ok: false,
error: { code: 'not_found', message: 'not found' },
_meta: { runtimeId: 'remote-runtime' }
})
/** Matches what main-process staging now records for a file entry. */
const stagedFile = (
relativePath: string,
byteLength: number,
inode: number
): Record<string, unknown> => ({
relativePath,
kind: 'file',
byteLength,
inode,
deviceId: 66,
modifiedAtMs: 1_700_000_000_000
})
/** The upload request main receives; `never[]` mock args widen to it without a cast. */
type UploadRequest = {
environmentId: string
sourceRootPath: string
entryRelativePath: string
expected: Record<string, unknown>
worktree: string
relativePath: string
expectedExecutionHostId?: string
expectedSshTargetId?: string
expectedSshConnectionGeneration?: number
expectedEnvironmentPairingRevision?: number
expectedEnvironmentRuntimeId?: string
}
function uploadRequests(): UploadRequest[] {
return fsUploadExternalFileToRuntime.mock.calls.flat()
}
const identityOf = (entry: Record<string, unknown>): Record<string, unknown> => ({
byteLength: entry.byteLength,
inode: entry.inode,
deviceId: entry.deviceId,
modifiedAtMs: entry.modifiedAtMs
})
describe('runtime file client', () => {
it('uploads a staged directory after one ownership and one cold compatibility preflight', async () => {
it('streams staged directory entries through main instead of sending base64 itself', async () => {
replaceRuntimeEnvironmentRevisions([{ id: 'env-1', createdAt: 1, pairingRevision: 17 }])
const firstChunk = 'A'.repeat(512 * 1024)
const secondChunk = 'BBBBBBBB'
const logo = stagedFile('logo.png', 3, 101)
const large = stagedFile('large.bin', 40 * 1024 * 1024, 102)
fsStageExternalPathsForRuntimeUpload.mockResolvedValue({
sources: [
{
@@ -23,85 +78,19 @@ describe('runtime file client', () => {
status: 'staged',
name: 'assets',
kind: 'directory',
entries: [
{ relativePath: '', kind: 'directory' },
{ relativePath: 'logo.png', kind: 'file', contentBase64: 'cG5n' },
{
relativePath: 'large.bin',
kind: 'file',
contentBase64: `${firstChunk}${secondChunk}`
}
]
entries: [{ relativePath: '', kind: 'directory' }, logo, large]
}
]
})
runtimeEnvironmentCall
.mockResolvedValueOnce({
id: 'stat-destination-miss',
ok: false,
error: { code: 'not_found', message: 'not found' },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'create-destination-dir',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'stat-miss',
ok: false,
error: { code: 'not_found', message: 'not found' },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'create-dir',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'write-file',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'commit-upload',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'delete-temp',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'write-chunk-1',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'write-chunk-2',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'commit-large-upload',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'delete-large-temp',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce(notFoundResponse('stat-destination-miss'))
.mockResolvedValueOnce(okResponse('create-destination-dir'))
.mockResolvedValueOnce(notFoundResponse('stat-miss'))
.mockResolvedValueOnce(okResponse('create-dir'))
.mockResolvedValueOnce(okResponse('commit-upload'))
.mockResolvedValueOnce(okResponse('delete-temp'))
.mockResolvedValueOnce(okResponse('commit-large-upload'))
.mockResolvedValueOnce(okResponse('delete-large-temp'))
await expect(
importExternalPathsToRuntime(
@@ -136,75 +125,48 @@ describe('runtime file client', () => {
'files.createDir',
'files.stat',
'files.createDirNoClobber',
'files.writeBase64',
'files.commitUpload',
'files.delete',
'files.writeBase64Chunk',
'files.writeBase64Chunk',
'files.commitUpload',
'files.delete'
])
expect(transportCalls.filter((args) => args.method === 'status.get')).toHaveLength(2)
// Why: the whole point of the change — no file body crosses this boundary.
expect(transportCalls.some((args) => String(args.method).startsWith('files.writeBase64'))).toBe(
false
)
expect(transportCalls.every((args) => args.expectedEnvironmentPairingRevision === 17)).toBe(
true
)
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(1, {
selector: 'env-1',
method: 'files.stat',
params: {
worktree: 'id:wt-1',
relativePath: 'uploads'
},
timeoutMs: 15_000,
expectedEnvironmentPairingRevision: 17
})
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(2, {
selector: 'env-1',
method: 'files.createDir',
params: {
worktree: 'id:wt-1',
relativePath: 'uploads',
expectedExecutionHostId: 'local'
},
timeoutMs: 15_000,
const uploads = uploadRequests()
expect(uploads).toHaveLength(2)
expect(uploads[0]?.relativePath).toMatch(/^uploads\/assets\/\.logo\.png\.orca-upload-/)
expect(uploads[0]).toEqual({
environmentId: 'env-1',
sourceRootPath: '/Users/me/assets',
entryRelativePath: 'logo.png',
expected: identityOf(logo),
worktree: 'id:wt-1',
relativePath: uploads[0]?.relativePath,
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
expectedSshConnectionGeneration: undefined,
expectedEnvironmentPairingRevision: 17,
expectedEnvironmentRuntimeId: 'remote-runtime'
})
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(3, {
selector: 'env-1',
method: 'files.stat',
params: {
worktree: 'id:wt-1',
relativePath: 'uploads/assets'
},
timeoutMs: 15_000,
expectedEnvironmentPairingRevision: 17
expect(uploads[1]).toMatchObject({
entryRelativePath: 'large.bin',
expected: identityOf(large)
})
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(4, {
selector: 'env-1',
method: 'files.createDirNoClobber',
params: {
worktree: 'id:wt-1',
relativePath: 'uploads/assets',
expectedExecutionHostId: 'local'
},
timeoutMs: 15_000,
expectedEnvironmentPairingRevision: 17,
expectedEnvironmentRuntimeId: 'remote-runtime'
})
const smallWriteCall = runtimeEnvironmentCall.mock.calls[4]?.[0] as {
params: { relativePath: string }
}
expect(smallWriteCall.params.relativePath).toMatch(
/^uploads\/assets\/\.logo\.png\.orca-upload-/
)
expect(uploads[1]?.relativePath).toMatch(/^uploads\/assets\/\.large\.bin\.orca-upload-/)
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(5, {
selector: 'env-1',
method: 'files.writeBase64',
method: 'files.commitUpload',
params: {
worktree: 'id:wt-1',
relativePath: smallWriteCall.params.relativePath,
contentBase64: 'cG5n',
tempRelativePath: uploads[0]?.relativePath,
finalRelativePath: 'uploads/assets/logo.png',
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
expectedSshConnectionGeneration: undefined
@@ -214,99 +176,11 @@ describe('runtime file client', () => {
expectedEnvironmentRuntimeId: 'remote-runtime'
})
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(6, {
selector: 'env-1',
method: 'files.commitUpload',
params: {
worktree: 'id:wt-1',
tempRelativePath: smallWriteCall.params.relativePath,
finalRelativePath: 'uploads/assets/logo.png',
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
expectedSshConnectionGeneration: undefined
},
timeoutMs: 30_000,
expectedEnvironmentPairingRevision: 17,
expectedEnvironmentRuntimeId: 'remote-runtime'
})
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(7, {
selector: 'env-1',
method: 'files.delete',
params: {
worktree: 'id:wt-1',
relativePath: smallWriteCall.params.relativePath,
recursive: false,
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
expectedSshConnectionGeneration: undefined
},
timeoutMs: 15_000,
expectedEnvironmentPairingRevision: 17,
expectedEnvironmentRuntimeId: 'remote-runtime'
})
const largeWriteParams = runtimeEnvironmentCall.mock.calls[7]?.[0].params
if (
typeof largeWriteParams !== 'object' ||
largeWriteParams === null ||
!('relativePath' in largeWriteParams) ||
typeof largeWriteParams.relativePath !== 'string'
) {
throw new Error('missing large file write call')
}
const largeWriteRelativePath = largeWriteParams.relativePath
expect(largeWriteRelativePath).toMatch(/^uploads\/assets\/\.large\.bin\.orca-upload-/)
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(8, {
selector: 'env-1',
method: 'files.writeBase64Chunk',
params: {
worktree: 'id:wt-1',
relativePath: largeWriteRelativePath,
contentBase64: firstChunk,
append: false,
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
expectedSshConnectionGeneration: undefined
},
timeoutMs: 30_000,
expectedEnvironmentPairingRevision: 17,
expectedEnvironmentRuntimeId: 'remote-runtime'
})
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(9, {
selector: 'env-1',
method: 'files.writeBase64Chunk',
params: {
worktree: 'id:wt-1',
relativePath: largeWriteRelativePath,
contentBase64: secondChunk,
append: true,
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
expectedSshConnectionGeneration: undefined
},
timeoutMs: 30_000,
expectedEnvironmentPairingRevision: 17,
expectedEnvironmentRuntimeId: 'remote-runtime'
})
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(10, {
selector: 'env-1',
method: 'files.commitUpload',
params: {
worktree: 'id:wt-1',
tempRelativePath: largeWriteRelativePath,
finalRelativePath: 'uploads/assets/large.bin',
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
expectedSshConnectionGeneration: undefined
},
timeoutMs: 30_000,
expectedEnvironmentPairingRevision: 17,
expectedEnvironmentRuntimeId: 'remote-runtime'
})
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(11, {
selector: 'env-1',
method: 'files.delete',
params: {
worktree: 'id:wt-1',
relativePath: largeWriteRelativePath,
relativePath: uploads[0]?.relativePath,
recursive: false,
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
@@ -319,9 +193,8 @@ describe('runtime file client', () => {
expect(fsImportExternalPaths).not.toHaveBeenCalled()
})
it('chunks large staged runtime uploads below the WebSocket frame budget', async () => {
const firstChunk = 'A'.repeat(512 * 1024)
const secondChunk = 'AA=='
it('forwards a single staged file with the identity staging measured', async () => {
const entry = stagedFile('', 40 * 1024 * 1024, 55)
fsStageExternalPathsForRuntimeUpload.mockResolvedValue({
sources: [
{
@@ -329,55 +202,16 @@ describe('runtime file client', () => {
status: 'staged',
name: 'large.bin',
kind: 'file',
entries: [
{ relativePath: '', kind: 'file', contentBase64: `${firstChunk}${secondChunk}` }
]
entries: [entry]
}
]
})
runtimeEnvironmentCall
.mockResolvedValueOnce({
id: 'stat-destination-miss',
ok: false,
error: { code: 'not_found', message: 'not found' },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'create-destination-dir',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'stat-miss',
ok: false,
error: { code: 'not_found', message: 'not found' },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'write-chunk-1',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'write-chunk-2',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'commit-upload',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'delete-temp',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce(notFoundResponse('stat-destination-miss'))
.mockResolvedValueOnce(okResponse('create-destination-dir'))
.mockResolvedValueOnce(notFoundResponse('stat-miss'))
.mockResolvedValueOnce(okResponse('commit-upload'))
.mockResolvedValueOnce(okResponse('delete-temp'))
await expect(
importExternalPathsToRuntime(
@@ -401,79 +235,20 @@ describe('runtime file client', () => {
]
})
const chunkWriteCall = runtimeEnvironmentCall.mock.calls[3]?.[0] as {
params: { relativePath: string }
}
expect(chunkWriteCall.params.relativePath).toMatch(/^uploads\/\.large\.bin\.orca-upload-/)
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(4, {
selector: 'env-1',
method: 'files.writeBase64Chunk',
params: {
worktree: 'id:wt-1',
relativePath: chunkWriteCall.params.relativePath,
contentBase64: firstChunk,
append: false,
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
expectedSshConnectionGeneration: undefined
},
timeoutMs: 30_000,
expectedEnvironmentPairingRevision: undefined,
expectedEnvironmentRuntimeId: 'remote-runtime'
})
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(5, {
selector: 'env-1',
method: 'files.writeBase64Chunk',
params: {
worktree: 'id:wt-1',
relativePath: chunkWriteCall.params.relativePath,
contentBase64: secondChunk,
append: true,
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
expectedSshConnectionGeneration: undefined
},
timeoutMs: 30_000,
expectedEnvironmentPairingRevision: undefined,
expectedEnvironmentRuntimeId: 'remote-runtime'
})
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(6, {
selector: 'env-1',
method: 'files.commitUpload',
params: {
worktree: 'id:wt-1',
tempRelativePath: chunkWriteCall.params.relativePath,
finalRelativePath: 'uploads/large.bin',
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
expectedSshConnectionGeneration: undefined
},
timeoutMs: 30_000,
expectedEnvironmentPairingRevision: undefined,
expectedEnvironmentRuntimeId: 'remote-runtime'
})
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(7, {
selector: 'env-1',
method: 'files.delete',
params: {
worktree: 'id:wt-1',
relativePath: chunkWriteCall.params.relativePath,
recursive: false,
expectedExecutionHostId: 'local',
expectedSshTargetId: undefined,
expectedSshConnectionGeneration: undefined
},
timeoutMs: 15_000,
expectedEnvironmentPairingRevision: undefined,
expectedEnvironmentRuntimeId: 'remote-runtime'
})
const upload = uploadRequests()[0]
expect(upload?.relativePath).toMatch(/^uploads\/\.large\.bin\.orca-upload-/)
expect(upload?.sourceRootPath).toBe('/Users/me/large.bin')
expect(upload?.entryRelativePath).toBe('')
expect(upload?.expected).toEqual(identityOf(entry))
expect(runtimeEnvironmentCall).not.toHaveBeenCalledWith(
expect.objectContaining({ method: 'files.writeBase64' })
)
expect(runtimeEnvironmentCall).not.toHaveBeenCalledWith(
expect.objectContaining({ method: 'files.writeBase64Chunk' })
)
})
it('stops a chunked upload when its owner generation changes between writes', async () => {
const firstChunk = 'A'.repeat(512 * 1024)
it('does not commit an upload when the owner generation changes while it streams', async () => {
fsStageExternalPathsForRuntimeUpload.mockResolvedValue({
sources: [
{
@@ -481,7 +256,7 @@ describe('runtime file client', () => {
status: 'staged',
name: 'large.bin',
kind: 'file',
entries: [{ relativePath: '', kind: 'file', contentBase64: `${firstChunk}BBBBBBBB` }]
entries: [stagedFile('', 40 * 1024 * 1024, 55)]
}
]
})
@@ -492,22 +267,12 @@ describe('runtime file client', () => {
result: { size: 0, isDirectory: true, mtime: 1 },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'stat-file-miss',
ok: false,
error: { code: 'not_found', message: 'not found' },
_meta: { runtimeId: 'remote-runtime' }
})
.mockImplementationOnce(async () => {
ownerChanged = true
return {
id: 'write-chunk-1',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
}
})
.mockResolvedValueOnce(notFoundResponse('stat-file-miss'))
let ownerChanged = false
fsUploadExternalFileToRuntime.mockImplementation(async () => {
ownerChanged = true
return { byteLength: 40 * 1024 * 1024 }
})
const assertCurrent = vi.fn(() => {
if (ownerChanged) {
throw new Error('runtime owner generation changed')
@@ -531,20 +296,14 @@ describe('runtime file client', () => {
expect(runtimeEnvironmentCall.mock.calls.map((call) => call[0].method)).toEqual([
'files.stat',
'files.stat',
'files.writeBase64Chunk'
'files.stat'
])
expect(runtimeEnvironmentCall).not.toHaveBeenCalledWith(
expect.objectContaining({ method: 'files.commitUpload' })
)
expect(runtimeEnvironmentCall).not.toHaveBeenCalledWith(
expect.objectContaining({ method: 'files.delete' })
)
})
it('cleans up staged runtime upload temp files when a later chunk fails', async () => {
const firstChunk = 'A'.repeat(512 * 1024)
const secondChunk = 'BBBBBBBB'
it('cleans up the staged temp path when the streamed upload fails', async () => {
fsStageExternalPathsForRuntimeUpload.mockResolvedValue({
sources: [
{
@@ -552,49 +311,19 @@ describe('runtime file client', () => {
status: 'staged',
name: 'large.bin',
kind: 'file',
entries: [
{ relativePath: '', kind: 'file', contentBase64: `${firstChunk}${secondChunk}` }
]
entries: [stagedFile('', 40 * 1024 * 1024, 55)]
}
]
})
runtimeEnvironmentCall
.mockResolvedValueOnce({
id: 'stat-destination-miss',
ok: false,
error: { code: 'not_found', message: 'not found' },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'create-destination-dir',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'stat-miss',
ok: false,
error: { code: 'not_found', message: 'not found' },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'write-chunk-1',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'write-chunk-2',
ok: false,
error: { code: 'write_failed', message: 'disk full' },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'delete-temp',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce(notFoundResponse('stat-destination-miss'))
.mockResolvedValueOnce(okResponse('create-destination-dir'))
.mockResolvedValueOnce(notFoundResponse('stat-miss'))
.mockResolvedValueOnce(okResponse('delete-temp'))
// Electron wraps a main-process throw; the reason must not leak that.
fsUploadExternalFileToRuntime.mockRejectedValue(
new Error("Error invoking remote method 'fs:uploadExternalFileToRuntime': Error: disk full")
)
await expect(
importExternalPathsToRuntime(
@@ -610,13 +339,7 @@ describe('runtime file client', () => {
results: [{ status: 'failed', reason: 'disk full' }]
})
const chunkCall = runtimeEnvironmentCall.mock.calls[3]?.[0] as
| { params: { relativePath: string } }
| undefined
if (!chunkCall) {
throw new Error('missing first chunk call')
}
const tempRelativePath = chunkCall.params.relativePath
const tempRelativePath = uploadRequests()[0]?.relativePath
expect(runtimeEnvironmentCall).not.toHaveBeenCalledWith(
expect.objectContaining({ method: 'files.commitUpload' })
)
@@ -645,10 +368,7 @@ describe('runtime file client', () => {
status: 'staged',
name: 'assets',
kind: 'directory',
entries: [
{ relativePath: '', kind: 'directory' },
{ relativePath: 'logo.png', kind: 'file', contentBase64: 'cG5n' }
]
entries: [{ relativePath: '', kind: 'directory' }, stagedFile('logo.png', 3, 101)]
}
]
})
@@ -659,36 +379,11 @@ describe('runtime file client', () => {
result: { size: 0, isDirectory: true, mtime: 1 },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'stat-import-root-miss',
ok: false,
error: { code: 'not_found', message: 'not found' },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'create-import-root',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'write-file',
ok: false,
error: { code: 'write_failed', message: 'disk full' },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'delete-temp',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce({
id: 'delete-import-root',
ok: true,
result: { ok: true },
_meta: { runtimeId: 'remote-runtime' }
})
.mockResolvedValueOnce(notFoundResponse('stat-import-root-miss'))
.mockResolvedValueOnce(okResponse('create-import-root'))
.mockResolvedValueOnce(okResponse('delete-temp'))
.mockResolvedValueOnce(okResponse('delete-import-root'))
fsUploadExternalFileToRuntime.mockRejectedValue(new Error('disk full'))
await expect(
importExternalPathsToRuntime(
@@ -704,13 +399,7 @@ describe('runtime file client', () => {
results: [{ status: 'failed', reason: 'disk full' }]
})
const writeCall = runtimeEnvironmentCall.mock.calls[3]?.[0] as
| { params: { relativePath: string } }
| undefined
if (!writeCall) {
throw new Error('missing failed file write call')
}
expect(writeCall.params.relativePath).toMatch(/^uploads\/assets\/\.logo\.png\.orca-upload-/)
expect(uploadRequests()[0]?.relativePath).toMatch(/^uploads\/assets\/\.logo\.png\.orca-upload-/)
expect(runtimeEnvironmentCall).toHaveBeenLastCalledWith({
selector: 'env-1',
method: 'files.delete',
@@ -763,6 +452,7 @@ describe('runtime file client', () => {
expectedSshConnectionGeneration: 5
})
expect(fsStageExternalPathsForRuntimeUpload).not.toHaveBeenCalled()
expect(fsUploadExternalFileToRuntime).not.toHaveBeenCalled()
expect(runtimeEnvironmentCall).not.toHaveBeenCalled()
})
})
@@ -54,6 +54,7 @@ export const fsFinishDownloadedFile: PreloadStub = vi.fn()
export const fsCancelDownloadedFile: PreloadStub = vi.fn()
export const fsImportExternalPaths: PreloadStub = vi.fn()
export const fsStageExternalPathsForRuntimeUpload: PreloadStub = vi.fn()
export const fsUploadExternalFileToRuntime: PreloadStub = vi.fn()
export const runtimeEnvironmentCall: RuntimeRpcStub = vi.fn()
export const runtimeEnvironmentTransportCall: RuntimeRpcStub = vi.fn()
export const runtimeEnvironmentSubscribe: RuntimeSubscribeStub = vi.fn()
@@ -88,6 +89,8 @@ export function installRuntimeFileClientEnvironment(): void {
fsCancelDownloadedFile.mockReset()
fsImportExternalPaths.mockReset()
fsStageExternalPathsForRuntimeUpload.mockReset()
fsUploadExternalFileToRuntime.mockReset()
fsUploadExternalFileToRuntime.mockResolvedValue({ byteLength: 0 })
runtimeEnvironmentCall.mockReset()
runtimeEnvironmentTransportCall.mockReset()
runtimeEnvironmentSubscribe.mockReset()
@@ -131,7 +134,8 @@ export function installRuntimeFileClientEnvironment(): void {
finishDownloadedFile: fsFinishDownloadedFile,
cancelDownloadedFile: fsCancelDownloadedFile,
importExternalPaths: fsImportExternalPaths,
stageExternalPathsForRuntimeUpload: fsStageExternalPathsForRuntimeUpload
stageExternalPathsForRuntimeUpload: fsStageExternalPathsForRuntimeUpload,
uploadExternalFileToRuntime: fsUploadExternalFileToRuntime
},
runtime: { call: runtimeCall },
runtimeEnvironments: {
@@ -21,25 +21,6 @@ import {
import { getActiveRuntimeTarget } from './runtime-rpc-client'
import { toRuntimeWorktreeSelector } from './runtime-worktree-selector'
type StagedRuntimeImportSource =
| {
sourcePath: string
status: 'staged'
name: string
kind: 'file' | 'directory'
entries: StagedRuntimeImportEntry[]
}
| {
sourcePath: string
status: 'skipped'
reason: 'missing' | 'symlink' | 'permission-denied' | 'unsupported'
}
| { sourcePath: string; status: 'failed'; reason: string }
type StagedRuntimeImportEntry =
| { relativePath: string; kind: 'directory' }
| { relativePath: string; kind: 'file'; contentBase64: string }
type RuntimeImportResult =
| {
sourcePath: string
@@ -113,7 +94,7 @@ export async function importExternalPathsToRuntime(
await ensureRuntimeDirectory(context, destinationDir, importSession)
for (const source of staged.sources as StagedRuntimeImportSource[]) {
for (const source of staged.sources) {
if (source.status !== 'staged') {
results.push(source)
continue
@@ -150,7 +131,16 @@ export async function importExternalPathsToRuntime(
importSession,
context.worktreeId,
entryRelativePath,
entry.contentBase64,
{
sourceRootPath: source.sourcePath,
entryRelativePath: entry.relativePath,
expected: {
byteLength: entry.byteLength,
inode: entry.inode,
deviceId: entry.deviceId,
modifiedAtMs: entry.modifiedAtMs
}
},
context.expectedSshConnectionGeneration,
context.expectedSshTargetId,
context.expectedExecutionHostId ??
@@ -32,6 +32,7 @@ type RuntimeCallArgs = {
const runtimeEnvironmentCall = vi.fn<(args: RuntimeCallArgs) => unknown>()
const stageExternalPathsForRuntimeUpload = vi.fn()
const uploadExternalFileToRuntime = vi.fn<(args: Record<string, unknown>) => unknown>()
const importExternalPaths = vi.fn()
const nestedSshContext = {
@@ -99,7 +100,8 @@ function repairedRuntimeResponse(method: string) {
}
}
function mockStagedFile(sourcePath: string, name: string, contentBase64: string): void {
/** Staging now hands over identity, not a body; the streamer in main reads the bytes. */
function mockStagedFile(sourcePath: string, name: string, byteLength: number): void {
stageExternalPathsForRuntimeUpload.mockResolvedValue({
sources: [
{
@@ -107,12 +109,28 @@ function mockStagedFile(sourcePath: string, name: string, contentBase64: string)
status: 'staged',
name,
kind: 'file',
entries: [{ relativePath: '', kind: 'file', contentBase64 }]
entries: [
{
relativePath: '',
kind: 'file',
byteLength,
inode: 91,
deviceId: 66,
modifiedAtMs: 1_700_000_000_000
}
]
}
]
})
}
function expectUploadsBoundToCapturedRevision(): void {
for (const [args] of uploadExternalFileToRuntime.mock.calls) {
expect(args.expectedEnvironmentPairingRevision).toBe(CAPTURED_REVISION)
expect(args.expectedEnvironmentRuntimeId).toBe('hub-runtime')
}
}
function expectEveryRuntimeCallBoundToCapturedRevision(ownership: {
expectedExecutionHostId: string
expectedSshTargetId?: string
@@ -146,12 +164,15 @@ beforeEach(() => {
markRuntimeEnvironmentCompatible(ENVIRONMENT_ID)
runtimeEnvironmentCall.mockReset()
stageExternalPathsForRuntimeUpload.mockReset()
uploadExternalFileToRuntime.mockReset()
uploadExternalFileToRuntime.mockResolvedValue({ byteLength: 0 })
importExternalPaths.mockReset()
vi.stubGlobal('window', {
api: {
fs: {
importExternalPaths,
stageExternalPathsForRuntimeUpload
stageExternalPathsForRuntimeUpload,
uploadExternalFileToRuntime
},
runtimeEnvironments: {
call: runtimeEnvironmentCall
@@ -183,7 +204,7 @@ describe('runtime file import pairing revision', () => {
})
it('stops when the HUB runtime changes without a pairing change', async () => {
mockStagedFile('/client/screenshot.png', 'screenshot.png', `${'A'.repeat(512 * 1024)}BBBBBBBB`)
mockStagedFile('/client/screenshot.png', 'screenshot.png', 40 * 1024 * 1024)
runtimeEnvironmentCall.mockImplementation(async (args: RuntimeCallArgs) => {
if (args.method === 'status.get') {
return runtimeStatusResponse()
@@ -191,14 +212,15 @@ describe('runtime file import pairing revision', () => {
if (args.method === 'files.stat') {
return missingRuntimePathResponse()
}
if (args.method === 'files.writeBase64Chunk') {
setRuntimeEnvironmentConnectionGenerationForTests(
ENVIRONMENT_ID,
REPLACEMENT_CONNECTION_GENERATION
)
}
return successfulRuntimeResponse(args.method)
})
uploadExternalFileToRuntime.mockImplementation(async () => {
setRuntimeEnvironmentConnectionGenerationForTests(
ENVIRONMENT_ID,
REPLACEMENT_CONNECTION_GENERATION
)
return { byteLength: 40 * 1024 * 1024 }
})
await expect(
importExternalPathsToRuntime(nestedSshContext, ['/client/screenshot.png'], '/ssh/repo')
@@ -208,9 +230,9 @@ describe('runtime file import pairing revision', () => {
expect(runtimeEnvironmentCall.mock.calls.map(([args]) => args.method)).toEqual([
'status.get',
'files.stat',
'files.writeBase64Chunk'
'files.stat'
])
expectUploadsBoundToCapturedRevision()
expect(runtimeEnvironmentCall).not.toHaveBeenCalledWith(
expect.objectContaining({ method: 'files.commitUpload' })
)
@@ -236,8 +258,8 @@ describe('runtime file import pairing revision', () => {
expect(importExternalPaths).not.toHaveBeenCalled()
})
it('stops a rich-markdown upload between chunks without contacting the replacement HUB', async () => {
mockStagedFile('/client/screenshot.png', 'screenshot.png', `${'A'.repeat(512 * 1024)}BBBBBBBB`)
it('never commits a streamed upload against a replacement HUB re-paired mid-stream', async () => {
mockStagedFile('/client/screenshot.png', 'screenshot.png', 40 * 1024 * 1024)
runtimeEnvironmentCall.mockImplementation(async (args: RuntimeCallArgs) => {
if (args.method === 'status.get') {
return runtimeStatusResponse()
@@ -245,11 +267,12 @@ describe('runtime file import pairing revision', () => {
if (args.method === 'files.stat') {
return missingRuntimePathResponse()
}
if (args.method === 'files.writeBase64Chunk') {
setEnvironmentRevision(REPLACEMENT_REVISION)
}
return successfulRuntimeResponse(args.method)
})
uploadExternalFileToRuntime.mockImplementation(async () => {
setEnvironmentRevision(REPLACEMENT_REVISION)
return { byteLength: 40 * 1024 * 1024 }
})
await expect(
importExternalPathsToRuntime(nestedSshContext, ['/client/screenshot.png'], '/ssh/repo')
@@ -259,20 +282,9 @@ describe('runtime file import pairing revision', () => {
expect(runtimeEnvironmentCall.mock.calls.map(([args]) => args.method)).toEqual([
'status.get',
'files.stat',
'files.writeBase64Chunk'
'files.stat'
])
expect(runtimeEnvironmentCall).toHaveBeenNthCalledWith(
3,
expect.objectContaining({
method: 'files.writeBase64Chunk',
expectedEnvironmentPairingRevision: CAPTURED_REVISION,
params: expect.objectContaining({
contentBase64: 'A'.repeat(512 * 1024),
append: false
})
})
)
expectUploadsBoundToCapturedRevision()
expectEveryRuntimeCallBoundToCapturedRevision(nestedSshContext)
expect(runtimeEnvironmentCall).not.toHaveBeenCalledWith(
expect.objectContaining({ method: 'files.commitUpload' })
@@ -283,7 +295,7 @@ describe('runtime file import pairing revision', () => {
})
it('keeps a HUB-local composer commit on its entry revision when re-paired during commit', async () => {
mockStagedFile('/client/note.txt', 'note.txt', 'bm90ZQ==')
mockStagedFile('/client/note.txt', 'note.txt', 4)
runtimeEnvironmentCall.mockImplementation(async (args: RuntimeCallArgs) => {
if (args.expectedEnvironmentPairingRevision !== CAPTURED_REVISION) {
throw new Error('replacement HUB received an import RPC')
@@ -310,14 +322,13 @@ describe('runtime file import pairing revision', () => {
expect(runtimeEnvironmentCall.mock.calls.map(([args]) => args.method)).toEqual([
'status.get',
'files.stat',
'files.writeBase64',
'files.commitUpload'
])
expectEveryRuntimeCallBoundToCapturedRevision(hubLocalContext)
})
it('does not clean up against a replacement HUB after commit', async () => {
mockStagedFile('/client/drop.txt', 'drop.txt', 'ZHJvcA==')
mockStagedFile('/client/drop.txt', 'drop.txt', 4)
runtimeEnvironmentCall.mockImplementation(async (args: RuntimeCallArgs) => {
if (args.method === 'status.get') {
return runtimeStatusResponse()
@@ -340,7 +351,6 @@ describe('runtime file import pairing revision', () => {
expect(runtimeEnvironmentCall.mock.calls.map(([args]) => args.method)).toEqual([
'status.get',
'files.stat',
'files.writeBase64',
'files.commitUpload'
])
expectEveryRuntimeCallBoundToCapturedRevision(nestedSshContext)
@@ -356,7 +366,14 @@ describe('runtime file import pairing revision', () => {
kind: 'directory',
entries: [
{ relativePath: '', kind: 'directory' },
{ relativePath: 'broken.txt', kind: 'file', contentBase64: 'YnJva2Vu' }
{
relativePath: 'broken.txt',
kind: 'file',
byteLength: 6,
inode: 92,
deviceId: 66,
modifiedAtMs: 1_700_000_000_000
}
]
}
]
@@ -368,16 +385,9 @@ describe('runtime file import pairing revision', () => {
if (args.method === 'files.stat') {
return missingRuntimePathResponse()
}
if (args.method === 'files.writeBase64') {
return {
id: args.method,
ok: false,
error: { code: 'write_failed', message: 'disk full' },
_meta: { runtimeId: 'hub-runtime' }
}
}
return successfulRuntimeResponse(args.method)
})
uploadExternalFileToRuntime.mockRejectedValue(new Error('disk full'))
await expect(
importExternalPathsToRuntime(nestedSshContext, ['/client/assets'], '/ssh/repo')
@@ -387,7 +397,6 @@ describe('runtime file import pairing revision', () => {
'status.get',
'files.stat',
'files.createDirNoClobber',
'files.writeBase64',
'files.delete',
'files.delete'
])
@@ -1,4 +1,6 @@
import { extractIpcErrorMessage } from '@/lib/ipc-error'
import { joinPath, normalizeRelativePath } from '@/lib/path'
import type { StagedRuntimeUploadFileIdentity } from '../../../shared/runtime-upload-staging-contract'
import type { RuntimeFileOperationArgs } from './runtime-file-client-types'
import {
callRuntimeFileImportMutation,
@@ -12,28 +14,48 @@ import {
import { runtimePathExists } from './runtime-file-metadata-client'
import { toRuntimeWorktreeSelector } from './runtime-worktree-selector'
const REMOTE_UPLOAD_BASE64_CHUNK_CHARS = 512 * 1024
/** Locates a staged file on the client so main can stream it without the renderer reading it. */
export type RuntimeUploadSource = {
sourceRootPath: string
entryRelativePath: string
/** What staging observed; main refuses the upload if the source no longer matches. */
expected: StagedRuntimeUploadFileIdentity
}
/** Stream one staged file to a temp path, then commit it; the temp path is always cleaned up. */
export async function uploadRuntimeFileWithoutClobber(
session: RuntimeFileImportSession,
worktreeId: string,
relativePath: string,
contentBase64: string,
source: RuntimeUploadSource,
expectedSshConnectionGeneration?: number,
expectedSshTargetId?: string,
expectedExecutionHostId?: 'local' | `ssh:${string}`
): Promise<void> {
const tempRelativePath = makeRuntimeUploadTempPath(relativePath)
try {
await writeRuntimeBase64File(
session,
worktreeId,
tempRelativePath,
contentBase64,
expectedSshConnectionGeneration,
expectedSshTargetId,
expectedExecutionHostId
)
session.assertCurrent()
// Why: main owns the file handle and the runtime socket, so it streams the
// body in slices; the renderer never holds the whole file.
try {
await window.api.fs.uploadExternalFileToRuntime({
environmentId: session.target.environmentId,
sourceRootPath: source.sourceRootPath,
entryRelativePath: source.entryRelativePath,
expected: source.expected,
worktree: toRuntimeWorktreeSelector(worktreeId),
relativePath: tempRelativePath,
expectedSshTargetId,
expectedSshConnectionGeneration,
expectedExecutionHostId,
expectedEnvironmentPairingRevision: session.expectedEnvironmentPairingRevision,
expectedEnvironmentRuntimeId: session.expectedEnvironmentRuntimeId
})
} catch (error) {
// Why: this surfaces in the import result as-is, and Electron wraps a
// main-process throw in "Error invoking remote method '…'".
throw new Error(extractIpcErrorMessage(error, 'Upload failed'))
}
await callRuntimeFileImportMutation(
session,
'files.commitUpload',
@@ -64,50 +86,7 @@ export async function uploadRuntimeFileWithoutClobber(
}
}
async function writeRuntimeBase64File(
session: RuntimeFileImportSession,
worktreeId: string,
relativePath: string,
contentBase64: string,
expectedSshConnectionGeneration?: number,
expectedSshTargetId?: string,
expectedExecutionHostId?: 'local' | `ssh:${string}`
): Promise<void> {
if (contentBase64.length <= REMOTE_UPLOAD_BASE64_CHUNK_CHARS) {
await callRuntimeFileImportMutation(
session,
'files.writeBase64',
{
worktree: toRuntimeWorktreeSelector(worktreeId),
relativePath,
contentBase64,
expectedSshTargetId,
expectedSshConnectionGeneration,
expectedExecutionHostId
},
30_000
)
return
}
for (let offset = 0; offset < contentBase64.length; offset += REMOTE_UPLOAD_BASE64_CHUNK_CHARS) {
await callRuntimeFileImportMutation(
session,
'files.writeBase64Chunk',
{
worktree: toRuntimeWorktreeSelector(worktreeId),
relativePath,
contentBase64: contentBase64.slice(offset, offset + REMOTE_UPLOAD_BASE64_CHUNK_CHARS),
append: offset > 0,
expectedSshTargetId,
expectedSshConnectionGeneration,
expectedExecutionHostId
},
30_000
)
}
}
/** Hidden sibling of the destination, so a failed upload never leaves a plausible-looking file. */
function makeRuntimeUploadTempPath(relativePath: string): string {
const normalized = normalizeRelativePath(relativePath)
const slashIndex = normalized.lastIndexOf('/')
@@ -126,6 +126,11 @@ export function createFileApi(): NonNullable<Partial<PreloadApi>['fs']> {
},
importExternalPaths: async () => ({ results: [] }),
stageExternalPathsForRuntimeUpload: async () => ({ sources: [] }),
// Why: the web client has no local filesystem to stream from, so staging
// never yields a source for this to upload.
uploadExternalFileToRuntime: async () => {
throw new Error('Uploading local files is not supported in the web client')
},
resolveDroppedPathsForAgent: async () => ({ resolvedPaths: [], skipped: [], failed: [] }),
watchWorktree: () => Promise.resolve(),
unwatchWorktree: () => Promise.resolve(),
@@ -0,0 +1,50 @@
import type { SshMutationExpectation } from './ssh-types'
export type RuntimeUploadSkipReason = 'missing' | 'symlink' | 'permission-denied' | 'unsupported'
/**
* What staging observed about a file, so the uploader can refuse a source that
* was swapped between the two calls. Size alone misses a same-size replacement.
*/
export type StagedRuntimeUploadFileIdentity = {
byteLength: number
/** 0 when the filesystem does not report one; compared only when both sides have it. */
inode: number
deviceId: number
modifiedAtMs: number
}
export type StagedRuntimeUploadEntry =
| { relativePath: string; kind: 'directory' }
// Why: file bodies are streamed in slices at upload time, so staging carries
// identity the uploader re-checks against the handle it actually reads.
| ({ relativePath: string; kind: 'file' } & StagedRuntimeUploadFileIdentity)
export type StagedRuntimeUploadSource =
| {
sourcePath: string
status: 'staged'
name: string
kind: 'file' | 'directory'
entries: StagedRuntimeUploadEntry[]
}
| { sourcePath: string; status: 'skipped'; reason: RuntimeUploadSkipReason }
| { sourcePath: string; status: 'failed'; reason: string }
export type StageRuntimeUploadResult = { sources: StagedRuntimeUploadSource[] }
/** Renderer → main request to pump one staged file's bytes to the runtime. */
export type RuntimeUploadFileStreamRequest = {
environmentId: string
/** Client-local path of the dropped source (file, or root of a dropped directory). */
sourceRootPath: string
/** Path of this file within the dropped directory; empty when the source is a file. */
entryRelativePath: string
/** Identity staging recorded; a source that no longer matches is refused, not streamed. */
expected: StagedRuntimeUploadFileIdentity
worktree: string
/** Destination path on the runtime, relative to the worktree. */
relativePath: string
expectedEnvironmentPairingRevision?: number
expectedEnvironmentRuntimeId?: string
} & SshMutationExpectation