fix(skills): recover interrupted remote transfers

This commit is contained in:
Jinwoo-H
2026-08-11 10:31:09 -04:00
parent 83d26e09fd
commit 149d2dc2b5
15 changed files with 616 additions and 115 deletions
+33 -84
View File
@@ -2,27 +2,18 @@ import { randomUUID } from 'node:crypto'
import { join } from 'node:path'
import { app, BrowserWindow, ipcMain } from 'electron'
import { z } from 'zod'
import {
SKILL_INSTALL_CAPABILITY,
SKILL_INSTALL_UPDATE_REQUIRED_MESSAGE
} from '../../shared/skill-install-capability'
import {
SkillInstallDestinationSchema,
type SkillInstallRequest
} from '../../shared/skill-install-contract'
import { SKILL_INSTALL_UPDATE_REQUIRED_MESSAGE } from '../../shared/skill-install-capability'
import { SkillInstallDestinationSchema } from '../../shared/skill-install-contract'
import type { SkillInstallProgress } from '../../shared/skill-sharing-contract'
import { SkillDiscoveryTargetSchema, type SkillDiscoveryResult } from '../../shared/skills'
import type { SkillCloudDownloadGrant } from '../../shared/skill-cloud-contract'
import type { OrcaRuntimeService } from '../runtime/orca-runtime'
import { installSkillOnRemoteRuntime } from '../skills/skill-remote-install-service'
import { installSkillCloudGrant } from '../skills/skill-cloud-grant-installation'
import { SkillRemoteInstallCancellation } from '../skills/skill-remote-install-cancellation'
import { classifySkillCloudInstallTarget } from '../skills/skill-cloud-install-target'
import { SkillSharePreparationService } from '../skills/skill-share-preparation-service'
import { skillInstallFailureFromError } from '../skills/skill-install-operation-error'
import { supportsSkillRuntimeInstall } from '../skills/skill-runtime-capability'
import {
callRuntimeEnvironment,
getRuntimeEnvironmentStatus
} from './runtime-environment-transport-routing'
import { callRuntimeEnvironment } from './runtime-environment-transport-routing'
import { registerSkillInstallManagementIpcHandlers } from './skill-install-management-ipc-handlers'
const environmentIdSchema = z.string().min(1).max(128)
@@ -84,70 +75,6 @@ const packageVersionSchema = z
})
.strict()
async function installGrant(
runtime: OrcaRuntimeService,
grant: SkillCloudDownloadGrant,
input: z.infer<typeof shareInstallSchema> | z.infer<typeof packageVersionInstallSchema>
) {
const request: SkillInstallRequest = {
operationId: input.operationId ?? randomUUID(),
package: {
packageId: grant.version.packageId,
versionId: grant.version.versionId,
packageDigest: grant.version.packageDigest,
archiveSha256: grant.version.archiveSha256,
compressedBytes: grant.version.compressedBytes
},
ingress: {
kind: 'download-grant',
url: grant.grant.url,
expiresAt: grant.grant.expiresAt
},
destination: input.destination,
conflictResolution: input.conflictResolution
}
try {
if (!input.environmentId) {
return { status: 'ok' as const, value: await runtime.installSharedSkillRequest(request) }
}
const userDataPath = app.getPath('userData')
const status = await getRuntimeEnvironmentStatus(userDataPath, input.environmentId, 15_000)
if (
status.ok !== true ||
status.result.capabilities?.includes(SKILL_INSTALL_CAPABILITY) !== true
) {
return { status: 'unsupported' as const, message: SKILL_INSTALL_UPDATE_REQUIRED_MESSAGE }
}
return {
status: 'ok' as const,
value: await installSkillOnRemoteRuntime({
userDataPath,
environmentId: input.environmentId,
request,
capabilities: status.result.capabilities ?? [],
requireHttps: app.isPackaged
})
}
} catch (error) {
const failure = skillInstallFailureFromError(error)
if (!failure) {
throw error
}
return {
status: 'ok' as const,
value: {
operationId: request.operationId,
status: failure.category === 'cancelled' ? ('cancelled' as const) : ('failed' as const),
name: grant.version.name,
packageDigest: request.package.packageDigest,
placements: [],
errorCategory: failure.code,
failure
}
}
}
}
function registerSharingHandlers(
runtime: OrcaRuntimeService,
discover: (target?: z.infer<typeof SkillDiscoveryTargetSchema>) => Promise<SkillDiscoveryResult>
@@ -187,6 +114,25 @@ function registerSharingHandlers(
}
function registerCloudInstallHandlers(runtime: OrcaRuntimeService): void {
const remoteInstallCancellation = new SkillRemoteInstallCancellation()
const installAuthorizedGrant = async (
grant: SkillCloudDownloadGrant,
input: z.infer<typeof shareInstallSchema> | z.infer<typeof packageVersionInstallSchema>
) => {
if (!input.environmentId || input.environmentId.startsWith('ssh:')) {
return installSkillCloudGrant(runtime, grant, {
...input,
operationId: input.operationId ?? randomUUID()
})
}
const operationId = input.operationId ?? randomUUID()
const signal = remoteInstallCancellation.begin(operationId)
try {
return await installSkillCloudGrant(runtime, grant, { ...input, operationId }, signal)
} finally {
remoteInstallCancellation.finish(operationId, signal)
}
}
ipcMain.handle('skills:resolveShare', (_event, shareId: unknown) =>
runtime.resolveSkillShare(z.string().min(1).max(128).parse(shareId), {})
)
@@ -219,7 +165,7 @@ function registerCloudInstallHandlers(runtime: OrcaRuntimeService): void {
if (grant.status === 'ok') {
sendProgress(event, { operationId: input.operationId, phase: 'installing' })
}
return grant.status === 'ok' ? installGrant(runtime, grant.value, input) : grant
return grant.status === 'ok' ? installAuthorizedGrant(grant.value, input) : grant
})
ipcMain.handle('skills:installPackageVersion', async (event, value: unknown) => {
const parsed = packageVersionInstallSchema.parse(value)
@@ -240,7 +186,7 @@ function registerCloudInstallHandlers(runtime: OrcaRuntimeService): void {
if (grant.status === 'ok') {
sendProgress(event, { operationId: input.operationId, phase: 'installing' })
}
return grant.status === 'ok' ? installGrant(runtime, grant.value, input) : grant
return grant.status === 'ok' ? installAuthorizedGrant(grant.value, input) : grant
})
ipcMain.handle('skills:cancelInstall', async (_event, value: unknown) => {
const input = z
@@ -253,16 +199,19 @@ function registerCloudInstallHandlers(runtime: OrcaRuntimeService): void {
if (!input.environmentId || input.environmentId.startsWith('ssh:')) {
return { cancelled: runtime.cancelSharedSkillInstall(input.operationId) }
}
const transferCancelled = remoteInstallCancellation.cancel(input.operationId)
const response = await callRuntimeEnvironment(
app.getPath('userData'),
input.environmentId,
'skills.cancelInstall',
{ operationId: input.operationId },
15_000
)
return response.ok === true && response.result && typeof response.result === 'object'
? { cancelled: (response.result as { cancelled?: unknown }).cancelled === true }
: { cancelled: false }
).catch(() => null)
const installCancelled =
response?.ok === true && response.result && typeof response.result === 'object'
? (response.result as { cancelled?: unknown }).cancelled === true
: false
return { cancelled: transferCancelled || installCancelled }
})
ipcMain.handle('skills:getPackage', (_event, packageId: unknown) =>
runtime.getSkillPackage(z.string().min(1).max(128).parse(packageId), {})
@@ -30,6 +30,36 @@ describe('transferSkillPackageToRuntime', () => {
await rm(root, { recursive: true, force: true })
})
it('cleans the local package when connection loss hides the begin result', async () => {
const archivePath = join(root, 'package.tar.gz')
await writeFile(archivePath, Buffer.from('package-data'))
const cleanup = vi.fn(async () => undefined)
mocks.downloadSkillPackageGrant.mockResolvedValue({ archivePath, cleanup })
mocks.callRuntimeEnvironment.mockRejectedValue(new Error('connection dropped'))
await expect(
transferSkillPackageToRuntime({
userDataPath: root,
environmentId: 'environment-1',
package: {
packageId: 'package-1',
versionId: 'version-1',
packageDigest: 'a'.repeat(64),
archiveSha256: 'b'.repeat(64),
compressedBytes: 12
},
grant: {
url: 'https://storage.googleapis.com/bucket/package.tar.gz',
expiresAt: '2099-01-01T00:00:00.000Z'
},
requireHttps: true
})
).rejects.toThrow('connection dropped')
expect(cleanup).toHaveBeenCalledOnce()
expect(mocks.callRuntimeEnvironment).toHaveBeenCalledOnce()
})
it('rejects an invalid acknowledgement and cancels the remote session', async () => {
const archivePath = join(root, 'package.tar.gz')
await writeFile(archivePath, Buffer.from('package-data'))
@@ -151,6 +181,194 @@ describe('transferSkillPackageToRuntime', () => {
).rejects.toThrow('skill-transfer-remote-skill-upload-commit-failed')
expect(downloadCleanup).toHaveBeenCalledOnce()
expect(
mocks.callRuntimeEnvironment.mock.calls.filter((call) => call[2] === 'skills.commitUpload')
).toHaveLength(1)
expect(mocks.callRuntimeEnvironment).toHaveBeenCalledWith(
root,
'environment-1',
'skills.cancelUpload',
{ uploadId: 'upload-1' },
5 * 60_000
)
})
it('cancels the session after bounded chunk retry exhaustion', async () => {
const archivePath = join(root, 'package.tar.gz')
await writeFile(archivePath, Buffer.from('package-data'))
const cleanup = vi.fn(async () => undefined)
mocks.downloadSkillPackageGrant.mockResolvedValue({ archivePath, cleanup })
mocks.callRuntimeEnvironment.mockImplementation(
async (_userData: string, _environment: string, method: string) => {
if (method === 'skills.beginUpload') {
return {
id: 'rpc-begin',
ok: true,
result: { uploadId: 'upload-1', chunkBytes: 256 * 1024 },
_meta: { runtimeId: 'runtime-1' }
}
}
if (method === 'skills.uploadChunk') {
throw new Error('connection dropped')
}
return {
id: 'rpc-cancel',
ok: true,
result: undefined,
_meta: { runtimeId: 'runtime-1' }
}
}
)
await expect(
transferSkillPackageToRuntime({
userDataPath: root,
environmentId: 'environment-1',
package: {
packageId: 'package-1',
versionId: 'version-1',
packageDigest: 'a'.repeat(64),
archiveSha256: 'b'.repeat(64),
compressedBytes: 12
},
grant: {
url: 'https://storage.googleapis.com/bucket/package.tar.gz',
expiresAt: '2099-01-01T00:00:00.000Z'
},
requireHttps: true
})
).rejects.toThrow('connection dropped')
expect(
mocks.callRuntimeEnvironment.mock.calls.filter((call) => call[2] === 'skills.uploadChunk')
).toHaveLength(3)
expect(
mocks.callRuntimeEnvironment.mock.calls.filter((call) => call[2] === 'skills.cancelUpload')
).toHaveLength(1)
expect(cleanup).toHaveBeenCalledOnce()
})
it('retries the identical chunk and idempotent commit after lost responses', async () => {
const archivePath = join(root, 'package.tar.gz')
await writeFile(archivePath, Buffer.from('package-data'))
const cleanup = vi.fn(async () => undefined)
mocks.downloadSkillPackageGrant.mockResolvedValue({ archivePath, cleanup })
let chunkAttempts = 0
let commitAttempts = 0
mocks.callRuntimeEnvironment.mockImplementation(
async (_userData: string, _environment: string, method: string, params: unknown) => {
if (method === 'skills.beginUpload') {
return {
id: 'rpc-begin',
ok: true,
result: { uploadId: 'upload-1', chunkBytes: 256 * 1024 },
_meta: { runtimeId: 'runtime-1' }
}
}
if (method === 'skills.uploadChunk') {
chunkAttempts += 1
if (chunkAttempts === 1) {
throw new Error('connection dropped after receiver write')
}
return {
id: 'rpc-chunk',
ok: true,
result: { acknowledgedOffset: 12 },
_meta: { runtimeId: 'runtime-1' }
}
}
if (method === 'skills.commitUpload') {
commitAttempts += 1
if (commitAttempts === 1) {
throw new Error('connection dropped after receiver commit')
}
return {
id: 'rpc-commit',
ok: true,
result: { uploadId: 'upload-1' },
_meta: { runtimeId: 'runtime-1' }
}
}
expect(method).toBe('skills.cancelUpload')
expect(params).toEqual({ uploadId: 'upload-1' })
return {
id: 'rpc-cancel',
ok: true,
result: undefined,
_meta: { runtimeId: 'runtime-1' }
}
}
)
const transferred = await transferSkillPackageToRuntime({
userDataPath: root,
environmentId: 'environment-1',
package: {
packageId: 'package-1',
versionId: 'version-1',
packageDigest: 'a'.repeat(64),
archiveSha256: 'b'.repeat(64),
compressedBytes: 12
},
grant: {
url: 'https://storage.googleapis.com/bucket/package.tar.gz',
expiresAt: '2099-01-01T00:00:00.000Z'
},
requireHttps: true
})
await transferred.cleanup()
expect(chunkAttempts).toBe(2)
expect(commitAttempts).toBe(2)
expect(cleanup).toHaveBeenCalledOnce()
})
it('cancels the begun session when cancellation arrives before chunking', async () => {
const archivePath = join(root, 'package.tar.gz')
await writeFile(archivePath, Buffer.from('package-data'))
const cleanup = vi.fn(async () => undefined)
const controller = new AbortController()
mocks.downloadSkillPackageGrant.mockResolvedValue({ archivePath, cleanup })
mocks.callRuntimeEnvironment.mockImplementation(
async (_userData: string, _environment: string, method: string) => {
if (method === 'skills.beginUpload') {
controller.abort()
return {
id: 'rpc-begin',
ok: true,
result: { uploadId: 'upload-1', chunkBytes: 256 * 1024 },
_meta: { runtimeId: 'runtime-1' }
}
}
return {
id: 'rpc-cancel',
ok: true,
result: undefined,
_meta: { runtimeId: 'runtime-1' }
}
}
)
await expect(
transferSkillPackageToRuntime({
userDataPath: root,
environmentId: 'environment-1',
package: {
packageId: 'package-1',
versionId: 'version-1',
packageDigest: 'a'.repeat(64),
archiveSha256: 'b'.repeat(64),
compressedBytes: 12
},
grant: {
url: 'https://storage.googleapis.com/bucket/package.tar.gz',
expiresAt: '2099-01-01T00:00:00.000Z'
},
requireHttps: true,
signal: controller.signal
})
).rejects.toThrow('skill-install-cancelled')
expect(cleanup).toHaveBeenCalledOnce()
expect(mocks.callRuntimeEnvironment).toHaveBeenCalledWith(
root,
'environment-1',
@@ -5,6 +5,7 @@ import type { SkillPackageIdentity } from '../../shared/skill-install-contract'
import { SKILL_UPLOAD_CHUNK_MAX_BYTES } from '../../shared/skill-upload-session-contract'
import { callRuntimeEnvironment } from '../ipc/runtime-environment-transport-routing'
import { downloadSkillPackageGrant } from './skill-package-download'
import { retrySkillTransferRpc, throwIfSkillTransferCancelled } from './skill-transfer-rpc-retry'
function allowedOrigins(allowConfiguredOrigins: boolean): string[] {
const origins = ['https://storage.googleapis.com']
if (allowConfiguredOrigins && process.env.ORCA_SKILL_PACKAGE_DOWNLOAD_ORIGINS) {
@@ -17,6 +18,10 @@ function allowedOrigins(allowConfiguredOrigins: boolean): string[] {
return [...new Set(origins)]
}
function retryableRemoteTransferError(error: unknown): boolean {
return !(error instanceof Error && error.message.startsWith('skill-transfer-remote-'))
}
async function remoteCall<T>(
userDataPath: string,
environmentId: string,
@@ -42,6 +47,7 @@ export async function transferSkillPackageToRuntime(input: {
package: SkillPackageIdentity
grant: { url: string; expiresAt: string }
requireHttps: boolean
signal?: AbortSignal
}): Promise<{ uploadId: string; cleanup(): Promise<void> }> {
const downloaded = await downloadSkillPackageGrant({
url: input.grant.url,
@@ -50,7 +56,8 @@ export async function transferSkillPackageToRuntime(input: {
expectedCompressedBytes: input.package.compressedBytes,
temporaryRoot: join(input.userDataPath, 'skill-installs', 'client-transfers'),
allowedOrigins: allowedOrigins(!input.requireHttps),
requireHttps: input.requireHttps
requireHttps: input.requireHttps,
signal: input.signal
})
let uploadId: string | null = null
try {
@@ -61,6 +68,7 @@ export async function transferSkillPackageToRuntime(input: {
{ package: input.package }
)
uploadId = begun.uploadId
throwIfSkillTransferCancelled(input.signal)
const chunkBytes = Math.min(begun.chunkBytes, SKILL_UPLOAD_CHUNK_MAX_BYTES)
if (!Number.isInteger(chunkBytes) || chunkBytes < 1) {
throw new Error('skill-transfer-chunk-size-invalid')
@@ -74,12 +82,17 @@ export async function transferSkillPackageToRuntime(input: {
if (read.bytesRead !== bytes.length) {
throw new Error('skill-transfer-source-changed')
}
const acknowledged = await remoteCall<{ acknowledgedOffset: number }>(
input.userDataPath,
input.environmentId,
'skills.uploadChunk',
{ uploadId, offset, bytesBase64: bytes.toString('base64') }
)
const acknowledged = await retrySkillTransferRpc({
signal: input.signal,
retryable: retryableRemoteTransferError,
call: () =>
remoteCall<{ acknowledgedOffset: number }>(
input.userDataPath,
input.environmentId,
'skills.uploadChunk',
{ uploadId, offset, bytesBase64: bytes.toString('base64') }
)
})
if (acknowledged.acknowledgedOffset !== offset + bytes.length) {
throw new Error('skill-transfer-ack-invalid')
}
@@ -88,7 +101,12 @@ export async function transferSkillPackageToRuntime(input: {
} finally {
await handle.close()
}
await remoteCall(input.userDataPath, input.environmentId, 'skills.commitUpload', { uploadId })
await retrySkillTransferRpc({
signal: input.signal,
retryable: retryableRemoteTransferError,
call: () =>
remoteCall(input.userDataPath, input.environmentId, 'skills.commitUpload', { uploadId })
})
const committedId = uploadId
uploadId = null
return {
@@ -0,0 +1,74 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import type { SkillCloudDownloadGrant } from '../../shared/skill-cloud-contract'
import type { OrcaRuntimeService } from '../runtime/orca-runtime'
const mocks = vi.hoisted(() => ({
getRuntimeEnvironmentStatus: vi.fn(),
installSkillOnRemoteRuntime: vi.fn()
}))
vi.mock('electron', () => ({
app: { getPath: () => '/state', isPackaged: true }
}))
vi.mock('../ipc/runtime-environment-transport-routing', () => ({
getRuntimeEnvironmentStatus: mocks.getRuntimeEnvironmentStatus
}))
vi.mock('./skill-remote-install-service', () => ({
installSkillOnRemoteRuntime: mocks.installSkillOnRemoteRuntime
}))
import { installSkillCloudGrant } from './skill-cloud-grant-installation'
const grant = {
grant: {
url: 'https://storage.googleapis.com/bucket/package.tar.gz',
expiresAt: '2099-01-01T00:00:00.000Z'
},
version: {
packageId: 'package-1',
versionId: 'version-1',
name: 'private-skill',
packageDigest: 'a'.repeat(64),
archiveSha256: 'b'.repeat(64),
compressedBytes: 12
}
} as unknown as SkillCloudDownloadGrant
describe('installSkillCloudGrant', () => {
beforeEach(() => {
mocks.getRuntimeEnvironmentStatus.mockReset()
mocks.installSkillOnRemoteRuntime.mockReset()
})
it('carries cancellation into client-mediated remote installation', async () => {
const signal = new AbortController().signal
const result = {
operationId: 'operation-1',
status: 'installed',
name: 'private-skill',
packageDigest: 'a'.repeat(64),
placements: []
}
mocks.getRuntimeEnvironmentStatus.mockResolvedValue({
ok: true,
result: { capabilities: ['skills.install.v1', 'skills.upload.v1'] }
})
mocks.installSkillOnRemoteRuntime.mockResolvedValue(result)
await expect(
installSkillCloudGrant(
{} as OrcaRuntimeService,
grant,
{
operationId: 'operation-1',
environmentId: 'environment-1',
destination: { scope: 'global' }
},
signal
)
).resolves.toEqual({ status: 'ok', value: result })
expect(mocks.installSkillOnRemoteRuntime).toHaveBeenCalledWith(
expect.objectContaining({ signal })
)
})
})
@@ -0,0 +1,87 @@
import { app } from 'electron'
import {
SKILL_INSTALL_CAPABILITY,
SKILL_INSTALL_UPDATE_REQUIRED_MESSAGE
} from '../../shared/skill-install-capability'
import type {
SkillInstallDestination,
SkillInstallRequest
} from '../../shared/skill-install-contract'
import type { SkillCloudDownloadGrant } from '../../shared/skill-cloud-contract'
import { getRuntimeEnvironmentStatus } from '../ipc/runtime-environment-transport-routing'
import type { OrcaRuntimeService } from '../runtime/orca-runtime'
import { skillInstallFailureFromError } from './skill-install-operation-error'
import { installSkillOnRemoteRuntime } from './skill-remote-install-service'
export type SkillCloudGrantInstallInput = {
operationId: string
environmentId?: string
destination: SkillInstallDestination
conflictResolution?: 'replace-unmodified' | 'replace-and-discard-local' | 'cancel'
}
export async function installSkillCloudGrant(
runtime: OrcaRuntimeService,
grant: SkillCloudDownloadGrant,
input: SkillCloudGrantInstallInput,
signal?: AbortSignal
) {
const request: SkillInstallRequest = {
operationId: input.operationId,
package: {
packageId: grant.version.packageId,
versionId: grant.version.versionId,
packageDigest: grant.version.packageDigest,
archiveSha256: grant.version.archiveSha256,
compressedBytes: grant.version.compressedBytes
},
ingress: {
kind: 'download-grant',
url: grant.grant.url,
expiresAt: grant.grant.expiresAt
},
destination: input.destination,
conflictResolution: input.conflictResolution
}
try {
if (!input.environmentId) {
return { status: 'ok' as const, value: await runtime.installSharedSkillRequest(request) }
}
const userDataPath = app.getPath('userData')
const status = await getRuntimeEnvironmentStatus(userDataPath, input.environmentId, 15_000)
if (
status.ok !== true ||
status.result.capabilities?.includes(SKILL_INSTALL_CAPABILITY) !== true
) {
return { status: 'unsupported' as const, message: SKILL_INSTALL_UPDATE_REQUIRED_MESSAGE }
}
return {
status: 'ok' as const,
value: await installSkillOnRemoteRuntime({
userDataPath,
environmentId: input.environmentId,
request,
capabilities: status.result.capabilities ?? [],
requireHttps: app.isPackaged,
signal
})
}
} catch (error) {
const failure = skillInstallFailureFromError(error)
if (!failure) {
throw error
}
return {
status: 'ok' as const,
value: {
operationId: request.operationId,
status: failure.category === 'cancelled' ? ('cancelled' as const) : ('failed' as const),
name: grant.version.name,
packageDigest: request.package.packageDigest,
placements: [],
errorCategory: failure.code,
failure
}
}
}
}
@@ -0,0 +1,21 @@
import { describe, expect, it } from 'vitest'
import { SkillRemoteInstallCancellation } from './skill-remote-install-cancellation'
describe('SkillRemoteInstallCancellation', () => {
it('aborts an active remote transfer and forgets it after settlement', () => {
const operations = new SkillRemoteInstallCancellation()
const signal = operations.begin('operation-1')
expect(operations.cancel('operation-1')).toBe(true)
expect(signal.aborted).toBe(true)
operations.finish('operation-1', signal)
expect(operations.cancel('operation-1')).toBe(false)
})
it('rejects overlapping operation identities', () => {
const operations = new SkillRemoteInstallCancellation()
operations.begin('operation-1')
expect(() => operations.begin('operation-1')).toThrow('skill-install-operation-in-progress')
})
})
@@ -0,0 +1,24 @@
export class SkillRemoteInstallCancellation {
private readonly operations = new Map<string, AbortController>()
begin(operationId: string): AbortSignal {
if (this.operations.has(operationId)) {
throw new Error('skill-install-operation-in-progress')
}
const controller = new AbortController()
this.operations.set(operationId, controller)
return controller.signal
}
finish(operationId: string, signal: AbortSignal): void {
if (this.operations.get(operationId)?.signal === signal) {
this.operations.delete(operationId)
}
}
cancel(operationId: string): boolean {
const controller = this.operations.get(operationId)
controller?.abort()
return Boolean(controller)
}
}
@@ -54,6 +54,7 @@ export async function installSkillOnRemoteRuntime(input: {
request: SkillInstallRequest
capabilities: readonly RuntimeCapability[]
requireHttps: boolean
signal?: AbortSignal
}): Promise<SkillInstallResult> {
if (input.request.ingress.kind !== 'download-grant') {
throw new Error('skill-install-remote-ingress-invalid')
@@ -74,7 +75,8 @@ export async function installSkillOnRemoteRuntime(input: {
environmentId: input.environmentId,
package: input.request.package,
grant: input.request.ingress,
requireHttps: input.requireHttps
requireHttps: input.requireHttps,
signal: input.signal
})
try {
const staged = await install(input.userDataPath, input.environmentId, {
@@ -52,6 +52,8 @@ describe('installSkillOnSshHost', () => {
it('uses client-mediated upload only after direct host download fails', async () => {
const bytes = Buffer.from('private skill archive')
let received = 0
let chunkAttempts = 0
let commitAttempts = 0
const requestHostRpc = vi.fn(async (method: string, params: unknown) => {
if (method === 'relay.status') {
return {
@@ -69,11 +71,22 @@ describe('installSkillOnSshHost', () => {
return { uploadId: 'upload_1', chunkBytes: 256 * 1024 }
}
if (method === 'skills.uploadChunk') {
chunkAttempts += 1
const chunk = params as { offset: number; bytesBase64: string }
received = chunk.offset + Buffer.from(chunk.bytesBase64, 'base64').length
if (chunkAttempts === 1) {
throw new Error('connection dropped after receiver write')
}
return { acknowledgedOffset: received }
}
if (method === 'skills.commitUpload' || method === 'skills.cancelUpload') {
if (method === 'skills.commitUpload') {
commitAttempts += 1
if (commitAttempts === 1) {
throw new Error('connection dropped after receiver commit')
}
return { ok: true }
}
if (method === 'skills.cancelUpload') {
return { ok: true }
}
throw new Error(`unexpected method ${method}`)
@@ -97,6 +110,8 @@ describe('installSkillOnSshHost', () => {
'skills.install',
'skills.beginUpload',
'skills.uploadChunk',
'skills.uploadChunk',
'skills.commitUpload',
'skills.commitUpload',
'skills.install',
'skills.cancelUpload'
+25 -10
View File
@@ -30,6 +30,7 @@ import {
import { SKILL_UPLOAD_CHUNK_MAX_BYTES } from '../../shared/skill-upload-session-contract'
import type { IPtyProvider } from '../providers/pty-provider-contract'
import { downloadSkillPackageGrant } from './skill-package-download'
import { retrySkillTransferRpc } from './skill-transfer-rpc-retry'
const REQUEST_TIMEOUT_MS = 5 * 60_000
const DIRECT_DOWNLOAD_FAILURE = 'skill-download-transport-failed'
@@ -225,11 +226,18 @@ async function transferSkillPackageToSshHost(
if (read.bytesRead !== bytes.length) {
throw new Error('skill-transfer-source-changed')
}
const acknowledged = (await client(
SKILL_SSH_RELAY_UPLOAD_CHUNK_METHOD,
{ uploadId, offset, bytesBase64: bytes.toString('base64') },
{ timeoutMs: REQUEST_TIMEOUT_MS, signal: input.signal }
)) as { acknowledgedOffset: number }
const acknowledged = (await retrySkillTransferRpc({
signal: input.signal,
retryable: (error) =>
typeof (error as { code?: unknown })?.code !== 'number' &&
(error as Error)?.name !== 'AbortError',
call: () =>
client(
SKILL_SSH_RELAY_UPLOAD_CHUNK_METHOD,
{ uploadId, offset, bytesBase64: bytes.toString('base64') },
{ timeoutMs: REQUEST_TIMEOUT_MS, signal: input.signal }
)
})) as { acknowledgedOffset: number }
if (acknowledged.acknowledgedOffset !== offset + bytes.length) {
throw new Error('skill-transfer-ack-invalid')
}
@@ -238,11 +246,18 @@ async function transferSkillPackageToSshHost(
} finally {
await handle.close()
}
await client(
SKILL_SSH_RELAY_COMMIT_UPLOAD_METHOD,
{ uploadId },
{ timeoutMs: REQUEST_TIMEOUT_MS, signal: input.signal }
)
await retrySkillTransferRpc({
signal: input.signal,
retryable: (error) =>
typeof (error as { code?: unknown })?.code !== 'number' &&
(error as Error)?.name !== 'AbortError',
call: () =>
client(
SKILL_SSH_RELAY_COMMIT_UPLOAD_METHOD,
{ uploadId },
{ timeoutMs: REQUEST_TIMEOUT_MS, signal: input.signal }
)
})
const committedId = uploadId
uploadId = null
return committedId
@@ -0,0 +1,30 @@
import { SKILL_INSTALL_CANCELLED_FAILURE } from '../../shared/skill-install-failure'
import { SkillInstallOperationError } from './skill-install-operation-error'
const TRANSFER_RPC_ATTEMPTS = 3
export function throwIfSkillTransferCancelled(signal?: AbortSignal): void {
if (signal?.aborted) {
throw new SkillInstallOperationError(SKILL_INSTALL_CANCELLED_FAILURE)
}
}
export async function retrySkillTransferRpc<T>(input: {
call: () => Promise<T>
signal?: AbortSignal
retryable?: (error: unknown) => boolean
}): Promise<T> {
for (let attempt = 1; ; attempt += 1) {
throwIfSkillTransferCancelled(input.signal)
try {
const result = await input.call()
throwIfSkillTransferCancelled(input.signal)
return result
} catch (error) {
throwIfSkillTransferCancelled(input.signal)
if (attempt >= TRANSFER_RPC_ATTEMPTS || input.retryable?.(error) === false) {
throw error
}
}
}
}
@@ -1,5 +1,5 @@
import { createHash } from 'node:crypto'
import { mkdtemp, readFile, rm } from 'node:fs/promises'
import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it } from 'vitest'
@@ -22,6 +22,19 @@ function identity(bytes: Buffer) {
}
describe('SkillUploadSessionService', () => {
it('removes abandoned staging bytes when a runtime starts a fresh service', async () => {
const root = await mkdtemp(join(tmpdir(), 'orca-skill-upload-session-'))
roots.push(root)
const uploads = join(root, 'uploads')
await mkdir(uploads)
await writeFile(join(uploads, 'abandoned.tar.gz'), 'partial package')
const service = new SkillUploadSessionService(uploads)
await service.begin({ package: identity(Buffer.from('new package')) })
expect(await readdir(uploads)).not.toContain('abandoned.tar.gz')
})
it('accepts monotonic chunks, acknowledges identical retries, and transfers ownership', async () => {
const root = await mkdtemp(join(tmpdir(), 'orca-skill-upload-session-'))
roots.push(root)
@@ -43,7 +56,8 @@ describe('SkillUploadSessionService', () => {
offset: first.length,
bytesBase64: second.toString('base64')
})
await service.commit(begun.uploadId)
await expect(service.commit(begun.uploadId)).resolves.toEqual({ uploadId: begun.uploadId })
await expect(service.commit(begun.uploadId)).resolves.toEqual({ uploadId: begun.uploadId })
const staged = await service.take(begun.uploadId, packageIdentity)
await expect(readFile(staged.archivePath)).resolves.toEqual(bytes)
await staged.cleanup()
@@ -87,7 +87,14 @@ export class SkillUploadSessionService {
}
async commit(uploadId: string): Promise<{ uploadId: string }> {
const session = this.requireActive(uploadId)
const session = this.sessions.get(uploadId)
if (!session || Date.now() - session.touchedAt > SESSION_IDLE_MS) {
throw new Error('skill-upload-session-unavailable')
}
if (session.committed) {
session.touchedAt = Date.now()
return { uploadId }
}
if (session.bytesReceived !== session.package.compressedBytes || !session.handle) {
throw new Error('skill-upload-size-mismatch')
}
+27 -1
View File
@@ -1,5 +1,10 @@
import { describe, expect, it } from 'vitest'
import { SkillInstallRequestSchema, SkillInstallResultSchema } from './skill-install-contract'
import {
SkillInstallRequestSchema,
SkillInstallResultSchema,
SkillPackageIdentitySchema
} from './skill-install-contract'
import { SKILL_PACKAGE_MAX_COMPRESSED_BYTES } from './skill-package-manifest'
const packageIdentity = {
packageId: 'package_1',
@@ -10,6 +15,27 @@ const packageIdentity = {
}
describe('skill install wire compatibility', () => {
it('enforces compressed package bytes below, at, and above the V1 limit', () => {
expect(
SkillPackageIdentitySchema.safeParse({
...packageIdentity,
compressedBytes: SKILL_PACKAGE_MAX_COMPRESSED_BYTES - 1
}).success
).toBe(true)
expect(
SkillPackageIdentitySchema.safeParse({
...packageIdentity,
compressedBytes: SKILL_PACKAGE_MAX_COMPRESSED_BYTES
}).success
).toBe(true)
expect(
SkillPackageIdentitySchema.safeParse({
...packageIdentity,
compressedBytes: SKILL_PACKAGE_MAX_COMPRESSED_BYTES + 1
}).success
).toBe(false)
})
it('accepts an older client request that omits every additive field', () => {
expect(
SkillInstallRequestSchema.parse({
+8 -7
View File
@@ -102,15 +102,16 @@ describe('skill package manifest', () => {
expect(() => parseSkillPackageManifest(manifest([{ ...file, size: fourMiB + 1 }]))).toThrow(
'skill-package-manifest-invalid'
)
const total = (extra: number): SkillPackageFile[] => [
const total = (lastSize: number): SkillPackageFile[] => [
{ ...file, size: 0 },
...Array.from({ length: 8 }, (_, index) => packageFile(`payload-${index}.bin`, fourMiB)),
...(extra ? [packageFile('z-extra.bin', extra)] : [])
...Array.from({ length: 7 }, (_, index) => packageFile(`payload-${index}.bin`, fourMiB)),
packageFile('payload-7.bin', lastSize)
]
expect(() => parseSkillPackageManifest(manifest(total(0)))).not.toThrow()
expect(() => parseSkillPackageManifest(manifest(total(1)))).toThrow(
'skill-package-total-size-limit'
)
expect(() => parseSkillPackageManifest(manifest(total(fourMiB - 1)))).not.toThrow()
expect(() => parseSkillPackageManifest(manifest(total(fourMiB)))).not.toThrow()
expect(() =>
parseSkillPackageManifest(manifest([...total(fourMiB), packageFile('z-extra.bin', 1)]))
).toThrow('skill-package-total-size-limit')
})
it('rejects Unicode normalization and case-fold collisions', () => {