diff --git a/src/main/ipc/skill-cloud-ipc-handlers.ts b/src/main/ipc/skill-cloud-ipc-handlers.ts index 475b3c0bbaa..cc7f9b69e41 100644 --- a/src/main/ipc/skill-cloud-ipc-handlers.ts +++ b/src/main/ipc/skill-cloud-ipc-handlers.ts @@ -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 | z.infer -) { - 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) => Promise @@ -187,6 +114,25 @@ function registerSharingHandlers( } function registerCloudInstallHandlers(runtime: OrcaRuntimeService): void { + const remoteInstallCancellation = new SkillRemoteInstallCancellation() + const installAuthorizedGrant = async ( + grant: SkillCloudDownloadGrant, + input: z.infer | z.infer + ) => { + 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), {}) diff --git a/src/main/skills/skill-client-mediated-transfer.test.ts b/src/main/skills/skill-client-mediated-transfer.test.ts index c58e94a3762..90d60c70025 100644 --- a/src/main/skills/skill-client-mediated-transfer.test.ts +++ b/src/main/skills/skill-client-mediated-transfer.test.ts @@ -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', diff --git a/src/main/skills/skill-client-mediated-transfer.ts b/src/main/skills/skill-client-mediated-transfer.ts index f59a801b23e..6074931bda9 100644 --- a/src/main/skills/skill-client-mediated-transfer.ts +++ b/src/main/skills/skill-client-mediated-transfer.ts @@ -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( 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 }> { 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 { diff --git a/src/main/skills/skill-cloud-grant-installation.test.ts b/src/main/skills/skill-cloud-grant-installation.test.ts new file mode 100644 index 00000000000..6804dba4016 --- /dev/null +++ b/src/main/skills/skill-cloud-grant-installation.test.ts @@ -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 }) + ) + }) +}) diff --git a/src/main/skills/skill-cloud-grant-installation.ts b/src/main/skills/skill-cloud-grant-installation.ts new file mode 100644 index 00000000000..c9a6e5b7725 --- /dev/null +++ b/src/main/skills/skill-cloud-grant-installation.ts @@ -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 + } + } + } +} diff --git a/src/main/skills/skill-remote-install-cancellation.test.ts b/src/main/skills/skill-remote-install-cancellation.test.ts new file mode 100644 index 00000000000..2732f40af6e --- /dev/null +++ b/src/main/skills/skill-remote-install-cancellation.test.ts @@ -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') + }) +}) diff --git a/src/main/skills/skill-remote-install-cancellation.ts b/src/main/skills/skill-remote-install-cancellation.ts new file mode 100644 index 00000000000..cc51ee0f1d4 --- /dev/null +++ b/src/main/skills/skill-remote-install-cancellation.ts @@ -0,0 +1,24 @@ +export class SkillRemoteInstallCancellation { + private readonly operations = new Map() + + 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) + } +} diff --git a/src/main/skills/skill-remote-install-service.ts b/src/main/skills/skill-remote-install-service.ts index 4bb752aa0a4..baa42cb7354 100644 --- a/src/main/skills/skill-remote-install-service.ts +++ b/src/main/skills/skill-remote-install-service.ts @@ -54,6 +54,7 @@ export async function installSkillOnRemoteRuntime(input: { request: SkillInstallRequest capabilities: readonly RuntimeCapability[] requireHttps: boolean + signal?: AbortSignal }): Promise { 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, { diff --git a/src/main/skills/skill-ssh-relay-service.test.ts b/src/main/skills/skill-ssh-relay-service.test.ts index 405236eae1f..29605cb9ad0 100644 --- a/src/main/skills/skill-ssh-relay-service.test.ts +++ b/src/main/skills/skill-ssh-relay-service.test.ts @@ -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' diff --git a/src/main/skills/skill-ssh-relay-service.ts b/src/main/skills/skill-ssh-relay-service.ts index 45f037729e6..ca801bb2da3 100644 --- a/src/main/skills/skill-ssh-relay-service.ts +++ b/src/main/skills/skill-ssh-relay-service.ts @@ -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 diff --git a/src/main/skills/skill-transfer-rpc-retry.ts b/src/main/skills/skill-transfer-rpc-retry.ts new file mode 100644 index 00000000000..caba8780468 --- /dev/null +++ b/src/main/skills/skill-transfer-rpc-retry.ts @@ -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(input: { + call: () => Promise + signal?: AbortSignal + retryable?: (error: unknown) => boolean +}): Promise { + 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 + } + } + } +} diff --git a/src/main/skills/skill-upload-session-service.test.ts b/src/main/skills/skill-upload-session-service.test.ts index d250aa3d2fe..df7b2b4f8f1 100644 --- a/src/main/skills/skill-upload-session-service.test.ts +++ b/src/main/skills/skill-upload-session-service.test.ts @@ -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() diff --git a/src/main/skills/skill-upload-session-service.ts b/src/main/skills/skill-upload-session-service.ts index 9db60e35078..4c9dc690bb1 100644 --- a/src/main/skills/skill-upload-session-service.ts +++ b/src/main/skills/skill-upload-session-service.ts @@ -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') } diff --git a/src/shared/skill-install-contract.test.ts b/src/shared/skill-install-contract.test.ts index 9a36f0dc16a..a41c24f3f12 100644 --- a/src/shared/skill-install-contract.test.ts +++ b/src/shared/skill-install-contract.test.ts @@ -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({ diff --git a/src/shared/skill-package-manifest.test.ts b/src/shared/skill-package-manifest.test.ts index 3b744ff5e4e..5e1a124025b 100644 --- a/src/shared/skill-package-manifest.test.ts +++ b/src/shared/skill-package-manifest.test.ts @@ -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', () => {