diff --git a/packages/server-ai/src/sandbox/commands/handlers/acquire-backend.handler.spec.ts b/packages/server-ai/src/sandbox/commands/handlers/acquire-backend.handler.spec.ts index 2e94c69694..f9bf218009 100644 --- a/packages/server-ai/src/sandbox/commands/handlers/acquire-backend.handler.spec.ts +++ b/packages/server-ai/src/sandbox/commands/handlers/acquire-backend.handler.spec.ts @@ -47,11 +47,13 @@ describe('SandboxAcquireBackendHandler', () => { expect(registry.get).toHaveBeenCalledTimes(1) expect(provider.create).toHaveBeenCalledTimes(1) - expect(first).toEqual({ - provider: 'local-shell-sandbox', - workingDirectory: '/workspace/a', - backend - }) + expect(first).toEqual( + expect.objectContaining({ + provider: 'local-shell-sandbox', + workingDirectory: '/workspace/a', + backend + }) + ) expect(second).toBe(first) }) @@ -90,16 +92,102 @@ describe('SandboxAcquireBackendHandler', () => { ) expect(provider.create).toHaveBeenCalledTimes(2) - expect(first).toEqual({ - provider: 'local-shell-sandbox', - workingDirectory: '/workspace/a', - backend: backendA - }) - expect(second).toEqual({ - provider: 'local-shell-sandbox', - workingDirectory: '/workspace/b', - backend: backendB - }) + expect(first).toEqual( + expect.objectContaining({ + provider: 'local-shell-sandbox', + workingDirectory: '/workspace/a', + backend: backendA + }) + ) + expect(second).toEqual( + expect.objectContaining({ + provider: 'local-shell-sandbox', + workingDirectory: '/workspace/b', + backend: backendB + }) + ) + }) + + it('creates distinct backends for different workspace bindings in the same scope and working directory', async () => { + const backendA = { + id: 'sandbox-a', + execute: jest.fn() + } + const backendB = { + id: 'sandbox-b', + execute: jest.fn() + } + provider.create.mockResolvedValueOnce(backendA).mockResolvedValueOnce(backendB) + + const first = await handler.execute( + new SandboxAcquireBackendCommand({ + tenantId: 'tenant-1', + provider: 'local-shell-sandbox', + workingDirectory: '/workspace', + workspaceBinding: { + volumeRoot: '/tmp/workspace-a', + workspaceRoot: '/workspace', + workspacePath: '/workspace' + }, + workFor: { + type: 'user', + id: 'user-1' + } + }) + ) + const second = await handler.execute( + new SandboxAcquireBackendCommand({ + tenantId: 'tenant-1', + provider: 'local-shell-sandbox', + workingDirectory: '/workspace', + workspaceBinding: { + volumeRoot: '/tmp/workspace-b', + workspaceRoot: '/workspace', + workspacePath: '/workspace' + }, + workFor: { + type: 'user', + id: 'user-1' + } + }) + ) + + expect(provider.create).toHaveBeenCalledTimes(2) + expect(first.backend).toBe(backendA) + expect(second.backend).toBe(backendB) + }) + + it('returns workspace binding with the sandbox context', async () => { + const backend = { + id: 'sandbox-a', + execute: jest.fn() + } + const workspaceBinding = { + volumeRoot: '/tmp/xpert-workspace', + workspaceRoot: '/tmp/xpert-workspace', + workspacePath: '/tmp/xpert-workspace' + } + provider.create.mockResolvedValue(backend) + + const result = await handler.execute( + new SandboxAcquireBackendCommand({ + tenantId: 'tenant-1', + provider: 'local-shell-sandbox', + workingDirectory: '/tmp/xpert-workspace', + workspaceBinding, + workFor: { + type: 'user', + id: 'user-1' + } + }) + ) + + expect(provider.create).toHaveBeenCalledWith( + expect.objectContaining({ + workspaceBinding + }) + ) + expect((result as { workspaceBinding?: unknown }).workspaceBinding).toBe(workspaceBinding) }) it('throws when provider is omitted', async () => { diff --git a/packages/server-ai/src/sandbox/commands/handlers/acquire-backend.handler.ts b/packages/server-ai/src/sandbox/commands/handlers/acquire-backend.handler.ts index e3b1eed431..5167e7d36b 100644 --- a/packages/server-ai/src/sandbox/commands/handlers/acquire-backend.handler.ts +++ b/packages/server-ai/src/sandbox/commands/handlers/acquire-backend.handler.ts @@ -1,12 +1,16 @@ import { TSandboxConfigurable } from '@xpert-ai/contracts' import { CommandHandler, ICommandHandler } from '@nestjs/cqrs' -import { SandboxProviderRegistry } from '@xpert-ai/plugin-sdk' +import { SandboxProviderCreateOptions, SandboxProviderRegistry } from '@xpert-ai/plugin-sdk' import { SandboxAcquireBackendCommand } from '../acquire-backend.command' type SandboxInstance = { configurable: TSandboxConfigurable } +type SandboxRuntimeConfigurable = TSandboxConfigurable & { + workspaceBinding?: SandboxProviderCreateOptions['workspaceBinding'] +} + @CommandHandler(SandboxAcquireBackendCommand) export class SandboxAcquireBackendHandler implements ICommandHandler @@ -25,7 +29,7 @@ export class SandboxAcquireBackendHandler } const sessionKey = this.getSessionKey(workFor.type, workFor.id) - const instanceKey = this.getInstanceKey(provider, workingDirectory) + const instanceKey = this.getInstanceKey(provider, workingDirectory, workspaceBinding) const sessionMap = this.instances.get(sessionKey) ?? new Map() const existing = sessionMap.get(instanceKey) if (existing) { @@ -40,12 +44,15 @@ export class SandboxAcquireBackendHandler workingDirectory, workspaceBinding }) - const configurable: TSandboxConfigurable = { + const configurable: SandboxRuntimeConfigurable = { environmentId: environmentId ?? null, provider, workingDirectory, backend } + if (workspaceBinding) { + configurable.workspaceBinding = workspaceBinding + } sessionMap.set(instanceKey, { configurable }) this.instances.set(sessionKey, sessionMap) return configurable @@ -55,7 +62,12 @@ export class SandboxAcquireBackendHandler return `${workForType}:${workForId}` } - private getInstanceKey(provider?: string | null, workingDirectory?: string | null) { - return `${provider ?? '__default__'}:${workingDirectory ?? '__default__'}` + private getInstanceKey( + provider?: string | null, + workingDirectory?: string | null, + workspaceBinding?: SandboxProviderCreateOptions['workspaceBinding'] + ) { + const workspaceIdentity = workspaceBinding?.volumeRoot ?? workspaceBinding?.bindSource ?? '' + return `${provider ?? '__default__'}:${workingDirectory ?? '__default__'}:${workspaceIdentity}` } } diff --git a/packages/server-ai/src/sandbox/commands/handlers/index.ts b/packages/server-ai/src/sandbox/commands/handlers/index.ts index 12763e9374..6637f28ed5 100644 --- a/packages/server-ai/src/sandbox/commands/handlers/index.ts +++ b/packages/server-ai/src/sandbox/commands/handlers/index.ts @@ -3,6 +3,7 @@ import { SandboxGetManagedServiceLogsHandler } from './get-managed-service-logs. import { SandboxListManagedServicesHandler } from './list-managed-services.handler' import { SandboxRestartManagedServiceHandler } from './restart-managed-service.handler' import { SandboxCopyFileHandler } from './sandbox.copy-file.handler' +import { SandboxCopyTreeHandler } from './sandbox.copy-tree.handler' import { SandboxStartManagedServiceHandler } from './start-managed-service.handler' import { SandboxStopManagedServiceHandler } from './stop-managed-service.handler' import { SandboxVMHandler } from './vm.handler' @@ -11,6 +12,7 @@ export const CommandHandlers = [ SandboxVMHandler, SandboxAcquireBackendHandler, SandboxCopyFileHandler, + SandboxCopyTreeHandler, SandboxStartManagedServiceHandler, SandboxListManagedServicesHandler, SandboxGetManagedServiceLogsHandler, diff --git a/packages/server-ai/src/sandbox/commands/handlers/sandbox.copy-tree.handler.spec.ts b/packages/server-ai/src/sandbox/commands/handlers/sandbox.copy-tree.handler.spec.ts new file mode 100644 index 0000000000..3be849716b --- /dev/null +++ b/packages/server-ai/src/sandbox/commands/handlers/sandbox.copy-tree.handler.spec.ts @@ -0,0 +1,287 @@ +import { mkdtemp, mkdir, readFile, rm, stat, writeFile } from 'fs/promises' +import { join } from 'path' +import { tmpdir } from 'os' +import { Logger } from '@nestjs/common' +import { SandboxCopyTreeHandler } from './sandbox.copy-tree.handler' +import { SandboxCopyTreeCommand } from '../sandbox.copy-tree.command' + +describe('SandboxCopyTreeHandler', () => { + let tempDir: string + let cacheManager: { + get: jest.Mock + set: jest.Mock + } + let backend: { + id: string + execute: jest.Mock + uploadFiles: jest.Mock + } + let logSpy: jest.SpyInstance + + beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), 'sandbox-copy-tree-')) + cacheManager = { + get: jest.fn().mockResolvedValue({}), + set: jest.fn().mockResolvedValue(undefined) + } + backend = { + id: 'backend-1', + execute: jest.fn().mockResolvedValue({ + output: '', + exitCode: 0, + truncated: false + }), + uploadFiles: jest.fn().mockResolvedValue([]) + } + logSpy = jest.spyOn(Logger.prototype, 'log').mockImplementation() + }) + + afterEach(async () => { + logSpy.mockRestore() + await rm(tempDir, { recursive: true, force: true }) + }) + + it('copies a local directory tree directly when the sandbox workspace is locally mapped', async () => { + await mkdir(join(tempDir, 'scripts'), { recursive: true }) + await writeFile(join(tempDir, 'SKILL.md'), 'alpha') + await writeFile(join(tempDir, 'scripts', 'run.sh'), 'beta') + const volumeRoot = await mkdtemp(join(tmpdir(), 'sandbox-copy-tree-volume-')) + + const handler = new SandboxCopyTreeHandler(cacheManager as never) + const result = await handler.execute( + new SandboxCopyTreeCommand( + { + backend, + workspaceBinding: { + volumeRoot, + workspaceRoot: '/workspace', + workspacePath: '/workspace', + containerMountPath: '/workspace' + } + } as never, + { + version: 'v1', + localPath: tempDir, + containerPath: '/workspace/.xpert/skills/a', + overwrite: true + } + ) + ) + + expect(backend.execute).not.toHaveBeenCalled() + expect(backend.uploadFiles).not.toHaveBeenCalled() + await expect(readFile(join(volumeRoot, '.xpert/skills/a/SKILL.md'), 'utf-8')).resolves.toBe('alpha') + await expect(readFile(join(volumeRoot, '.xpert/skills/a/scripts/run.sh'), 'utf-8')).resolves.toBe('beta') + expect(cacheManager.set).toHaveBeenCalledWith(`sandbox:copy-tree:backend-1:${volumeRoot}`, { + '/workspace/.xpert/skills/a': 'v1' + }) + expect(result).toEqual( + expect.objectContaining({ + containerId: 'backend-1', + localPath: tempDir, + containerPath: '/workspace/.xpert/skills/a', + version: 'v1', + status: 'success', + fileCount: 2, + totalBytes: 9, + scanMs: expect.any(Number), + uploadMs: expect.any(Number), + totalMs: expect.any(Number), + mode: 'local' + }) + ) + expect(logSpy).toHaveBeenCalledWith(expect.stringContaining('Copied tree')) + expect(logSpy).toHaveBeenCalledWith(expect.stringContaining('mode=local')) + expect(logSpy).toHaveBeenCalledWith(expect.stringContaining('files=2 bytes=9')) + await rm(volumeRoot, { recursive: true, force: true }) + }) + + it('falls back to uploadFiles when the sandbox workspace is not locally mapped', async () => { + await mkdir(join(tempDir, 'scripts'), { recursive: true }) + await writeFile(join(tempDir, 'SKILL.md'), 'alpha') + await writeFile(join(tempDir, 'scripts', 'run.sh'), 'beta') + backend.uploadFiles.mockImplementation(async (files: Array<[string, Uint8Array]>) => + files.map(([filePath]) => ({ path: filePath, error: null })) + ) + + const handler = new SandboxCopyTreeHandler(cacheManager as never) + await handler.execute( + new SandboxCopyTreeCommand({ backend } as never, { + version: 'v1', + localPath: tempDir, + containerPath: '/workspace/.xpert/skills/a', + overwrite: true + }) + ) + + expect(backend.uploadFiles).toHaveBeenCalledTimes(1) + expect(backend.uploadFiles).toHaveBeenCalledWith( + expect.arrayContaining([ + ['/workspace/.xpert/skills/a/SKILL.md', expect.any(Uint8Array)], + ['/workspace/.xpert/skills/a/scripts/run.sh', expect.any(Uint8Array)] + ]) + ) + }) + + it('falls back to uploadFiles when the target path is outside the mapped workspace', async () => { + await writeFile(join(tempDir, 'SKILL.md'), 'alpha') + backend.uploadFiles.mockImplementation(async (files: Array<[string, Uint8Array]>) => + files.map(([filePath]) => ({ path: filePath, error: null })) + ) + + const handler = new SandboxCopyTreeHandler(cacheManager as never) + await handler.execute( + new SandboxCopyTreeCommand( + { + backend, + workspaceBinding: { + volumeRoot: tempDir, + workspaceRoot: '/workspace', + workspacePath: '/workspace' + } + } as never, + { + version: 'v1', + localPath: tempDir, + containerPath: '/tmp/.xpert/skills/a', + overwrite: true + } + ) + ) + + expect(backend.execute).toHaveBeenCalledWith(`rm -rf '/tmp/.xpert/skills/a'`) + expect(backend.uploadFiles).toHaveBeenCalledTimes(1) + expect(backend.uploadFiles).toHaveBeenCalledWith( + expect.arrayContaining([['/tmp/.xpert/skills/a/SKILL.md', expect.any(Uint8Array)]]) + ) + }) + + it('keeps an empty directory when copying directly', async () => { + const volumeRoot = await mkdtemp(join(tmpdir(), 'sandbox-copy-tree-volume-')) + + const handler = new SandboxCopyTreeHandler(cacheManager as never) + const result = await handler.execute( + new SandboxCopyTreeCommand( + { + backend, + workspaceBinding: { + volumeRoot, + workspaceRoot: '/workspace', + workspacePath: '/workspace' + } + } as never, + { + version: 'v1', + localPath: tempDir, + containerPath: '/workspace/.xpert/skills/empty' + } + ) + ) + + await expect(stat(join(volumeRoot, '.xpert/skills/empty'))).resolves.toEqual(expect.objectContaining({})) + expect(backend.execute).not.toHaveBeenCalled() + expect(backend.uploadFiles).not.toHaveBeenCalled() + expect(result).toEqual( + expect.objectContaining({ + fileCount: 0, + totalBytes: 0, + mode: 'local' + }) + ) + await rm(volumeRoot, { recursive: true, force: true }) + }) + + it('skips copying when the same tree version has already been synced', async () => { + cacheManager.get.mockResolvedValue({ + '/workspace/.xpert/skills/a': 'v1' + }) + + const handler = new SandboxCopyTreeHandler(cacheManager as never) + const result = await handler.execute( + new SandboxCopyTreeCommand({ backend } as never, { + version: 'v1', + localPath: tempDir, + containerPath: '/workspace/.xpert/skills/a' + }) + ) + + expect(backend.execute).not.toHaveBeenCalled() + expect(backend.uploadFiles).not.toHaveBeenCalled() + expect(cacheManager.set).not.toHaveBeenCalled() + expect(result).toEqual( + expect.objectContaining({ + containerId: 'backend-1', + containerPath: '/workspace/.xpert/skills/a', + version: 'v1', + status: 'skipped', + reason: 'Same version, skip syncing', + totalMs: expect.any(Number) + }) + ) + expect(logSpy).toHaveBeenCalledWith(expect.stringContaining('Skip copy tree')) + }) + + it('scopes the version cache by workspace binding identity', async () => { + await writeFile(join(tempDir, 'SKILL.md'), 'alpha') + const volumeRootA = await mkdtemp(join(tmpdir(), 'sandbox-copy-tree-volume-a-')) + const volumeRootB = await mkdtemp(join(tmpdir(), 'sandbox-copy-tree-volume-b-')) + cacheManager.get.mockImplementation(async (key: string) => + key === `sandbox:copy-tree:backend-1:${volumeRootA}` + ? { + '/workspace/.xpert/skills/a': 'v1' + } + : {} + ) + + const handler = new SandboxCopyTreeHandler(cacheManager as never) + const result = await handler.execute( + new SandboxCopyTreeCommand( + { + backend, + workspaceBinding: { + volumeRoot: volumeRootB, + workspaceRoot: '/workspace', + workspacePath: '/workspace' + } + } as never, + { + version: 'v1', + localPath: tempDir, + containerPath: '/workspace/.xpert/skills/a' + } + ) + ) + + expect(cacheManager.get).toHaveBeenCalledWith(`sandbox:copy-tree:backend-1:${volumeRootB}`) + expect(result).toEqual( + expect.objectContaining({ + status: 'success', + mode: 'local' + }) + ) + await expect(readFile(join(volumeRootB, '.xpert/skills/a/SKILL.md'), 'utf-8')).resolves.toBe('alpha') + expect(cacheManager.set).toHaveBeenCalledWith(`sandbox:copy-tree:backend-1:${volumeRootB}`, { + '/workspace/.xpert/skills/a': 'v1' + }) + await rm(volumeRootA, { recursive: true, force: true }) + await rm(volumeRootB, { recursive: true, force: true }) + }) + + it('rejects non-directory local paths', async () => { + const filePath = join(tempDir, 'SKILL.md') + await writeFile(filePath, 'alpha') + + const handler = new SandboxCopyTreeHandler(cacheManager as never) + await expect( + handler.execute( + new SandboxCopyTreeCommand({ backend } as never, { + version: 'v1', + localPath: filePath, + containerPath: '/workspace/.xpert/skills/a' + }) + ) + ).rejects.toThrow(`localPath is not a directory: ${filePath}`) + + expect(backend.uploadFiles).not.toHaveBeenCalled() + }) +}) diff --git a/packages/server-ai/src/sandbox/commands/handlers/sandbox.copy-tree.handler.ts b/packages/server-ai/src/sandbox/commands/handlers/sandbox.copy-tree.handler.ts new file mode 100644 index 0000000000..329b7e3d7f --- /dev/null +++ b/packages/server-ai/src/sandbox/commands/handlers/sandbox.copy-tree.handler.ts @@ -0,0 +1,313 @@ +import { CACHE_MANAGER } from '@nestjs/cache-manager' +import { HttpException, Inject, Logger } from '@nestjs/common' +import { CommandHandler, ICommandHandler } from '@nestjs/cqrs' +import type { Cache } from 'cache-manager' +import * as fs from 'fs' +import * as path from 'path' +import { resolveSandboxBackend } from '@xpert-ai/plugin-sdk' +import type { SandboxBackendProtocol } from '@xpert-ai/plugin-sdk' +import type { WorkspaceBinding } from '../../../shared' +import { SandboxCopyTreeCommand, SandboxCopyTreeMode, SandboxCopyTreeStatus } from '../sandbox.copy-tree.command' + +type CopyTreePayload = { + files: Array<[string, Uint8Array]> + fileCount: number + totalBytes: number +} + +type CopyTreeStats = { + fileCount: number + totalBytes: number +} + +type SandboxCopyTreeRuntime = { + workspaceBinding?: WorkspaceBinding +} + +@CommandHandler(SandboxCopyTreeCommand) +export class SandboxCopyTreeHandler implements ICommandHandler { + private readonly logger = new Logger(SandboxCopyTreeHandler.name) + + constructor( + @Inject(CACHE_MANAGER) + private readonly cacheManager: Cache + ) {} + + async execute(command: SandboxCopyTreeCommand) { + const { localPath, containerPath, version, overwrite = true } = command.copyTree + const totalStart = Date.now() + const backend = this.getSandboxBackend(command.sandbox) + + if (!version) { + throw new HttpException(`Version is required for SandboxCopyTreeCommand`, 400) + } + if (!localPath) { + throw new HttpException(`localPath is required for SandboxCopyTreeCommand`, 400) + } + if (!containerPath) { + throw new HttpException(`containerPath is required for SandboxCopyTreeCommand`, 400) + } + + const versionCacheKey = this.getVersionCacheKey(backend.id, command.sandbox) + const initializedTrees = (await this.cacheManager.get>(versionCacheKey)) ?? {} + + const lastVersion = initializedTrees[containerPath] + if (lastVersion && String(lastVersion) === String(version)) { + const totalMs = Date.now() - totalStart + this.logger.log( + `Skip copy tree: containerPath=${containerPath} already synced with version=${version} backend=${backend.id} totalMs=${totalMs}` + ) + return { + containerId: backend.id, + containerPath, + version, + status: 'skipped' as SandboxCopyTreeStatus, + totalMs, + reason: 'Same version, skip syncing' + } + } + + const directTargetPath = this.resolveDirectTargetPath(command.sandbox, containerPath) + const scanStart = Date.now() + const copyPayload: CopyTreePayload = directTargetPath + ? { ...(await this.collectLocalTreeStats(localPath)), files: [] } + : await this.buildUploadPayload(localPath, containerPath) + const scanMs = Date.now() - scanStart + + const uploadStart = Date.now() + let mode: SandboxCopyTreeMode = 'upload' + if (directTargetPath) { + await this.copyTreeToLocalPath(localPath, directTargetPath, overwrite) + mode = 'local' + } else { + if (overwrite) { + await this.tryRemoveExisting(backend, containerPath) + } + + if (copyPayload.files.length === 0) { + await this.ensureContainerDir(backend, containerPath) + } else { + const uploadResults = await backend.uploadFiles(copyPayload.files) + const failed = uploadResults.find((result) => result.error) + if (failed) { + throw new HttpException(`Copy tree failed: path=${failed.path}, error=${failed.error}`, 500) + } + } + } + const uploadMs = Date.now() - uploadStart + const totalMs = Date.now() - totalStart + + this.logger.log( + `Copied tree ${localPath} -> sandbox:${containerPath} via backend=${backend.id} mode=${mode} files=${copyPayload.fileCount} bytes=${copyPayload.totalBytes} scanMs=${scanMs} uploadMs=${uploadMs} totalMs=${totalMs}` + ) + + initializedTrees[containerPath] = String(version) + await this.cacheManager.set(versionCacheKey, initializedTrees) + + return { + containerId: backend.id, + localPath, + containerPath, + version, + status: 'success' as SandboxCopyTreeStatus, + fileCount: copyPayload.fileCount, + totalBytes: copyPayload.totalBytes, + scanMs, + uploadMs, + totalMs, + mode + } + } + + private getSandboxBackend(sandbox: unknown): SandboxBackendProtocol { + const backend = resolveSandboxBackend(sandbox) + if (!backend) { + throw new HttpException(`Sandbox backend unavailable for tree copy`, 500) + } + return backend + } + + private getVersionCacheKey(sandboxId: string, sandbox: unknown): string { + const workspaceIdentity = this.getWorkspaceBindingIdentity(sandbox) + return workspaceIdentity ? `sandbox:copy-tree:${sandboxId}:${workspaceIdentity}` : `sandbox:copy-tree:${sandboxId}` + } + + private getWorkspaceBindingIdentity(sandbox: unknown): string { + if (!sandbox || typeof sandbox !== 'object' || Array.isArray(sandbox)) { + return '' + } + + const workspaceBinding = (sandbox as SandboxCopyTreeRuntime).workspaceBinding + if (!workspaceBinding || typeof workspaceBinding !== 'object' || Array.isArray(workspaceBinding)) { + return '' + } + + return workspaceBinding.volumeRoot || workspaceBinding.bindSource || '' + } + + private async ensureContainerDir(backend: SandboxBackendProtocol, containerPath: string): Promise { + const result = await backend.execute(`mkdir -p ${this.escapeForShell(containerPath)}`) + if (result.exitCode !== 0) { + throw new HttpException( + `Failed to create directory in sandbox: ${containerPath}, output=${result.output}`, + 500 + ) + } + } + + private async tryRemoveExisting(backend: SandboxBackendProtocol, containerPath: string): Promise { + await backend.execute(`rm -rf ${this.escapeForShell(containerPath)}`) + } + + private resolveDirectTargetPath(sandbox: unknown, containerPath: string): string | null { + const workspaceBinding = this.getWorkspaceBinding(sandbox) + if (!workspaceBinding) { + return null + } + + const workspaceRoot = this.normalizeContainerPath(workspaceBinding.workspaceRoot) + const volumeRoot = path.resolve(workspaceBinding.volumeRoot) + const normalizedContainerPath = this.normalizeContainerPath(containerPath) + const relativePath = path.posix.relative(workspaceRoot, normalizedContainerPath) + if (!relativePath || relativePath.startsWith('..') || path.posix.isAbsolute(relativePath)) { + return null + } + + const targetPath = path.resolve(volumeRoot, relativePath) + if (!this.isPathInside(volumeRoot, targetPath)) { + return null + } + return targetPath + } + + private getWorkspaceBinding(sandbox: unknown): WorkspaceBinding | null { + if (!sandbox || typeof sandbox !== 'object' || Array.isArray(sandbox)) { + return null + } + + const workspaceBinding = (sandbox as SandboxCopyTreeRuntime).workspaceBinding + if (!workspaceBinding || typeof workspaceBinding !== 'object' || Array.isArray(workspaceBinding)) { + return null + } + if (!workspaceBinding.volumeRoot || !workspaceBinding.workspaceRoot) { + return null + } + return workspaceBinding + } + + private normalizeContainerPath(value: string): string { + const normalized = path.posix.normalize(value.replace(/\\/g, '/')) + return normalized.startsWith('/') ? normalized : `/${normalized}` + } + + private isPathInside(rootPath: string, targetPath: string): boolean { + const relativePath = path.relative(rootPath, targetPath) + return relativePath === '' || (!relativePath.startsWith('..') && !path.isAbsolute(relativePath)) + } + + private async copyTreeToLocalPath(localPath: string, targetPath: string, overwrite: boolean): Promise { + await fs.promises.mkdir(path.dirname(targetPath), { recursive: true }) + if (overwrite) { + await fs.promises.rm(targetPath, { recursive: true, force: true }) + } + await fs.promises.cp(localPath, targetPath, { + recursive: true, + force: true, + dereference: true + }) + } + + private escapeForShell(input: string): string { + return `'${input.replace(/'/g, `'\"'\"'`)}'` + } + + private async collectLocalTreeStats(localPath: string): Promise { + let stats: fs.Stats + try { + stats = await fs.promises.stat(localPath) + } catch { + throw new HttpException(`localPath not found: ${localPath}`, 400) + } + + if (!stats.isDirectory()) { + throw new HttpException(`localPath is not a directory: ${localPath}`, 400) + } + + let fileCount = 0 + let totalBytes = 0 + const walk = async (currentDir: string) => { + const entries = await fs.promises.readdir(currentDir, { withFileTypes: true }) + for (const entry of entries) { + const fullPath = path.join(currentDir, entry.name) + const entryStats = await fs.promises.stat(fullPath) + if (entryStats.isDirectory()) { + await walk(fullPath) + continue + } + if (!entryStats.isFile()) { + continue + } + fileCount += 1 + totalBytes += entryStats.size + } + } + + await walk(localPath) + return { + fileCount, + totalBytes + } + } + + private async buildUploadPayload(localPath: string, containerPath: string): Promise { + let stats: fs.Stats + try { + stats = await fs.promises.stat(localPath) + } catch { + throw new HttpException(`localPath not found: ${localPath}`, 400) + } + + if (!stats.isDirectory()) { + throw new HttpException(`localPath is not a directory: ${localPath}`, 400) + } + + return this.collectDirectoryFiles(localPath, containerPath) + } + + private async collectDirectoryFiles(baseDir: string, containerPath: string): Promise { + const files: Array<[string, Uint8Array]> = [] + let totalBytes = 0 + + const walk = async (currentDir: string) => { + const entries = await fs.promises.readdir(currentDir, { withFileTypes: true }) + for (const entry of entries) { + const fullPath = path.join(currentDir, entry.name) + const stats = await fs.promises.stat(fullPath) + if (stats.isDirectory()) { + await walk(fullPath) + continue + } + if (!stats.isFile()) { + continue + } + + const relativePath = path.relative(baseDir, fullPath) + const destinationPath = path.posix.join(containerPath, this.toPosixPath(relativePath)) + const content = await fs.promises.readFile(fullPath) + totalBytes += content.byteLength + files.push([destinationPath, new Uint8Array(content)]) + } + } + + await walk(baseDir) + return { + files, + fileCount: files.length, + totalBytes + } + } + + private toPosixPath(value: string) { + return value.split(path.sep).join(path.posix.sep) + } +} diff --git a/packages/server-ai/src/sandbox/commands/index.ts b/packages/server-ai/src/sandbox/commands/index.ts index 5a6f84fe0c..2b6218c8af 100644 --- a/packages/server-ai/src/sandbox/commands/index.ts +++ b/packages/server-ai/src/sandbox/commands/index.ts @@ -2,6 +2,7 @@ export * from './vm.command' export * from './load.command' export * from './acquire-backend.command' export * from './sandbox.copy-file.command' +export * from './sandbox.copy-tree.command' export * from './start-managed-service.command' export * from './list-managed-services.command' export * from './get-managed-service-logs.command' diff --git a/packages/server-ai/src/sandbox/commands/sandbox.copy-tree.command.ts b/packages/server-ai/src/sandbox/commands/sandbox.copy-tree.command.ts new file mode 100644 index 0000000000..195413d99a --- /dev/null +++ b/packages/server-ai/src/sandbox/commands/sandbox.copy-tree.command.ts @@ -0,0 +1,37 @@ +import { Command } from '@nestjs/cqrs' +import type { TSandboxConfigurable } from '@xpert-ai/contracts' + +export type SandboxCopyTreeStatus = 'skipped' | 'success' + +export type SandboxCopyTreeSandboxParams = TSandboxConfigurable | null + +export type SandboxCopyTreeParams = { + version: string | number + localPath: string + containerPath: string + overwrite?: boolean +} + +export type SandboxCopyTreeMode = 'local' | 'upload' + +export class SandboxCopyTreeCommand extends Command<{ + containerId: string + localPath?: string + containerPath: string + status: SandboxCopyTreeStatus + version: string | number + fileCount?: number + totalBytes?: number + scanMs?: number + uploadMs?: number + totalMs?: number + mode?: SandboxCopyTreeMode + reason?: string +}> { + constructor( + public readonly sandbox: SandboxCopyTreeSandboxParams, + public readonly copyTree: SandboxCopyTreeParams + ) { + super() + } +} diff --git a/packages/server-ai/src/skill-package/plugins/skills-middleware/index.spec.ts b/packages/server-ai/src/skill-package/plugins/skills-middleware/index.spec.ts index 23a53be81f..efe89afe95 100644 --- a/packages/server-ai/src/skill-package/plugins/skills-middleware/index.spec.ts +++ b/packages/server-ai/src/skill-package/plugins/skills-middleware/index.spec.ts @@ -22,8 +22,8 @@ jest.mock('../../../sandbox', () => ({ SandboxAcquireBackendCommand: class SandboxAcquireBackendCommand { constructor(public readonly payload: unknown) {} }, - SandboxCopyFileCommand: class SandboxCopyFileCommand { - constructor(public readonly sandbox: unknown, public readonly copyFile: unknown) {} + SandboxCopyTreeCommand: class SandboxCopyTreeCommand { + constructor(public readonly sandbox: unknown, public readonly copyTree: unknown) {} } })) @@ -49,12 +49,7 @@ describe('SkillsMiddleware', () => { } type MiddlewareInternals = { loadSkillMetadata: (workspaceRoot: string, skillIds: string[], workspaceId: string) => Promise - syncSkillsToSandbox: ( - tenantId: string, - sandbox: unknown, - runtimeSkillsRootInContainer: string, - skills: unknown[] - ) => Promise + resolveLocalPackagePath: (workspaceRoot: string, packagePath: string) => Promise } function createMiddleware() { @@ -114,6 +109,16 @@ describe('SkillsMiddleware', () => { return found } + function deferred() { + let resolve!: (value: T | PromiseLike) => void + let reject!: (reason?: unknown) => void + const promise = new Promise((innerResolve, innerReject) => { + resolve = innerResolve + reject = innerReject + }) + return { promise, resolve, reject } + } + it('keeps auto discovery disabled by default and exposes only read_skill_file', async () => { const middleware = createMiddleware() @@ -332,7 +337,9 @@ describe('SkillsMiddleware', () => { version: '1' } ]) - const syncSkillsToSandbox = jest.spyOn(internals, 'syncSkillsToSandbox').mockResolvedValue(undefined) + const execute = jest.fn().mockResolvedValue({ ok: true }) + ;(middleware as any).commandBus = { execute } + jest.spyOn(internals, 'resolveLocalPackagePath').mockResolvedValue('/workspace-root/skills/weather') const instance = await middleware.createMiddleware( { @@ -374,7 +381,7 @@ describe('SkillsMiddleware', () => { 'workspace-1', 'index-calendar' ) - expect(syncSkillsToSandbox).toHaveBeenCalled() + expect(execute).toHaveBeenCalled() expect(installResult.workspaceId).toBe('workspace-1') expect(installResult.installed).toHaveLength(2) @@ -941,10 +948,187 @@ describe('SkillsMiddleware', () => { expect(execute).toHaveBeenCalledWith( expect.objectContaining({ - copyFile: expect.objectContaining({ + copyTree: expect.objectContaining({ containerPath: `${runtimeSkillsRoot}/skill-a` }) }) ) }) + + it('does not block model calls while scheduling skill sync', async () => { + const middleware = createMiddleware() + const copy = deferred<{ ok: true }>() + const execute = jest.fn(() => copy.promise) + ;(middleware as any).commandBus = { execute } + jest.spyOn(middleware as any, 'loadSkillMetadata').mockResolvedValue([ + { + id: 'skill-a', + name: 'skill-a', + description: 'skill-a description', + path: `${runtimeSkillsRoot}/skill-a/SKILL.md`, + packagePath: 'skill-a', + workspaceId: 'workspace-1', + version: '2026-04-17T00:00:00.000Z' + } + ]) + jest.spyOn(middleware as any, 'resolveLocalPackagePath').mockResolvedValue('/workspace-root/skills/skill-a') + + const instance = await middleware.createMiddleware( + { + skills: ['skill-a'] + }, + createContext() + ) + + const handler = jest.fn(async (request) => request) + await instance.wrapModelCall( + { + runtime: { + configurable: { + sandbox: { + workingDirectory: runtimeWorkingDirectory + } + } + }, + state: {}, + systemMessage: new SystemMessage('base') + } as any, + handler + ) + + expect(execute).toHaveBeenCalledTimes(1) + expect(handler).toHaveBeenCalledTimes(1) + copy.resolve({ ok: true }) + await Promise.resolve() + }) + + it('reuses an inflight skill sync for duplicate schedules', async () => { + const middleware = createMiddleware() + const copy = deferred<{ ok: true }>() + const execute = jest.fn(() => copy.promise) + ;(middleware as any).commandBus = { execute } + const skillPackageService = Reflect.get(middleware, 'skillPackageService') as { + ensureInstalledSkillPackage: jest.Mock + } + skillPackageService.ensureInstalledSkillPackage.mockResolvedValue({ + id: 'skill-a' + }) + jest.spyOn(middleware as any, 'loadSkillMetadata').mockResolvedValue([ + { + id: 'skill-a', + name: 'skill-a', + description: 'skill-a description', + path: `${runtimeSkillsRoot}/skill-a/SKILL.md`, + packagePath: 'skill-a', + workspaceId: 'workspace-1', + version: '2026-04-17T00:00:00.000Z' + } + ]) + jest.spyOn(middleware as any, 'resolveLocalPackagePath').mockResolvedValue('/workspace-root/skills/skill-a') + + const instance = await middleware.createMiddleware( + { + autoDiscovery: { + enabled: true + }, + skills: ['skill-a'] + }, + createContext() + ) + const handler = jest.fn(async (request) => request) + const request = { + runtime: { + configurable: { + sandbox: { + backend: { + id: 'sandbox-1' + }, + workingDirectory: runtimeWorkingDirectory + } + } + }, + state: {}, + systemMessage: new SystemMessage('base') + } as any + + await instance.wrapModelCall(request, handler) + const installTool = getTool(instance, 'install_workspace_skills') + await installTool.invoke({ + indexIds: ['index-skill-a'] + }) + + expect(execute).toHaveBeenCalledTimes(1) + copy.resolve({ ok: true }) + await Promise.resolve() + }) + + it('waits for an inflight skill sync before reading a synced skill file', async () => { + const middleware = createMiddleware() + const copy = deferred<{ ok: true }>() + const execute = jest.fn(() => copy.promise) + const readExecute = jest.fn().mockResolvedValue({ + exitCode: 0, + output: 'skill instructions' + }) + ;(middleware as any).commandBus = { execute } + jest.spyOn(middleware as any, 'loadSkillMetadata').mockResolvedValue([ + { + id: 'skill-a', + name: 'skill-a', + description: 'skill-a description', + path: `${runtimeSkillsRoot}/skill-a/SKILL.md`, + packagePath: 'skill-a', + workspaceId: 'workspace-1', + version: '2026-04-17T00:00:00.000Z' + } + ]) + jest.spyOn(middleware as any, 'resolveLocalPackagePath').mockResolvedValue('/workspace-root/skills/skill-a') + + const instance = await middleware.createMiddleware( + { + skills: ['skill-a'] + }, + createContext() + ) + const handler = jest.fn(async (request) => request) + await instance.wrapModelCall( + { + runtime: { + configurable: { + sandbox: { + backend: { + id: 'sandbox-1', + execute: readExecute + }, + workingDirectory: runtimeWorkingDirectory + } + } + }, + state: {}, + systemMessage: new SystemMessage('base') + } as any, + handler + ) + + const readTool = getTool(instance, 'read_skill_file') + let readResolved = false + const readPromise = readTool + .invoke({ + path: `${runtimeSkillsRoot}/skill-a/SKILL.md` + }) + .then((value) => { + readResolved = true + return value + }) + + await Promise.resolve() + expect(readResolved).toBe(false) + expect(execute).toHaveBeenCalledTimes(1) + expect(readExecute).not.toHaveBeenCalled() + + copy.resolve({ ok: true }) + await expect(readPromise).resolves.toBe('skill instructions') + expect(execute).toHaveBeenCalledTimes(1) + expect(readExecute).toHaveBeenCalledTimes(1) + }) }) diff --git a/packages/server-ai/src/skill-package/plugins/skills-middleware/index.ts b/packages/server-ai/src/skill-package/plugins/skills-middleware/index.ts index 31b90a8585..3946a7503f 100644 --- a/packages/server-ai/src/skill-package/plugins/skills-middleware/index.ts +++ b/packages/server-ai/src/skill-package/plugins/skills-middleware/index.ts @@ -43,7 +43,7 @@ import { z } from 'zod/v3' import { SkillPackage } from '../../skill-package.entity' import { SkillPackageService } from '../../skill-package.service' import { SKILLS_MIDDLEWARE_NAME } from '../../types' -import { SandboxAcquireBackendCommand, SandboxCopyFileCommand } from '../../../sandbox' +import { SandboxAcquireBackendCommand, SandboxCopyTreeCommand } from '../../../sandbox' import { SkillRepositoryIndexService } from '../../../skill-repository/repository-index/skill-repository-index.service' import { getWorkspaceRoot } from '../../../xpert-workspace' import { XpertWorkspace } from '../../../xpert-workspace/workspace.entity' @@ -118,6 +118,14 @@ type SkillPromptMetadata = { version: string } +type SkillSyncRecord = { + containerPath: string + error?: Error + key: string + promise: Promise + skill: SkillPromptMetadata +} + type SkillPackageInstallMetadata = Partial & { skillMdPath?: string skillPath?: string @@ -639,6 +647,134 @@ export class SkillsMiddleware implements IAgentMiddlewareStrategy>() + const inflightSyncs = new Map() + const skillSyncsByContainerPath = new Map() + + const buildSkillSyncKey = (sandbox: unknown, containerPath: string, version: string) => { + const backendId = resolveSandboxBackend(sandbox)?.id ?? 'unknown' + return `${backendId}:${containerPath}:${version}` + } + + const clearSkillSyncRecord = (record: SkillSyncRecord) => { + if (inflightSyncs.get(record.key) === record) { + inflightSyncs.delete(record.key) + } + if (skillSyncsByContainerPath.get(record.containerPath) === record) { + skillSyncsByContainerPath.delete(record.containerPath) + } + } + + const syncSkillToSandbox = async ( + sandbox: unknown, + runtimeSkillsRoot: string, + skill: SkillPromptMetadata + ): Promise => { + const packagePath = skill.packagePath?.trim() + if (!packagePath) { + return + } + + const workspaceRoot = getWorkspaceRoot(tenantId, skill.workspaceId) + const localSkillPath = await this.resolveLocalPackagePath(workspaceRoot, packagePath) + if (!localSkillPath) { + this.#logger.warn( + `Skip syncing skill "${skill.name}" because package path "${packagePath}" is missing in workspace "${skill.workspaceId}".` + ) + return + } + await this.commandBus.execute( + new SandboxCopyTreeCommand(sandbox, { + version: skill.version, + localPath: localSkillPath, + containerPath: join(runtimeSkillsRoot, packagePath), + overwrite: true + }) + ) + } + + const scheduleSkillSync = ( + sandbox: unknown, + runtimeSkillsRoot: string, + skill: SkillPromptMetadata + ): SkillSyncRecord | null => { + const packagePath = skill.packagePath?.trim() + if (!packagePath) { + return null + } + + const containerPath = join(runtimeSkillsRoot, packagePath) + const key = buildSkillSyncKey(sandbox, containerPath, String(skill.version)) + const existing = inflightSyncs.get(key) + if (existing) { + skillSyncsByContainerPath.set(containerPath, existing) + this.#logger.log(`Reuse inflight skill sync skill=${skill.name} containerPath=${containerPath} key=${key}`) + return existing + } + + const start = Date.now() + let record: SkillSyncRecord + const promise = syncSkillToSandbox(sandbox, runtimeSkillsRoot, skill) + .then(() => { + this.#logger.log( + `Finished async skill sync skill=${skill.name} containerPath=${containerPath} key=${key} totalMs=${Date.now() - start}` + ) + }) + .catch((error) => { + record.error = error as Error + this.#logger.error( + `Failed async skill sync skill=${skill.name} containerPath=${containerPath} key=${key}: ${(error as Error).message}` + ) + }) + .finally(() => clearSkillSyncRecord(record)) + record = { + containerPath, + key, + promise, + skill + } + + inflightSyncs.set(key, record) + skillSyncsByContainerPath.set(containerPath, record) + this.#logger.log(`Scheduled skill sync skill=${skill.name} containerPath=${containerPath} key=${key}`) + return record + } + + const scheduleSkillsToSandbox = ( + sandbox: unknown, + runtimeSkillsRoot: string, + skills: SkillPromptMetadata[] + ) => { + for (const skill of skills) { + scheduleSkillSync(sandbox, runtimeSkillsRoot, skill) + } + } + + const findSkillSyncForPath = (fullPath: string): SkillSyncRecord | null => { + let matched: SkillSyncRecord | null = null + for (const record of skillSyncsByContainerPath.values()) { + const relativePath = relative(record.containerPath, fullPath) + const isInside = relativePath === '' || (!relativePath.startsWith('..') && !isAbsolute(relativePath)) + if (!isInside) { + continue + } + if (!matched || record.containerPath.length > matched.containerPath.length) { + matched = record + } + } + return matched + } + + const waitForSkillSyncIfNeeded = async (fullPath: string) => { + const record = findSkillSyncForPath(fullPath) + if (!record) { + return + } + this.#logger.log(`Waiting for skill sync before reading file skill=${record.skill.name} path=${fullPath}`) + await record.promise + if (record.error) { + throw record.error + } + } /** * Read skill's file tool @@ -652,6 +788,8 @@ export class SkillsMiddleware implements IAgentMiddlewareStrategy { try { await access(path)