From 95ebbd3cca39948c42e92d6c8228ace6a2f44298 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sat, 19 Sep 2026 21:42:37 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=8C=B3=20feat:=20Provision=20Conversation?= =?UTF-8?q?-Scoped=20Code=20Worktrees=20(#239)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: provision conversation-scoped code worktrees * fix: isolate conversation checkout metadata * docs: clarify isolated conversation checkouts * fix: revalidate conversation checkout sources * fix: preserve synchronous legacy execution startup * fix: harden conversation worktree lifecycle * test: use canonical workspace isolation keys * fix: secure conversation worktree provisioning * fix: preserve isolated workspace lifecycle * fix: harden conversation worktree provisioning * fix: fence worktree setup and credential routing * fix: use kernel-backed provisioning locks * fix: load worktree locking only when provisioned * fix: retain conversation provisioning ownership through recovery * fix: reserve provisioning before launching checkout writers * fix: pin Git provisioning inputs and close instance admission gaps --- packages/code/README.md | 59 +- packages/code/src/cli.ts | 190 +++- packages/code/src/git-snapshot.ts | 343 ++++++ packages/code/src/native-pool.test.ts | 120 ++- packages/code/src/native-pool.ts | 106 +- packages/code/src/native-process.test.ts | 1 + packages/code/src/native-sandbox.test.ts | 31 + packages/code/src/process-lock.test.ts | 61 ++ packages/code/src/process-lock.ts | 59 ++ packages/code/src/protocol.test.ts | 42 +- packages/code/src/protocol.ts | 42 +- packages/code/src/root-identity.ts | 16 + packages/code/src/worker-slots.test.ts | 35 +- packages/code/src/worker.ts | 163 ++- packages/code/src/workspace-cli.test.ts | 32 + packages/code/src/workspace-instances.test.ts | 234 +++++ packages/code/src/workspace-instances.ts | 232 ++++ packages/code/src/workspace-worker.test.ts | 100 ++ packages/code/src/worktrees.test.ts | 988 ++++++++++++++++++ packages/code/src/worktrees.ts | 718 +++++++++++++ service/src/bridge/concurrent-store.test.ts | 69 +- service/src/bridge/router.ts | 1 + service/src/bridge/store.test.ts | 12 +- service/src/bridge/store.ts | 55 +- service/src/bridge/workspace-instance.test.ts | 40 + service/src/bridge/workspace-instance.ts | 17 + service/src/service/programmatic-router.ts | 34 +- .../src/service/programmatic-state.test.ts | 16 + service/src/service/programmatic-state.ts | 12 + service/src/service/replay-state.ts | 2 + service/src/types/service.ts | 4 + service/src/workspace-tools/router.test.ts | 31 + service/src/workspace-tools/router.ts | 19 +- 33 files changed, 3803 insertions(+), 81 deletions(-) create mode 100644 packages/code/src/git-snapshot.ts create mode 100644 packages/code/src/process-lock.test.ts create mode 100644 packages/code/src/process-lock.ts create mode 100644 packages/code/src/workspace-instances.test.ts create mode 100644 packages/code/src/workspace-instances.ts create mode 100644 packages/code/src/worktrees.test.ts create mode 100644 packages/code/src/worktrees.ts create mode 100644 service/src/bridge/workspace-instance.test.ts create mode 100644 service/src/bridge/workspace-instance.ts diff --git a/packages/code/README.md b/packages/code/README.md index 69d85b16..e30b80cb 100644 --- a/packages/code/README.md +++ b/packages/code/README.md @@ -706,8 +706,63 @@ Slots are per machine, not a fleet-wide execution limit. A busy machine does not consume another machine's slots. Requests for the same root remain serialized, including commands started through background tools. Independent checkouts can use different slots; selecting subdirectories beneath one registered parent root -does not create separate scheduling boundaries. Linked Git worktrees share Git -metadata and are not supported by selected-project registration. +does not create separate scheduling boundaries. + +To bind each conversation to an isolated checkout of the selected Git +repository, configure worker-owned conversation worktrees: + +```sh +librechat-code run \ + --worker-dir /projects/LibreChat \ + --workspace-lease-slots 4 \ + --conversation-worktree-root /var/lib/librechat-code/worktrees \ + --conversation-worktree-max 64 \ + --conversation-worktree-clone-timeout-ms 300000 \ + --allow-workspace-writes \ + --allow-workspace-commands +``` + +`LIBRECHAT_CODE_CONVERSATION_WORKTREE_ROOT` and +`LIBRECHAT_CODE_CONVERSATION_WORKTREE_MAX` are the environment equivalents; +`LIBRECHAT_CODE_CONVERSATION_WORKTREE_CLONE_TIMEOUT_MS` controls the bounded +clone budget (five minutes by default, from 30 seconds through 30 minutes). +The storage root must be owner-controlled, must not overlap a registered +workspace, and every registered source must be a Git repository. The worker +creates a deterministic branch in an isolated local checkout for the opaque +conversation identity supplied by LibreChat. Each checkout owns its writable +Git metadata and object storage, without alternates or hardlinks to the source. +Provisioning pins the source Git-directory and object-store identities. It copies +Git data through no-follow, descriptor-relative reads into private staging before +running Git; source hooks and config includes are not used. The clone budget +also bounds this snapshot. Local hardlinks only connect private staging to its +new checkout, never to the source; staging is removed before setup. Source +alternates admitted at worker startup are materialized into independent objects. +Git metadata replacement requires operator recovery, not automatic re-admission. +Host paths remain private. The configured count +is a hard per-machine quota, provisioning is serialized, and operations for one +conversation remain serialized while different conversations may occupy +different lease slots. Recognizable abandoned checkouts without a lifecycle +record are discarded before admission. New provisioning reserves its record +before starting Git or setup; a worker crash leaves that checkout reserved for +operator recovery because child processes might still be running. +Reservations count even when a crash happens before a checkout directory exists. + +Cancellation also covers waiting for the provisioning lock, cloning, and setup. +The worker waits for setup cleanup before releasing the assignment. If cleanup +cannot be confirmed, the checkout stays reserved and fails closed on restart. +A completed checkout with a changed source identity or invalid completion record +is preserved for operator recovery, including any uncommitted work. After stopping +the worker and confirming no executor still uses the checkout, an operator can +archive the affected checkout and its adjacent `.complete` record before retrying. +Also archive any adjacent `.source` staging directory. Pre-release version-1 +completion records are deliberately preserved but not admitted by this version; +they do not contain the required source Git identity binding. + +GitHub App routing is inherited from the operator-admitted source repository; +commands cannot select a different installation by rewriting a worktree remote. +Legacy requests without a conversation identity continue to use the selected +source root. Older Code API deployments do not negotiate the capability, so the +worker omits it until every request path understands the isolation boundary. Admission waits at most 30 seconds. A `WORKSPACE_QUEUE_TIMEOUT` response (HTTP 503, `Retry-After: 1`) means the operation was not assigned or started; wait for diff --git a/packages/code/src/cli.ts b/packages/code/src/cli.ts index 96757d57..26c60e9a 100644 --- a/packages/code/src/cli.ts +++ b/packages/code/src/cli.ts @@ -36,6 +36,9 @@ import { import { RuntimeWorkspaceCommandSandbox } from './workspace-runtime.js'; import { NativeProcessWorkspaceCommandSandbox } from './native-process.js'; import { NativeWorkspaceCommandPool } from './native-pool.js'; +import { GitWorktreeWorkspaceTools, internalWorkspaceId } from './workspace-instances.js'; +import { GitWorktreeManager } from './worktrees.js'; +import { captureWorkspaceRootIdentity } from './root-identity.js'; import { resolveNativeSrtCommandPolicy, serializeNativeSrtCommandPolicy, @@ -61,8 +64,9 @@ import type { WorkspaceToolExecutor } from './workspace.js'; import { BRIDGE_WORKSPACE_NAME_MAX_LENGTH, BridgeProtocolError, - isValidBridgeWorkerCapabilities, - isValidBridgeWorkerId, + isValidBridgeWorkerCapabilities, + isValidBridgeWorkerId, + workspaceIsolationKey, } from './protocol.js'; function workspaceSecurityIdentity( @@ -600,6 +604,32 @@ async function run( ); if (workspaceLeaseSlots > 8) throw new Error('Workspace lease slots cannot exceed 8'); + const conversationWorktreeRoot = + option(args, '--conversation-worktree-root') ?? + process.env.LIBRECHAT_CODE_CONVERSATION_WORKTREE_ROOT?.trim(); + const conversationWorktreeMax = positiveInteger( + 'LIBRECHAT_CODE_CONVERSATION_WORKTREE_MAX', + option(args, '--conversation-worktree-max') ?? + process.env.LIBRECHAT_CODE_CONVERSATION_WORKTREE_MAX, + 64, + ); + if (conversationWorktreeMax > 1024) { + throw new Error('Conversation worktree capacity cannot exceed 1024'); + } + const conversationWorktreeCloneTimeoutMs = positiveInteger( + 'LIBRECHAT_CODE_CONVERSATION_WORKTREE_CLONE_TIMEOUT_MS', + option(args, '--conversation-worktree-clone-timeout-ms') ?? + process.env.LIBRECHAT_CODE_CONVERSATION_WORKTREE_CLONE_TIMEOUT_MS, + 5 * 60_000, + ); + if ( + conversationWorktreeCloneTimeoutMs < 30_000 || + conversationWorktreeCloneTimeoutMs > 30 * 60_000 + ) { + throw new Error( + 'Conversation worktree clone timeout must be between 30000 and 1800000 milliseconds', + ); + } const roots: LocalWorkspaceConfig[] = canonicalWorkerDirectory ? [ { @@ -701,6 +731,16 @@ async function run( 'Concurrent workspace leases require native-srt commands', ); } + if ( + conversationWorktreeRoot && + (!allowWorkspaceCommands || + commandSandboxMode !== 'native-srt' || + workspaceLeaseSlots < 2) + ) { + throw new Error( + 'Conversation worktrees require native-srt commands and at least two workspace lease slots', + ); + } if ( roots.length > 1 && process.env.LIBRECHAT_CODE_WORKSPACE_QUARANTINE_FILE?.trim() @@ -732,6 +772,14 @@ async function run( ), ) : undefined; + const repositoriesByWorkspace = admittedGitHubRepositories + ? new Map( + roots.map((root) => [ + root.id, + admittedGitHubRepositories.get(root.root), + ]), + ) + : undefined; const localWorkspaceTools = workerDirectory ? await LocalWorkspaceTools.create({ workspaces: roots, @@ -981,7 +1029,7 @@ async function run( }; const nativeCommandSandbox = allowWorkspaceCommands && commandSandboxMode === 'native-srt' - ? roots.length > 1 || workspaceLeaseSlots > 1 + ? roots.length > 1 || workspaceLeaseSlots > 1 || conversationWorktreeRoot ? new NativeWorkspaceCommandPool( new Map( roots.map(root => [ @@ -1008,6 +1056,105 @@ async function run( incarnationId, }), }); + } + const conversationWorktrees = conversationWorktreeRoot + ? new GitWorktreeManager({ + cloneTimeoutMs: conversationWorktreeCloneTimeoutMs, + maxCount: conversationWorktreeMax, + root: conversationWorktreeRoot, + ...(nativeCommandSandbox instanceof NativeWorkspaceCommandPool + ? { + prepareInstance: async (instance, signal) => { + const setup = environments.find( + (environment) => + environment.definition.name === instance.sourceWorkspaceId, + )?.definition.setup; + if (!setup) return; + if (admittedGitHubRepositories) { + admittedGitHubRepositories.set( + instance.root, + repositoriesByWorkspace?.get(instance.sourceWorkspaceId), + ); + } + const id = internalWorkspaceId(instance.sourceWorkspaceId, instance.id); + await nativeCommandSandbox.registerRoot(id, { + ...nativeOptions, + workspaceIdentity: instance.identity, + workspaceRoot: instance.root, + }); + const result = await nativeCommandSandbox.execute( + { + protocolVersion: 1, + operation: 'execute_command', + workspaceId: id, + command: setup.command, + timeoutMs: setup.timeoutMs, + maxOutputBytes: 8192, + }, + signal, + ); + if (result.exitCode !== 0 || result.timedOut) { + throw new Error( + `Environment ${instance.sourceWorkspaceId} setup failed for its conversation worktree`, + ); + } + }, + discardInstance: async (instance) => { + await nativeCommandSandbox.unregisterRoot( + internalWorkspaceId(instance.sourceWorkspaceId, instance.id), + ); + admittedGitHubRepositories?.delete(instance.root); + }, + } + : {}), + sources: new Map( + await Promise.all( + roots.map(async (root) => [ + root.id, + { + root: root.root, + identity: + root.identity ?? + (await captureWorkspaceRootIdentity(root.root)), + }, + ] as const), + ), + ), + }) + : undefined; + let conversationWorkspaceTools: GitWorktreeWorkspaceTools | undefined; + if (conversationWorktrees && workspaceTools) { + if (!(nativeCommandSandbox instanceof NativeWorkspaceCommandPool)) { + throw new Error('Conversation worktrees require a native command pool'); + } + conversationWorkspaceTools = new GitWorktreeWorkspaceTools({ + commandPool: nativeCommandSandbox, + delegate: workspaceTools, + manager: conversationWorktrees, + onResolve(workspaceId, root) { + if (admittedGitHubRepositories) { + admittedGitHubRepositories.set( + root, + repositoriesByWorkspace?.get(workspaceId), + ); + } + }, + sources: new Map( + roots.map((root) => [ + root.id, + { + command: { + ...nativeOptions, + workspaceIdentity: root.identity, + workspaceRoot: root.root, + }, + repositoryInstructions: args.includes('--repository-instructions'), + writable: root.writable ?? false, + }, + ]), + ), + }); + workspaceTools = conversationWorkspaceTools; } if (workspaceTools && environments.length) { workspaceTools = new EnvironmentWorkspaceTools( @@ -1059,6 +1206,7 @@ async function run( if (github.provider && !github.provider.validate) { await github.provider.getCredential(controller.signal); } + await conversationWorktrees?.prepare(); await nativeCommandSandbox?.prepare(); for (const environment of option(args, '--reset-workspace-quarantine') == null ? environments : []) { const setup = environment.definition.setup; @@ -1110,7 +1258,34 @@ async function run( capabilities, workspaceTools, ...(nativeProgrammaticEnabled && nativeCommandSandbox - ? { workspaceProgrammatic: nativeCommandSandbox } + ? { + workspaceProgrammatic: + conversationWorkspaceTools ?? nativeCommandSandbox, + } + : {}), + ...(conversationWorktrees + ? { + workspaceQuarantineResolver: async ( + selectedWorkspaceId: string, + workspaceInstanceId: string, + ) => + workspaceMutationGuard( + defaultWorkspaceQuarantinePath({ + codeApiUrl, + workerId, + workspaceRoot: await conversationWorktrees.plannedRoot( + selectedWorkspaceId, + workspaceInstanceId, + ), + }), + workerId, + workspaceIsolationKey( + selectedWorkspaceId, + workspaceInstanceId, + ), + incarnationId, + ), + } : {}), ...(workspaceLeaseSlots > 1 || roots.length > 1 ? { @@ -1222,14 +1397,19 @@ async function run( } const resetNativeRoot = option(args, '--reset-workspace-quarantine'); if (resetNativeRoot != null) { + const resetWorkspaceInstance = option( + args, + '--reset-workspace-instance', + ); await worker.refreshCredential(controller.signal); await worker.registerForMaintenance(controller.signal); await worker.resetNativeWorkspace( resetNativeRoot, controller.signal, + resetWorkspaceInstance, ); process.stdout.write( - `librechat-code: reset acknowledged for native workspace ${resetNativeRoot}\n`, + `librechat-code: reset acknowledged for native workspace ${resetNativeRoot}${resetWorkspaceInstance ? ` instance ${resetWorkspaceInstance}` : ''}\n`, ); return; } diff --git a/packages/code/src/git-snapshot.ts b/packages/code/src/git-snapshot.ts new file mode 100644 index 00000000..7c57beda --- /dev/null +++ b/packages/code/src/git-snapshot.ts @@ -0,0 +1,343 @@ +import { constants, close, fstat, read } from 'node:fs'; +import { mkdir, open, opendir, writeFile } from 'node:fs/promises'; +import { createHash } from 'node:crypto'; +import { join, resolve } from 'node:path'; +import { promisify } from 'node:util'; +import { captureWorkspaceRootIdentity } from './root-identity.js'; +import type { WorkspaceRootIdentity } from './root-identity.js'; + +const closeFd = promisify(close); +const statFd = promisify(fstat); +const readFd = promisify(read); +let binding: + | Promise<{ + openat: (fd: number, name: string, flags: number) => number; + errno: () => number; + }> + | undefined; + +async function childFd( + parent: number, + name: string +): Promise { + if (!name || name === '.' || name === '..' || name.includes('/')) + throw new Error('Invalid Git metadata entry'); + binding ??= import('koffi').then(({ default: koffi }) => ({ + openat: koffi + .load(null) + .func('int openat(int dirfd, const char *path, int flags)'), + errno: () => koffi.errno(), + })); + const native = await binding; + // Node does not expose O_CLOEXEC. Set it atomically with openat so unrelated + // concurrent executor spawns cannot inherit privileged source descriptors. + if (process.platform !== 'darwin' && process.platform !== 'linux') + throw new Error('Git snapshots require a POSIX host'); + const closeOnExec = process.platform === 'darwin' ? 0x1000000 : 0x80000; + const fd = native.openat( + parent, + name, + constants.O_RDONLY | + constants.O_NOFOLLOW | + constants.O_NONBLOCK | + closeOnExec + ); + if (fd >= 0) return fd; + if (native.errno() === 2) return undefined; // ENOENT on supported POSIX hosts + throw new Error('Git metadata entry is unavailable or is a symbolic link'); +} + +async function withDirectory( + identity: WorkspaceRootIdentity, + operation: (fd: number) => Promise +): Promise { + const handle = await open( + identity.path, + constants.O_RDONLY | constants.O_DIRECTORY | constants.O_NOFOLLOW + ); + try { + const current = await handle.stat({ bigint: true }); + if ( + current.dev.toString() !== identity.dev || + current.ino.toString() !== identity.ino + ) { + throw new Error('Source Git metadata changed after admission'); + } + return await operation(handle.fd); + } finally { + await handle.close(); + } +} + +async function textAt( + parent: number, + name: string, + limit = 16 * 1024 +): Promise { + const fd = await childFd(parent, name); + if (fd == null) return undefined; + try { + const metadata = await statFd(fd); + if (!metadata.isFile() || metadata.size > limit) + throw new Error('Invalid Git metadata file'); + const buffer = Buffer.alloc(metadata.size); + let offset = 0; + while (offset < buffer.length) { + const { bytesRead } = await readFd( + fd, + buffer, + offset, + buffer.length - offset, + offset + ); + if (!bytesRead) throw new Error('Git metadata changed during snapshot'); + offset += bytesRead; + } + return buffer.toString('utf8'); + } finally { + await closeFd(fd); + } +} + +async function alternatesAt(parent: number): Promise { + const fd = await childFd(parent, 'info'); + if (fd == null) return undefined; + try { + if (!(await statFd(fd)).isDirectory()) + throw new Error('Invalid Git objects info directory'); + return await textAt(fd, 'alternates'); + } finally { + await closeFd(fd); + } +} + +/** Copies only regular files/directories through descriptor-relative, no-follow opens. + * Renaming a parent or replacing a child with a symlink never expands the read grant. + * No source config, hooks, object alternates, or executable helpers reach Git. + */ +async function copyEntry( + parent: number, + name: string, + destination: string, + signal: AbortSignal | undefined, + depth = 0 +): Promise { + signal?.throwIfAborted(); + if (depth > 64) + throw new Error('Git metadata nesting exceeds snapshot limit'); + const fd = await childFd(parent, name); + if (fd == null) return; + try { + const metadata = await statFd(fd); + if (metadata.isDirectory()) { + await mkdir(destination, { recursive: true, mode: 0o700 }); + const directory = await opendir(`/dev/fd/${fd}`); + for await (const entry of directory) { + await copyEntry( + fd, + entry.name, + join(destination, entry.name), + signal, + depth + 1 + ); + } + } else if (metadata.isFile()) { + const target = await open(destination, 'w', 0o600); + try { + const buffer = Buffer.alloc(128 * 1024); + let offset = 0; + while (offset < metadata.size) { + signal?.throwIfAborted(); + const { bytesRead } = await readFd( + fd, + buffer, + 0, + Math.min(buffer.length, metadata.size - offset), + offset + ); + if (!bytesRead) + throw new Error('Git metadata changed during snapshot'); + await target.writeFile(buffer.subarray(0, bytesRead)); + offset += bytesRead; + } + } finally { + await target.close(); + } + } else { + throw new Error('Git snapshot requires regular files and directories'); + } + } finally { + await closeFd(fd); + } +} + +export class GitSourceSnapshot { + private constructor( + private readonly source: WorkspaceRootIdentity, + private readonly gitDirectory: WorkspaceRootIdentity, + private readonly common: WorkspaceRootIdentity, + private readonly gitfile: string | undefined, + private readonly commondir: string | undefined, + private readonly objects: Array<{ + identity: WorkspaceRootIdentity; + alternates: string | undefined; + }> + ) {} + + get fingerprint(): string { + return createHash('sha256') + .update( + JSON.stringify([ + this.gitDirectory, + this.common, + this.gitfile, + this.commondir, + this.objects, + ]) + ) + .digest('hex'); + } + + static async admit( + source: WorkspaceRootIdentity + ): Promise { + let gitfile: string | undefined; + await withDirectory(source, async (fd) => { + const git = await childFd(fd, '.git'); + if (git == null) + throw new Error('Source workspace is not a Git repository'); + try { + if (!(await statFd(git)).isDirectory()) + gitfile = await textAt(fd, '.git'); + } finally { + await closeFd(git); + } + }); + if (gitfile != null && !/^gitdir: .+\n?$/.test(gitfile)) + throw new Error('Invalid source Git directory pointer'); + const gitDirectory = await captureWorkspaceRootIdentity( + gitfile == null + ? join(source.path, '.git') + : resolve(source.path, gitfile.slice(8).trim()) + ); + const commondir = await withDirectory(gitDirectory, (fd) => + textAt(fd, 'commondir') + ); + const common = + commondir == null + ? gitDirectory + : await captureWorkspaceRootIdentity( + resolve(gitDirectory.path, commondir.trim()) + ); + const objects: Array<{ + identity: WorkspaceRootIdentity; + alternates: string | undefined; + }> = []; + const visit = async (path: string): Promise => { + const identity = await captureWorkspaceRootIdentity(path); + if (objects.some((entry) => entry.identity.path === identity.path)) + return; + if (objects.length >= 32) + throw new Error('Too many source Git object stores'); + const alternates = await withDirectory(identity, alternatesAt); + objects.push({ identity, alternates }); + for (const alternate of alternates?.split('\n').filter(Boolean) ?? []) { + if (alternate.startsWith('"')) + throw new Error('Quoted Git alternate paths are unsupported'); + await visit(resolve(identity.path, alternate)); + } + }; + await visit(join(common.path, 'objects')); + const admitted = new GitSourceSnapshot( + source, + gitDirectory, + common, + gitfile, + commondir, + objects + ); + await admitted.validate(); + return admitted; + } + + async validate(): Promise { + await withDirectory(this.source, async (fd) => { + if (this.gitfile != null) { + if ((await textAt(fd, '.git')) !== this.gitfile) + throw new Error('Source Git metadata changed after admission'); + } else { + // Compare the directory reached from the admitted root, not just its name. + const child = await childFd(fd, '.git'); + if (child == null) + throw new Error('Source Git metadata changed after admission'); + try { + const current = await promisify(fstat)(child, { + bigint: true, + }); + if ( + current.dev.toString() !== this.gitDirectory.dev || + current.ino.toString() !== this.gitDirectory.ino + ) { + throw new Error('Source Git metadata changed after admission'); + } + } finally { + await closeFd(child); + } + } + }); + await withDirectory(this.gitDirectory, async (fd) => { + if ((await textAt(fd, 'commondir')) !== this.commondir) + throw new Error('Source Git metadata changed after admission'); + }); + await withDirectory(this.common, async () => {}); + for (const entry of this.objects) { + await withDirectory(entry.identity, async (fd) => { + if ((await alternatesAt(fd)) !== entry.alternates) + throw new Error('Source Git alternates changed after admission'); + }); + } + } + + async copyTo(destination: string, signal?: AbortSignal): Promise { + await this.validate(); + await mkdir(destination, { mode: 0o700 }); + await mkdir(join(destination, 'objects'), { mode: 0o700 }); + await mkdir(join(destination, 'refs'), { mode: 0o700 }); + await withDirectory(this.gitDirectory, (fd) => + copyEntry(fd, 'HEAD', join(destination, 'HEAD'), signal) + ); + await withDirectory(this.common, async (fd) => { + for (const name of ['refs', 'packed-refs', 'shallow']) + await copyEntry(fd, name, join(destination, name), signal); + // Kept outside Git's config name; caller may query origin with --no-includes. + const config = await textAt(fd, 'config', 1024 * 1024); + if (config != null) + await writeFile(join(destination, 'source-config'), config, { + mode: 0o600, + }); + }); + for (const entry of [...this.objects].reverse()) { + await withDirectory(entry.identity, async (fd) => { + const directory = await opendir(`/dev/fd/${fd}`); + for await (const child of directory) { + // Object info (notably alternates) never crosses into private staging. + if (child.name === 'pack' || /^[a-f0-9]{2}$/.test(child.name)) { + await copyEntry( + fd, + child.name, + join(destination, 'objects', child.name), + signal + ); + } + } + }); + } + await writeFile( + join(destination, 'config'), + '[core]\nrepositoryformatversion = 0\nbare = true\n', + { mode: 0o600 } + ); + await this.validate(); + signal?.throwIfAborted(); + } +} diff --git a/packages/code/src/native-pool.test.ts b/packages/code/src/native-pool.test.ts index cad30a52..ce25caa4 100644 --- a/packages/code/src/native-pool.test.ts +++ b/packages/code/src/native-pool.test.ts @@ -5,7 +5,7 @@ import { WorkspaceToolError } from './workspace.js'; import type { WorkspaceExecuteCommandRequest } from './protocol.js'; const roots = new Map( - ['a', 'b', 'c'].map((id) => [id, { workspaceRoot: `/fixture/${id}` }]), + ['a', 'b', 'c'].map((id) => [id, { workspaceRoot: `/fixture/${id}` }]) ); const request = (workspaceId: string): WorkspaceExecuteCommandRequest => ({ protocolVersion: 1, @@ -14,6 +14,36 @@ const request = (workspaceId: string): WorkspaceExecuteCommandRequest => ({ command: 'fixture', }); +test('failed provisioning roots are removed only after confirmed executor cleanup', async () => { + let failClose = true; + const pool = new NativeWorkspaceCommandPool(roots, 2, () => ({ + async prepare() {}, + async execute() { + throw new Error('not used'); + }, + async close() { + if (failClose) throw new Error('cleanup unconfirmed'); + }, + })); + await pool.registerRoot('instance', { workspaceRoot: '/fixture/instance' }); + // Allocate this executor without dispatching a command. + const controller = new AbortController(); + controller.abort(); + await assert.rejects( + pool.execute(request('instance'), controller.signal), + /cancelled/ + ); + await assert.rejects(pool.unregisterRoot('instance'), /cleanup unconfirmed/); + failClose = false; + await pool.unregisterRoot('instance'); + await assert.rejects(pool.execute(request('instance')), /unavailable/); + await pool.close(); + await assert.rejects( + pool.registerRoot('new', { workspaceRoot: '/fixture/new' }), + /unavailable/ + ); +}); + test('native pool preflights every registered root with bounded concurrency', async () => { const prepared: string[] = []; let active = 0; @@ -22,7 +52,7 @@ test('native pool preflights every registered root with bounded concurrency', as async prepare() { active += 1; peak = Math.max(peak, active); - await new Promise(resolve => setTimeout(resolve, 5)); + await new Promise((resolve) => setTimeout(resolve, 5)); prepared.push(options.workspaceRoot); active -= 1; }, @@ -38,6 +68,90 @@ test('native pool preflights every registered root with bounded concurrency', as await pool.close(); }); +test('native pool admits worker-owned roots after startup', async () => { + const created: string[] = []; + const pool = new NativeWorkspaceCommandPool( + new Map([['primary', { workspaceRoot: '/fixture/primary' }]]), + 2, + (options) => ({ + async prepare() {}, + async close() {}, + async execute(req) { + created.push(options.workspaceRoot); + return { + protocolVersion: 1, + operation: 'execute_command', + workspaceId: req.workspaceId, + stdout: '', + stderr: '', + exitCode: 0, + truncated: false, + timedOut: false, + }; + }, + }) + ); + await pool.registerRoot('conversation', { + workspaceRoot: '/fixture/conversation', + }); + await pool.execute(request('conversation')); + assert.deepEqual(created, ['/fixture/conversation']); + await assert.rejects( + async () => + pool.registerRoot('conversation', { + workspaceRoot: '/fixture/replaced', + }), + { code: 'REGISTRATION_INVALID' } + ); + await pool.close(); +}); + +test('native pool retires a cached executor when a root inode changes', async () => { + let created = 0; + let closed = 0; + const pool = new NativeWorkspaceCommandPool( + new Map([['primary', { workspaceRoot: '/fixture/primary' }]]), + 2, + () => { + created++; + return { + async prepare() {}, + async close() { + closed++; + }, + async execute(req) { + return { + protocolVersion: 1, + operation: 'execute_command', + workspaceId: req.workspaceId, + stdout: '', + stderr: '', + exitCode: 0, + truncated: false, + timedOut: false, + }; + }, + }; + } + ); + const options = (ino: string) => ({ + workspaceRoot: '/fixture/conversation', + workspaceIdentity: { + path: '/fixture/conversation', + dev: '1', + ino, + }, + }); + await pool.registerRoot('conversation', options('1')); + await pool.execute(request('conversation')); + await pool.registerRoot('conversation', options('2')); + await pool.execute(request('conversation')); + + assert.equal(created, 2); + assert.equal(closed, 1); + await pool.close(); +}); + test('a known-clean executor failure is retired without replaying the command', async () => { let created = 0; let executed = 0; @@ -55,7 +169,7 @@ test('a known-clean executor failure is retired without replaying the command', throw new WorkspaceToolError( 'prepare failed', 'COMMAND_UNAVAILABLE', - false, + false ); return { protocolVersion: 1, diff --git a/packages/code/src/native-pool.ts b/packages/code/src/native-pool.ts index 766409b1..59db3e0a 100644 --- a/packages/code/src/native-pool.ts +++ b/packages/code/src/native-pool.ts @@ -12,9 +12,7 @@ interface Entry { NativeProcessWorkspaceCommandSandbox, 'prepare' | 'execute' | 'close' > & - Partial< - Pick - >; + Partial>; busy: boolean; } @@ -27,12 +25,12 @@ export class NativeWorkspaceCommandPool { private allocation: Promise = Promise.resolve(); private closing = false; constructor( - private readonly roots: ReadonlyMap, + roots: ReadonlyMap, private readonly capacity: number, private readonly createSandbox: ( - options: NativeProcessSandboxOptions, + options: NativeProcessSandboxOptions ) => Entry['sandbox'] = (options) => - new NativeProcessWorkspaceCommandSandbox(options), + new NativeProcessWorkspaceCommandSandbox(options) ) { if ( !Number.isSafeInteger(capacity) || @@ -42,6 +40,74 @@ export class NativeWorkspaceCommandPool { ) { throw new Error('Native executor capacity must be between 1 and 8'); } + this.roots = new Map(roots); + } + + private readonly roots: Map; + + /** Confirm executor cleanup before a failed provisioning attempt removes its root. */ + async unregisterRoot(id: string): Promise { + const pending = this.allocation.then(async () => { + const entry = this.entries.get(id); + if (entry?.busy) { + throw new WorkspaceToolError( + 'Native workspace still executing', + 'COMMAND_UNAVAILABLE' + ); + } + if (entry) await entry.sandbox.close(); + this.entries.delete(id); + this.roots.delete(id); + }); + this.allocation = pending.catch(() => undefined); + await pending; + } + + /** Add or safely replace a worker-owned isolated root. */ + async registerRoot( + id: string, + options: NativeProcessSandboxOptions + ): Promise { + const pending = this.allocation.then(async () => { + if (this.closing) { + throw new WorkspaceToolError( + 'Native workspace unavailable', + 'REGISTRATION_INVALID' + ); + } + const existing = this.roots.get(id); + if (!existing) { + this.roots.set(id, options); + return; + } + if (existing.workspaceRoot !== options.workspaceRoot) { + throw new WorkspaceToolError( + 'Native workspace identity changed', + 'REGISTRATION_INVALID' + ); + } + if ( + existing.workspaceIdentity?.dev === options.workspaceIdentity?.dev && + existing.workspaceIdentity?.ino === options.workspaceIdentity?.ino && + existing.workspaceIdentity?.path === options.workspaceIdentity?.path + ) { + return; + } + const entry = this.entries.get(id); + if (entry?.busy) { + throw new WorkspaceToolError( + 'Native workspace changed during execution', + 'REGISTRATION_INVALID' + ); + } + if (entry) { + await entry.sandbox.close(); + this.entries.delete(id); + } + this.roots.set(id, options); + }); + this.allocation = pending.catch(() => undefined); + await pending; } private allocate(root: string): Promise { @@ -50,23 +116,23 @@ export class NativeWorkspaceCommandPool { if (this.closing || !options) throw new WorkspaceToolError( 'Native workspace unavailable', - 'REGISTRATION_INVALID', + 'REGISTRATION_INVALID' ); let entry = this.entries.get(root); if (entry?.busy) throw new WorkspaceToolError( 'Native workspace already executing', - 'COMMAND_UNAVAILABLE', + 'COMMAND_UNAVAILABLE' ); if (!entry) { if (this.entries.size >= this.capacity) { const idle = [...this.entries].find( - ([, candidate]) => !candidate.busy, + ([, candidate]) => !candidate.busy ); if (!idle) throw new WorkspaceToolError( 'Native executor capacity reached', - 'COMMAND_UNAVAILABLE', + 'COMMAND_UNAVAILABLE' ); await idle[1].sandbox.close(); this.entries.delete(idle[0]); @@ -88,7 +154,7 @@ export class NativeWorkspaceCommandPool { if (error instanceof WorkspaceToolError) throw error; throw new WorkspaceToolError( 'Native executor allocation failed', - 'COMMAND_UNAVAILABLE', + 'COMMAND_UNAVAILABLE' ); }); this.allocation = checked.catch(() => undefined); @@ -112,14 +178,14 @@ export class NativeWorkspaceCommandPool { entry.busy = false; } } - }, - ), + } + ) ); } async execute( request: WorkspaceExecuteCommandRequest, - signal?: AbortSignal, + signal?: AbortSignal ): Promise { const entry = await this.allocate(request.workspaceId); let enteredExecutor = false; @@ -127,7 +193,7 @@ export class NativeWorkspaceCommandPool { if (signal?.aborted) throw new WorkspaceToolError( 'Command cancelled before dispatch', - 'EXECUTION_ABORTED', + 'EXECUTION_ABORTED' ); enteredExecutor = true; return await entry.sandbox.execute(request, signal); @@ -156,7 +222,7 @@ export class NativeWorkspaceCommandPool { async executeProgrammatic( workspaceId: string, request: BridgeWorkspaceProgrammaticRequest, - signal?: AbortSignal, + signal?: AbortSignal ): Promise { const entry = await this.allocate(workspaceId); let enteredExecutor = false; @@ -164,19 +230,19 @@ export class NativeWorkspaceCommandPool { if (signal?.aborted) throw new WorkspaceToolError( 'Programmatic execution cancelled before dispatch', - 'EXECUTION_ABORTED', + 'EXECUTION_ABORTED' ); enteredExecutor = true; if (!entry.sandbox.executeProgrammatic) { throw new WorkspaceToolError( 'Native programmatic executor is unavailable', - 'COMMAND_UNAVAILABLE', + 'COMMAND_UNAVAILABLE' ); } return await entry.sandbox.executeProgrammatic( workspaceId, request, - signal, + signal ); } catch (error) { if ( @@ -202,7 +268,7 @@ export class NativeWorkspaceCommandPool { this.closing = true; await this.allocation; const results = await Promise.allSettled( - [...this.entries.values()].map((entry) => entry.sandbox.close()), + [...this.entries.values()].map((entry) => entry.sandbox.close()) ); this.entries.clear(); const errors = results diff --git a/packages/code/src/native-process.test.ts b/packages/code/src/native-process.test.ts index b3016937..5f7eedbb 100644 --- a/packages/code/src/native-process.test.ts +++ b/packages/code/src/native-process.test.ts @@ -146,6 +146,7 @@ test('executor bootstrap excludes bridge credentials and Node injection variable assert.deepEqual(fake.options?.execArgv, []); assert.deepEqual(fake.options?.env, { PATH: '/bin' }); assert.equal(JSON.stringify(fake.messages).includes('secret'), false); + assert.equal('gitSharedObjectDirectory' in fake.messages[0].options, false); await sandbox.close(); }); diff --git a/packages/code/src/native-sandbox.test.ts b/packages/code/src/native-sandbox.test.ts index b1e4c5df..eab72966 100644 --- a/packages/code/src/native-sandbox.test.ts +++ b/packages/code/src/native-sandbox.test.ts @@ -590,6 +590,37 @@ test('trusted-vm permits unmatched egress and local development sockets', async ]); }); +test('recreated executors never grant reads through a replaced Git object directory', async t => { + const parent = await mkdtemp(join(tmpdir(), 'librechat-code-worktree-')); + const root = join(parent, 'worktree'); + const gitSharedObjectDirectory = join(parent, 'source.git', 'objects'); + await mkdir(join(root, '.git'), { recursive: true }); + await mkdir(gitSharedObjectDirectory, { recursive: true }); + await symlink(gitSharedObjectDirectory, join(root, '.git', 'objects')); + t.after(() => rm(parent, { recursive: true, force: true })); + const fake = fakeManager(); + const sandbox = new NativeSrtWorkspaceCommandSandbox({ + workspaceRoot: root, + manager: fake.manager, + }); + t.after(() => sandbox.close()); + + await sandbox.prepare(); + + assert.equal( + fake.config?.filesystem.allowRead?.includes( + await realpath(gitSharedObjectDirectory), + ), + false, + ); + assert.equal( + fake.config?.filesystem.allowWrite?.includes( + await realpath(gitSharedObjectDirectory), + ), + false, + ); +}); + test('provides an isolated scratch directory to commands and restores the host environment', async t => { const root = await mkdtemp(join(tmpdir(), 'librechat-code-native-')); t.after(() => rm(root, { recursive: true, force: true })); diff --git a/packages/code/src/process-lock.test.ts b/packages/code/src/process-lock.test.ts new file mode 100644 index 00000000..3594a1a9 --- /dev/null +++ b/packages/code/src/process-lock.test.ts @@ -0,0 +1,61 @@ +import assert from 'node:assert/strict'; +import { spawn } from 'node:child_process'; +import { once } from 'node:events'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import test from 'node:test'; +import { withProcessLock } from './process-lock.js'; + +test( + 'kernel lock survives contention and is released when the owning process crashes', + { timeout: 10_000 }, + async (t) => { + const directory = await mkdtemp(join(tmpdir(), 'librechat-lock-')); + const path = join(directory, '.provision.lock'); + const moduleUrl = new URL('./process-lock.js', import.meta.url).href; + const child = spawn( + process.execPath, + [ + '--input-type=module', + '-e', + ` + import { withProcessLock } from ${JSON.stringify(moduleUrl)}; + await withProcessLock(${JSON.stringify(path)}, async () => { + process.stdout.write('locked'); + await new Promise(() => { setInterval(() => {}, 1000); }); + }); + `, + ], + { stdio: ['ignore', 'pipe', 'pipe'] } + ); + const closed = once(child, 'close'); + t.after(async () => { + child.kill('SIGKILL'); + await closed; + await rm(directory, { recursive: true, force: true }); + }); + await once(child.stdout!, 'data'); + let entered = false; + await assert.rejects( + withProcessLock( + path, + async () => { + entered = true; + }, + AbortSignal.timeout(100) + ) + ); + assert.equal(entered, false); + child.kill('SIGKILL'); + await closed; + await withProcessLock( + path, + async () => { + entered = true; + }, + AbortSignal.timeout(1000) + ); + assert.equal(entered, true); + } +); diff --git a/packages/code/src/process-lock.ts b/packages/code/src/process-lock.ts new file mode 100644 index 00000000..aeca561a --- /dev/null +++ b/packages/code/src/process-lock.ts @@ -0,0 +1,59 @@ +import { constants } from 'node:fs'; +import { open } from 'node:fs/promises'; +import { setTimeout as delay } from 'node:timers/promises'; + +const LOCK_EX = 2; +const LOCK_NB = 4; +const LOCK_UN = 8; +let binding: + | Promise<{ + flock: (fd: number, operation: number) => number; + eagain: number; + errno: () => number; + }> + | undefined; + +async function lockBinding() { + binding ??= import('koffi').then(({ default: koffi }) => ({ + flock: koffi.load(null).func('int flock(int fd, int operation)'), + eagain: koffi.os.errno.EAGAIN, + errno: () => koffi.errno(), + })); + return await binding; +} + +/** Process-lifetime advisory lock; the kernel releases it on crash or restart. */ +export async function withProcessLock( + path: string, + operation: () => Promise, + signal?: AbortSignal +): Promise { + signal?.throwIfAborted(); + if (process.platform !== 'darwin' && process.platform !== 'linux') { + throw new Error('Conversation worktree locking requires a POSIX host'); + } + const native = await lockBinding(); + const handle = await open( + path, + constants.O_CREAT | constants.O_RDWR | constants.O_NOFOLLOW, + 0o600 + ); + try { + for (;;) { + signal?.throwIfAborted(); + if (native.flock(handle.fd, LOCK_EX | LOCK_NB) === 0) break; + const errno = native.errno(); + if (errno !== native.eagain) { + throw new Error( + `Conversation worktree lock failed with errno ${errno}` + ); + } + await delay(50, undefined, { signal }); + } + signal?.throwIfAborted(); + return await operation(); + } finally { + native.flock(handle.fd, LOCK_UN); + await handle.close(); + } +} diff --git a/packages/code/src/protocol.test.ts b/packages/code/src/protocol.test.ts index 08cba1c7..5c26c462 100644 --- a/packages/code/src/protocol.test.ts +++ b/packages/code/src/protocol.test.ts @@ -10,6 +10,7 @@ import { isValidBridgeWorkerId, isWorkspaceToolRequest, isWorkspaceToolResult, + workspaceIsolationKey, } from './protocol.js'; import type { WorkspaceEditFileRequest, @@ -90,6 +91,19 @@ test('bridgeWorkerPath encodes worker-controlled path segments', () => { ); }); +test('workspace isolation keys keep roots and instances in disjoint namespaces', () => { + const instanceId = 'a'.repeat(64); + assert.notEqual( + workspaceIsolationKey(`foo:git-worktree:${instanceId}`), + workspaceIsolationKey('foo', instanceId), + ); + assert.equal(workspaceIsolationKey('foo'), 'foo'); + assert.notEqual( + workspaceIsolationKey('foo'), + workspaceIsolationKey('workspace:foo'), + ); +}); + test('bridge worker IDs reject path, whitespace, and oversized values', () => { assert.equal(isValidBridgeWorkerId('engineering-vm:1'), true); assert.equal(isValidBridgeWorkerId('engineering/vm'), false); @@ -262,6 +276,20 @@ test('workspace file listing accepts only bounded portable requests and results' afterPath: 'src/app.ts', }; assert.equal(isWorkspaceToolRequest(request), true); + assert.equal( + isWorkspaceToolRequest({ + ...request, + workspaceInstanceId: 'a'.repeat(64), + }), + true, + ); + assert.equal( + isWorkspaceToolRequest({ + ...request, + workspaceInstanceId: 'conversation-1', + }), + false, + ); assert.equal( isWorkspaceToolRequest({ ...request, path: '../outside' }), false, @@ -596,7 +624,11 @@ test('workspace capabilities allow per-workspace operation restrictions', () => operations: ['read_file', 'write_file'], workspaces: [ { id: 'readonly', operations: ['read_file'] }, - { id: 'writable', operations: ['read_file', 'write_file'] }, + { + id: 'writable', + operations: ['read_file', 'write_file'], + workspaceInstances: ['git_worktree'], + }, ], }, }; @@ -660,6 +692,7 @@ test('workspace programmatic requests accept only stable input cache identities' max_output_files: 50, max_output_file_bytes: 10_000_000, session_id: 'session-1', + workspace_instance_id: 'a'.repeat(64), files: [ { name: 'main.sh', content: 'echo ready' }, { @@ -672,6 +705,13 @@ test('workspace programmatic requests accept only stable input cache identities' }, }; assert.equal(isBridgeWorkspaceProgrammaticRequest(request), true); + assert.equal( + isBridgeWorkspaceProgrammaticRequest({ + ...request, + body: { ...request.body, workspace_instance_id: '../escape' }, + }), + false, + ); assert.equal( isBridgeWorkspaceProgrammaticRequest({ ...request, diff --git a/packages/code/src/protocol.ts b/packages/code/src/protocol.ts index 27220735..c678a1af 100644 --- a/packages/code/src/protocol.ts +++ b/packages/code/src/protocol.ts @@ -259,6 +259,16 @@ export function bridgeArtifactMediaType(name: string): string { export type BridgeProtocolVersion = typeof BRIDGE_PROTOCOL_VERSION; +/** Collision-free identity shared by scheduling and worker quarantine state. */ +export function workspaceIsolationKey( + workspaceId: string, + instanceId?: string, +): string { + return instanceId === undefined + ? workspaceId + : `\0git-worktree\0${workspaceId}\0${instanceId}`; +} + export type BridgeWorkspaceToolOperation = | 'read_file' | 'search_text' @@ -280,6 +290,8 @@ export interface BridgeWorkspaceDescriptor { instructions?: RepositoryInstructionDescriptor[]; /** Optional per-workspace restriction. Omitted by protocol-v1 readers. */ operations?: BridgeWorkspaceToolOperation[]; + /** Worker-owned isolation schemes available beneath this selected root. */ + workspaceInstances?: ['git_worktree']; environment?: { fingerprint: string; repo?: string; @@ -308,6 +320,7 @@ export interface WorkspaceReadFileRequest { protocolVersion: BridgeProtocolVersion; operation: 'read_file'; workspaceId: string; + workspaceInstanceId?: string; path: string; startLine?: number; maxLines?: number; @@ -350,6 +363,7 @@ export interface WorkspaceSearchTextRequest { protocolVersion: BridgeProtocolVersion; operation: 'search_text'; workspaceId: string; + workspaceInstanceId?: string; query: string; path?: string; maxResults?: number; @@ -374,6 +388,7 @@ export interface WorkspaceListFilesRequest { protocolVersion: BridgeProtocolVersion; operation: 'list_files'; workspaceId: string; + workspaceInstanceId?: string; path?: string; maxResults?: number; /** Continue strictly after this canonical path from a previous page. */ @@ -394,6 +409,7 @@ export interface WorkspaceWriteFileRequest { protocolVersion: BridgeProtocolVersion; operation: 'write_file'; workspaceId: string; + workspaceInstanceId?: string; path: string; content: string; /** False requires an atomic create and refuses to replace an existing file. */ @@ -413,6 +429,7 @@ interface WorkspaceEditFileRequestBase { protocolVersion: BridgeProtocolVersion; operation: 'edit_file'; workspaceId: string; + workspaceInstanceId?: string; path: string; /** Refuses the mutation unless current file bytes match this preview revision. */ expectedBaseSha256?: string; @@ -457,6 +474,7 @@ interface WorkspacePreviewEditRequestBase { protocolVersion: BridgeProtocolVersion; operation: 'preview_edit'; workspaceId: string; + workspaceInstanceId?: string; path: string; } @@ -494,6 +512,7 @@ export interface WorkspaceExecuteCommandRequest { protocolVersion: BridgeProtocolVersion; operation: 'execute_command'; workspaceId: string; + workspaceInstanceId?: string; /** Shell source evaluated only inside the selected sandbox runtime. */ command: string; /** Portable path relative to the workspace root; defaults to '.'. */ @@ -538,6 +557,7 @@ const WORKSPACE_READ_REQUEST_KEYS = new Set([ 'protocolVersion', 'operation', 'workspaceId', + 'workspaceInstanceId', 'path', 'startLine', 'maxLines', @@ -546,6 +566,7 @@ const WORKSPACE_SEARCH_REQUEST_KEYS = new Set([ 'protocolVersion', 'operation', 'workspaceId', + 'workspaceInstanceId', 'query', 'path', 'maxResults', @@ -554,6 +575,7 @@ const WORKSPACE_LIST_REQUEST_KEYS = new Set([ 'protocolVersion', 'operation', 'workspaceId', + 'workspaceInstanceId', 'path', 'maxResults', 'afterPath', @@ -562,6 +584,7 @@ const WORKSPACE_WRITE_REQUEST_KEYS = new Set([ 'protocolVersion', 'operation', 'workspaceId', + 'workspaceInstanceId', 'path', 'content', 'overwrite', @@ -570,6 +593,7 @@ const WORKSPACE_EDIT_REQUEST_KEYS = new Set([ 'protocolVersion', 'operation', 'workspaceId', + 'workspaceInstanceId', 'path', 'oldText', 'newText', @@ -580,6 +604,7 @@ const WORKSPACE_PREVIEW_EDIT_REQUEST_KEYS = new Set([ 'protocolVersion', 'operation', 'workspaceId', + 'workspaceInstanceId', 'path', 'oldText', 'newText', @@ -591,6 +616,7 @@ const WORKSPACE_COMMAND_REQUEST_KEYS = new Set([ 'protocolVersion', 'operation', 'workspaceId', + 'workspaceInstanceId', 'command', 'cwd', 'timeoutMs', @@ -702,6 +728,8 @@ export interface BridgeWorkerRegistrationResponse { supportedWorkspaceListFileFeatures?: WorkspaceListFileFeature[]; /** PTC languages this Code API can safely route into a selected workspace. */ supportedWorkspaceProgrammaticLanguages?: WorkspaceProgrammaticLanguage[]; + /** Workspace isolation schemes this Code API understands and can route. */ + supportedWorkspaceInstanceTypes?: ['git_worktree']; } /** Administrator-visible liveness for a configured worker. Credentials, @@ -748,6 +776,7 @@ export type BridgeProgrammaticPayloadFile = export interface BridgeWorkspaceProgrammaticBody { language: 'bash'; version: string; + workspace_instance_id?: string; /** Stable identity shared by every replay iteration of one execution. */ execution_id?: string; /** Declared replay tools; zero allows the worker to skip the probe pass. */ @@ -912,6 +941,9 @@ export function isBridgeWorkspaceProgrammaticRequest( typeof body.version !== 'string' || body.version.length === 0 || body.version.length > BRIDGE_RUNTIME_MAX_LENGTH || + (body.workspace_instance_id !== undefined && + (typeof body.workspace_instance_id !== 'string' || + !/^[a-f0-9]{64}$/.test(body.workspace_instance_id))) || (body.execution_id !== undefined && (typeof body.execution_id !== 'string' || !/^[A-Za-z0-9_-]{1,128}$/.test(body.execution_id))) || @@ -1161,7 +1193,10 @@ export function isWorkspaceToolRequest( if ( request.protocolVersion !== BRIDGE_PROTOCOL_VERSION || typeof request.workspaceId !== 'string' || - !isValidBridgeWorkerId(request.workspaceId) + !isValidBridgeWorkerId(request.workspaceId) || + (request.workspaceInstanceId !== undefined && + (typeof request.workspaceInstanceId !== 'string' || + !/^[a-f0-9]{64}$/.test(request.workspaceInstanceId))) ) { return false; } @@ -1614,12 +1649,17 @@ export function isValidBridgeWorkspaceToolCapabilities( key !== 'id' && key !== 'name' && key !== 'operations' && + key !== 'workspaceInstances' && key !== 'instructions' && key !== 'environment', ) || typeof descriptor.id !== 'string' || !isValidBridgeWorkerId(descriptor.id) || workspaceIds.has(descriptor.id) || + (descriptor.workspaceInstances !== undefined && + (!Array.isArray(descriptor.workspaceInstances) || + descriptor.workspaceInstances.length !== 1 || + descriptor.workspaceInstances[0] !== 'git_worktree')) || (descriptor.instructions !== undefined && (!Array.isArray(descriptor.instructions) || descriptor.instructions.length > 1 || !descriptor.instructions.every(isRepositoryInstructionDescriptor))) || (descriptor.environment !== undefined && !isValidCodeEnvironmentDescriptor(descriptor.environment)) || diff --git a/packages/code/src/root-identity.ts b/packages/code/src/root-identity.ts index 3ddfe453..dd3bb2f5 100644 --- a/packages/code/src/root-identity.ts +++ b/packages/code/src/root-identity.ts @@ -6,6 +6,22 @@ export interface WorkspaceRootIdentity { ino: string; } +/** Capture the inode-bound identity of a canonical workspace grant. */ +export async function captureWorkspaceRootIdentity( + root: string, +): Promise { + const canonical = await realpath(root); + const current = await lstat(canonical, { bigint: true }); + if (!current.isDirectory() || current.isSymbolicLink()) { + throw new Error('Workspace root must be a real directory'); + } + return { + path: canonical, + dev: current.dev.toString(), + ino: current.ino.toString(), + }; +} + /** Revalidation of a trusted snapshot, never a fresh grant to a replacement. */ export async function matchesWorkspaceRoot( root: string, diff --git a/packages/code/src/worker-slots.test.ts b/packages/code/src/worker-slots.test.ts index 6e8cc8b1..fda6c659 100644 --- a/packages/code/src/worker-slots.test.ts +++ b/packages/code/src/worker-slots.test.ts @@ -5,6 +5,7 @@ import type { BridgeAssignment, BridgeWorkspaceToolCapabilities, } from './protocol.js'; +import { workspaceIsolationKey } from './protocol.js'; const capabilities: BridgeWorkspaceToolCapabilities = { protocolVersion: 1, @@ -229,7 +230,8 @@ for (const cancelled of [false, true]) { rejectUnexecutedAssignment: () => Promise; executeOwned: () => Promise; }; - internals.activeWorkspaceAssignments.set('a', { + const workspaceKey = workspaceIsolationKey('a'); + internals.activeWorkspaceAssignments.set(workspaceKey, { id: 'previous', done: new Promise(() => {}), }); @@ -257,7 +259,10 @@ for (const cancelled of [false, true]) { controller.signal, ); assert.equal(rejected, true); - assert.equal(internals.activeWorkspaceAssignments.get('a')?.id, 'previous'); + assert.equal( + internals.activeWorkspaceAssignments.get(workspaceKey)?.id, + 'previous', + ); }); } @@ -281,7 +286,8 @@ test('a local cleanup handoff preserves the new assignment owner and remaining b executeOwned: (assignment: BridgeAssignment) => Promise; }; let release!: () => void; - internals.activeWorkspaceAssignments.set('a', { + const workspaceKey = workspaceIsolationKey('a'); + internals.activeWorkspaceAssignments.set(workspaceKey, { id: 'previous', done: new Promise((resolve) => { release = resolve; @@ -290,7 +296,10 @@ test('a local cleanup handoff preserves the new assignment owner and remaining b let executed = false; internals.executeOwned = async (assignment) => { executed = true; - assert.equal(internals.activeWorkspaceAssignments.get('a')?.id, 'next'); + assert.equal( + internals.activeWorkspaceAssignments.get(workspaceKey)?.id, + 'next', + ); assert.ok(assignment.remainingMs! < 1000 && assignment.remainingMs! > 0); }; const pending = worker.executeAndSettle({ @@ -306,7 +315,7 @@ test('a local cleanup handoff preserves the new assignment owner and remaining b } as BridgeAssignment); await new Promise((resolve) => setTimeout(resolve, 5)); assert.equal(executed, false); - internals.activeWorkspaceAssignments.delete('a'); + internals.activeWorkspaceAssignments.delete(workspaceKey); release(); await pending; assert.equal(executed, true); @@ -332,14 +341,19 @@ test('programmatic work on an independent workspace bypasses another root cleanu >; executeOwned: (assignment: BridgeAssignment) => Promise; }; - internals.activeWorkspaceAssignments.set('a', { + const workspaceAKey = workspaceIsolationKey('a'); + const workspaceBKey = workspaceIsolationKey('b'); + internals.activeWorkspaceAssignments.set(workspaceAKey, { id: 'previous', done: new Promise(() => {}), }); let executed = false; internals.executeOwned = async () => { executed = true; - assert.equal(internals.activeWorkspaceAssignments.get('b')?.id, 'next'); + assert.equal( + internals.activeWorkspaceAssignments.get(workspaceBKey)?.id, + 'next', + ); }; await worker.executeAndSettle({ assignmentId: 'next', @@ -357,6 +371,9 @@ test('programmatic work on an independent workspace bypasses another root cleanu }, } as BridgeAssignment); assert.equal(executed, true); - assert.equal(internals.activeWorkspaceAssignments.has('b'), false); - assert.equal(internals.activeWorkspaceAssignments.get('a')?.id, 'previous'); + assert.equal(internals.activeWorkspaceAssignments.has(workspaceBKey), false); + assert.equal( + internals.activeWorkspaceAssignments.get(workspaceAKey)?.id, + 'previous', + ); }); diff --git a/packages/code/src/worker.ts b/packages/code/src/worker.ts index 697a3c55..2a5896fd 100644 --- a/packages/code/src/worker.ts +++ b/packages/code/src/worker.ts @@ -7,6 +7,7 @@ import { bridgeWorkerPath, isBridgeWorkspaceProgrammaticRequest, isWorkspaceToolResult, + workspaceIsolationKey, } from './protocol.js'; import { EndpointRuntimeSupervisor } from './runtime.js'; import { signBridgeRequest } from './identity.js'; @@ -54,6 +55,13 @@ export interface BridgeWorkerOptions { workspaceMutationQuarantine?: WorkspaceMutationQuarantine; /** Required per-root durable guards when opting into concurrent workspace leases. */ workspaceQuarantines?: ReadonlyMap; + /** Resolve a durable guard for a worker-owned dynamic workspace instance. */ + workspaceQuarantineResolver?: ( + workspaceId: string, + workspaceInstanceId: string, + ) => + | WorkspaceMutationQuarantine + | Promise; leaseWaitMs?: number; leaseTransportGraceMs?: number; registrationTransportTimeoutMs?: number; @@ -208,7 +216,14 @@ function workspaceCapabilitiesMatch( operation === executor.workspaces[index]?.operations?.[operationIndex], ) ?? - executor.workspaces[index]?.operations == null), + executor.workspaces[index]?.operations == null) && + workspace.workspaceInstances?.length === + executor.workspaces[index]?.workspaceInstances?.length && + (workspace.workspaceInstances?.every( + (instanceType, instanceIndex) => + instanceType === + executor.workspaces[index]?.workspaceInstances?.[instanceIndex], + ) ?? executor.workspaces[index]?.workspaceInstances == null), ) ); } @@ -223,7 +238,8 @@ function registrationCompatibleCapabilities( (operation) => operation === 'read_file' || operation === 'search_text', ) && workspaceTools.workspaces.every( - (workspace) => workspace.operations == null, + (workspace) => + workspace.operations == null && workspace.workspaceInstances == null, )) ) { return capabilities; @@ -244,7 +260,11 @@ function registrationCompatibleCapabilities( ) { return []; } - const { operations: _operations, ...compatibleWorkspace } = workspace; + const { + operations: _operations, + workspaceInstances: _workspaceInstances, + ...compatibleWorkspace + } = workspace; return [{ ...compatibleWorkspace, ...(workspace.environment ? { environment: { ...workspace.environment, actions: [] }, } : {}) }]; @@ -329,6 +349,12 @@ function supportedWorkspaceCapabilities( return workspaceOperations.length === 0 ? [] : [{ ...workspace, + ...(workspace.workspaceInstances != null && + registration.supportedWorkspaceInstanceTypes?.includes( + 'git_worktree', + ) + ? { workspaceInstances: workspace.workspaceInstances } + : { workspaceInstances: undefined }), ...(workspace.operations ? { operations: workspaceOperations } : {}), ...(workspace.environment && !workspaceOperations.includes('execute_command') ? { environment: { ...workspace.environment, actions: [] }, @@ -479,12 +505,23 @@ export class BridgeWorker { operation === 'execute_command', ) === true && options.workspaceMutationQuarantine == null && - options.workspaceQuarantines == null + options.workspaceQuarantines == null && + options.workspaceQuarantineResolver == null ) { throw new BridgeProtocolError( 'Workspace mutation capabilities require durable quarantine storage', ); } + if ( + options.capabilities.workspaceTools?.workspaces.some( + (root) => (root.workspaceInstances?.length ?? 0) > 0, + ) && + options.workspaceQuarantineResolver == null + ) { + throw new BridgeProtocolError( + 'Workspace instance capabilities require a durable quarantine resolver', + ); + } if ((options.capabilities.workspaceLeaseSlots ?? 1) > 1) { if ( options.capabilities.requiresReadyConfirmation !== true || @@ -785,14 +822,25 @@ export class BridgeWorker { async resetNativeWorkspace( workspaceId: string, signal?: AbortSignal, + workspaceInstanceId?: string, ): Promise { - const guard = this.options.workspaceQuarantines?.get(workspaceId); + const workspace = this.options.capabilities.workspaceTools?.workspaces.find( + (root) => root.id === workspaceId, + ); + const key = workspaceIsolationKey(workspaceId, workspaceInstanceId); + const guard = + workspaceInstanceId == null + ? this.options.workspaceQuarantines?.get(workspaceId) + : await this.options.workspaceQuarantineResolver?.( + workspaceId, + workspaceInstanceId, + ); if ( !guard || this.activeWorkspaceAssignments.size > 0 || - !this.options.capabilities.workspaceTools?.workspaces.some( - (root) => root.id === workspaceId, - ) + workspace == null || + (workspaceInstanceId != null && + workspace.workspaceInstances?.includes('git_worktree') !== true) ) { throw new BridgeProtocolError( 'Native workspace reset requires an idle registered root', @@ -808,14 +856,14 @@ export class BridgeWorker { { protocolVersion: BRIDGE_PROTOCOL_VERSION, incarnationId: this.incarnationId, - runtimeSessionId: `native-workspace:${workspaceId}`, + runtimeSessionId: `native-workspace:${key}`, confirmDiscarded: true, }, this.options.resetTransportTimeoutMs ?? DEFAULT_CONTROL_TRANSPORT_TIMEOUT_MS, signal, ); - this.quarantinedWorkspaces.delete(workspaceId); + this.quarantinedWorkspaces.delete(key); } async lease( @@ -1340,8 +1388,18 @@ export class BridgeWorker { private workspaceGuard( assignment: BridgeAssignment, - ): WorkspaceMutationQuarantine | undefined { - const workspaceId = this.assignmentWorkspaceId(assignment); + ): + | WorkspaceMutationQuarantine + | Promise + | undefined { + const workspaceId = this.assignmentBaseWorkspaceId(assignment); + const instanceId = this.assignmentWorkspaceInstanceId(assignment); + if (workspaceId != null && instanceId != null) { + return this.options.workspaceQuarantineResolver?.( + workspaceId, + instanceId, + ); + } return workspaceId != null ? (this.options.workspaceQuarantines?.get(workspaceId) ?? this.options.workspaceMutationQuarantine) @@ -1350,6 +1408,15 @@ export class BridgeWorker { private assignmentWorkspaceId( assignment: BridgeAssignment, + ): string | undefined { + const workspaceId = this.assignmentBaseWorkspaceId(assignment); + if (workspaceId == null) return undefined; + const instanceId = this.assignmentWorkspaceInstanceId(assignment); + return workspaceIsolationKey(workspaceId, instanceId); + } + + private assignmentBaseWorkspaceId( + assignment: BridgeAssignment, ): string | undefined { if ( assignment.executionKind === 'workspace_tool' && @@ -1366,11 +1433,33 @@ export class BridgeWorker { return undefined; } + private assignmentWorkspaceInstanceId( + assignment: BridgeAssignment, + ): string | undefined { + if ( + assignment.executionKind === 'workspace_tool' && + isWorkspaceToolRequest(assignment.request) + ) { + return assignment.request.workspaceInstanceId; + } + if ( + assignment.executionKind === 'workspace_programmatic' && + isBridgeWorkspaceProgrammaticRequest(assignment.request) + ) { + return assignment.request.body.workspace_instance_id; + } + return undefined; + } + private async executeOwned( assignment: BridgeAssignment, signal?: AbortSignal, ): Promise { - const guard = this.workspaceGuard(assignment); + const unresolvedGuard = this.workspaceGuard(assignment); + const guard = + unresolvedGuard instanceof Promise + ? await unresolvedGuard + : unresolvedGuard; if (signal?.aborted === true) { throw signal.reason instanceof Error ? signal.reason @@ -1484,12 +1573,17 @@ export class BridgeWorker { throw new BridgeProtocolError('Invalid workspace tool request'); } const workspaceRequest = assignment.request; + const workspaceKey = this.assignmentWorkspaceId(assignment)!; try { - if (this.quarantinedWorkspaces.has(workspaceRequest.workspaceId)) { + if (this.quarantinedWorkspaces.has(workspaceKey)) { throw new Error('Workspace requires an explicit quarantine reset'); } - if (this.options.workspaceQuarantines != null) + if ( + this.options.workspaceQuarantines != null || + this.options.workspaceQuarantineResolver != null + ) { await guard?.assertAvailable(); + } } catch (error) { throw new BridgeWorkspaceQuarantinedError( 'Workspace is quarantined', @@ -1513,6 +1607,14 @@ export class BridgeWorker { if (workspace == null) { throw new BridgeProtocolError('Workspace is not advertised'); } + if ( + workspaceRequest.workspaceInstanceId != null && + workspace.workspaceInstances?.includes('git_worktree') !== true + ) { + throw new BridgeProtocolError( + 'Workspace instance type is not advertised', + ); + } if ( workspace.operations != null && !workspace.operations.includes(workspaceRequest.operation) @@ -1575,7 +1677,7 @@ export class BridgeWorker { if (isMutation) { this.mutationGuardArmed = true; try { - this.armedWorkspaces.add(workspaceRequest.workspaceId); + this.armedWorkspaces.add(workspaceKey); await guard!.arm( `Workspace mutation ${workspaceRequest.operation} is pending settlement`, assignment.assignmentId, @@ -1630,12 +1732,17 @@ export class BridgeWorker { 'Worker does not provide valid selected-workspace programmatic execution', ); } + const workspaceKey = this.assignmentWorkspaceId(assignment)!; try { - if (this.quarantinedWorkspaces.has(workspaceId)) { + if (this.quarantinedWorkspaces.has(workspaceKey)) { throw new Error('Workspace requires an explicit quarantine reset'); } - if (this.options.workspaceQuarantines != null) + if ( + this.options.workspaceQuarantines != null || + this.options.workspaceQuarantineResolver != null + ) { await guard?.assertAvailable(); + } } catch (error) { throw new BridgeWorkspaceQuarantinedError( 'Workspace is quarantined', @@ -1657,9 +1764,17 @@ export class BridgeWorker { 'Selected-workspace programmatic execution is not advertised', ); } + if ( + assignment.request.body.workspace_instance_id != null && + workspace.workspaceInstances?.includes('git_worktree') !== true + ) { + throw new BridgeProtocolError( + 'Workspace instance type is not advertised', + ); + } this.mutationGuardArmed = true; try { - this.armedWorkspaces.add(workspaceId); + this.armedWorkspaces.add(workspaceKey); await guard!.arm( 'Workspace programmatic execution is pending settlement', assignment.assignmentId, @@ -2079,11 +2194,15 @@ export class BridgeWorker { ): Promise { if (runtimeSessionId == null) { try { - await ( + const unresolvedGuard = assignment == null ? this.options.workspaceMutationQuarantine - : this.workspaceGuard(assignment) - )?.quarantine(message, cause, assignment?.assignmentId); + : this.workspaceGuard(assignment); + const guard = + unresolvedGuard instanceof Promise + ? await unresolvedGuard + : unresolvedGuard; + await guard?.quarantine(message, cause, assignment?.assignmentId); return new BridgeWorkspaceQuarantinedError(message, cause); } catch (error) { return new BridgeWorkspaceQuarantinedError( diff --git a/packages/code/src/workspace-cli.test.ts b/packages/code/src/workspace-cli.test.ts index c1bc7fce..1de68818 100644 --- a/packages/code/src/workspace-cli.test.ts +++ b/packages/code/src/workspace-cli.test.ts @@ -115,6 +115,38 @@ test('CLI supports native SRT by default and validates explicit runtime mode', a assert.match(noWorkspace.stderr, /require.*registered directory/i); }); +test('CLI requires concurrent native slots for conversation worktrees', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'librechat-code-conversation-')); + const workspaceRoot = join(root, 'workspace'); + const worktreeRoot = join(root, 'worktrees'); + await mkdir(workspaceRoot); + t.after(() => rm(root, { recursive: true, force: true })); + const result = spawnSync( + process.execPath, + [ + fileURLToPath(new URL('./cli.js', import.meta.url)), + 'run', + '--worker-dir', + workspaceRoot, + '--allow-workspace-writes', + '--allow-workspace-commands', + '--conversation-worktree-root', + worktreeRoot, + ], + { + encoding: 'utf8', + env: { + ...process.env, + LIBRECHAT_CODE_URL: 'http://127.0.0.1:1/v1', + LIBRECHAT_CODE_WORKER_TOKEN: 'worker-secret', + LIBRECHAT_CODE_WORKER_ID: 'engineering-vm', + }, + }, + ); + assert.notEqual(result.status, 0); + assert.match(result.stderr, /at least two workspace lease slots/i); +}); + test('CLI advertises explicitly enabled writes without exposing the workspace root', async (t) => { const root = await mkdtemp(join(tmpdir(), 'librechat-code-cli-')); const workspaceRoot = join(root, ' '); diff --git a/packages/code/src/workspace-instances.test.ts b/packages/code/src/workspace-instances.test.ts new file mode 100644 index 00000000..a1cd6859 --- /dev/null +++ b/packages/code/src/workspace-instances.test.ts @@ -0,0 +1,234 @@ +import assert from 'node:assert/strict'; +import { execFile } from 'node:child_process'; +import { mkdtemp, readFile, realpath, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { promisify } from 'node:util'; +import test from 'node:test'; + +import { GitWorktreeWorkspaceTools } from './workspace-instances.js'; +import { LocalWorkspaceTools } from './workspace.js'; +import { GitWorktreeManager } from './worktrees.js'; +import { readRepositoryInstructions } from './instructions.js'; +import { captureWorkspaceRootIdentity } from './root-identity.js'; + +const execFileAsync = promisify(execFile); + +async function repository(): Promise<{ parent: string; root: string }> { + const parent = await mkdtemp(join(tmpdir(), 'librechat-instance-tools-')); + const root = join(parent, 'source'); + await execFileAsync('git', ['init', root]); + await writeFile(join(root, 'README.md'), 'source\n'); + await writeFile(join(root, 'AGENTS.md'), 'follow repository rules\n'); + await execFileAsync('git', ['-C', root, 'add', 'README.md', 'AGENTS.md']); + await execFileAsync('git', [ + '-C', + root, + '-c', + 'user.name=Test', + '-c', + 'user.email=test@example.com', + 'commit', + '-m', + 'initial', + ]); + return { parent, root: await realpath(root) }; +} + +async function source(root: string) { + return { root, identity: await captureWorkspaceRootIdentity(root) }; +} + +test('routes each conversation to its own writable Git worktree', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const delegate = await LocalWorkspaceTools.create({ + repositoryInstructions: true, + workspaces: [{ id: 'primary', root: fixture.root, writable: true }], + }); + const manager = new GitWorktreeManager({ + maxCount: 4, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + const tools = new GitWorktreeWorkspaceTools({ + delegate, + manager, + sources: new Map([ + ['primary', { repositoryInstructions: true, writable: true }], + ]), + }); + const firstId = 'a'.repeat(64); + const secondId = 'b'.repeat(64); + + assert.deepEqual(tools.capabilities.workspaces[0]?.workspaceInstances, [ + 'git_worktree', + ]); + await tools.execute({ + protocolVersion: 1, + operation: 'write_file', + workspaceId: 'primary', + workspaceInstanceId: firstId, + path: 'conversation.txt', + content: 'first', + }); + await tools.execute({ + protocolVersion: 1, + operation: 'write_file', + workspaceId: 'primary', + workspaceInstanceId: secondId, + path: 'conversation.txt', + content: 'second', + }); + + const first = await manager.resolve('primary', firstId); + const second = await manager.resolve('primary', secondId); + assert.equal( + await readFile(join(first.root, 'conversation.txt'), 'utf8'), + 'first', + ); + assert.equal( + await readFile(join(second.root, 'conversation.txt'), 'utf8'), + 'second', + ); + await assert.rejects(readFile(join(fixture.root, 'conversation.txt')), { + code: 'ENOENT', + }); + + const result = await tools.execute({ + protocolVersion: 1, + operation: 'read_file', + workspaceId: 'primary', + workspaceInstanceId: firstId, + path: 'conversation.txt', + }); + assert.equal(result.workspaceId, 'primary'); + assert.equal(result.operation, 'read_file'); + assert.equal(result.content, 'first'); + + await writeFile(join(fixture.root, 'AGENTS.md'), 'local repository rules\n'); + const instructions = await readRepositoryInstructions(fixture.root); + assert.ok(instructions); + const instructionResult = await tools.execute({ + protocolVersion: 1, + operation: 'read_file', + workspaceId: 'primary', + workspaceInstanceId: firstId, + path: instructions.descriptor.path, + instructionSha256: instructions.descriptor.sha256, + }); + assert.equal(instructionResult.operation, 'read_file'); + assert.equal(instructionResult.content, 'local repository rules\n'); +}); + +test('reports provisioning rejection as an atomic workspace error', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const delegate = await LocalWorkspaceTools.create({ + workspaces: [{ id: 'primary', root: fixture.root, writable: true }], + }); + const tools = new GitWorktreeWorkspaceTools({ + delegate, + manager: new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }), + sources: new Map([ + ['primary', { repositoryInstructions: false, writable: true }], + ]), + }); + await tools.execute({ + protocolVersion: 1, + operation: 'write_file', + workspaceId: 'primary', + workspaceInstanceId: 'a'.repeat(64), + path: 'first.txt', + content: 'first', + }); + await assert.rejects( + tools.execute({ + protocolVersion: 1, + operation: 'write_file', + workspaceId: 'primary', + workspaceInstanceId: 'b'.repeat(64), + path: 'second.txt', + content: 'second', + }), + { + code: 'WRITE_UNAVAILABLE', + mutationMayHaveCommitted: false, + requiresQuarantine: false, + }, + ); +}); + +test('rebuilds file executors only after operator recovery releases the reservation', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + const delegate = await LocalWorkspaceTools.create({ + workspaces: [{ id: 'primary', root: fixture.root, writable: true }], + }); + const tools = new GitWorktreeWorkspaceTools({ + delegate, + manager, + sources: new Map([ + ['primary', { repositoryInstructions: false, writable: true }], + ]), + }); + const instanceId = 'c'.repeat(64); + const request = { + protocolVersion: 1 as const, + operation: 'read_file' as const, + workspaceId: 'primary', + workspaceInstanceId: instanceId, + path: 'README.md', + }; + await tools.execute(request); + const initial = await manager.resolve('primary', instanceId); + await rm(initial.root, { recursive: true, force: true }); + + await assert.rejects(tools.execute(request), { + code: 'WRITE_UNAVAILABLE', + }); + await assert.rejects(tools.execute(request), /capacity is exhausted/); + // Missing checkout directories do not prove an interrupted writer is gone. + // Simulate operator recovery after confirming there is no active executor. + await rm(`${initial.root}.complete`); + const recovered = await tools.execute(request); + assert.equal(recovered.operation, 'read_file'); + assert.equal(recovered.content, 'source'); +}); + +test('leaves legacy requests on the selected source workspace', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const delegate = await LocalWorkspaceTools.create({ + workspaces: [{ id: 'primary', root: fixture.root, writable: false }], + }); + const tools = new GitWorktreeWorkspaceTools({ + delegate, + manager: new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }), + sources: new Map([ + ['primary', { repositoryInstructions: false, writable: false }], + ]), + }); + + const result = await tools.execute({ + protocolVersion: 1, + operation: 'read_file', + workspaceId: 'primary', + path: 'README.md', + }); + assert.equal(result.operation, 'read_file'); + assert.equal(result.content, 'source'); +}); diff --git a/packages/code/src/workspace-instances.ts b/packages/code/src/workspace-instances.ts new file mode 100644 index 00000000..68e76a10 --- /dev/null +++ b/packages/code/src/workspace-instances.ts @@ -0,0 +1,232 @@ +import { createHash } from 'node:crypto'; + +import { NativeWorkspaceCommandPool } from './native-pool.js'; +import { GitWorktreeManager } from './worktrees.js'; +import { LocalWorkspaceTools, WorkspaceToolError } from './workspace.js'; + +import type { NativeProcessSandboxOptions } from './native-process.js'; +import type { WorkspaceRootIdentity } from './root-identity.js'; +import type { + BridgeWorkspaceProgrammaticRequest, + WorkspaceExecuteCommandRequest, + WorkspaceToolRequest, + WorkspaceToolResult, +} from './protocol.js'; +import type { WorkspaceToolExecutor } from './workspace.js'; + +interface WorkspaceInstanceSource { + command?: NativeProcessSandboxOptions; + repositoryInstructions: boolean; + writable: boolean; +} + +export interface GitWorktreeWorkspaceToolsOptions { + commandPool?: NativeWorkspaceCommandPool; + delegate: WorkspaceToolExecutor; + manager: GitWorktreeManager; + onResolve?: (workspaceId: string, root: string) => void; + sources: ReadonlyMap; +} + +export function internalWorkspaceId(workspaceId: string, instanceId: string): string { + return `instance-${createHash('sha256') + .update(`${workspaceId}\0${instanceId}`) + .digest('hex')}`; +} + +function publicResult( + result: WorkspaceToolResult, + workspaceId: string, +): WorkspaceToolResult { + return { ...result, workspaceId }; +} + +/** Resolve an opaque conversation binding into an isolated Git worktree. */ +export class GitWorktreeWorkspaceTools implements WorkspaceToolExecutor { + readonly mutationFailuresAreAtomic?: true; + readonly capabilities: WorkspaceToolExecutor['capabilities']; + private readonly executors = new Map< + string, + { identity: WorkspaceRootIdentity; value: Promise } + >(); + + constructor(private readonly options: GitWorktreeWorkspaceToolsOptions) { + this.mutationFailuresAreAtomic = options.delegate.mutationFailuresAreAtomic; + this.capabilities = { + ...options.delegate.capabilities, + workspaces: options.delegate.capabilities.workspaces.map((workspace) => ({ + ...workspace, + ...(options.sources.has(workspace.id) + ? { workspaceInstances: ['git_worktree' as const] } + : {}), + })), + }; + } + + private async executor( + workspaceId: string, + instanceId: string, + signal?: AbortSignal, + ): Promise<{ + executor: LocalWorkspaceTools; + identity: WorkspaceRootIdentity; + internalId: string; + root: string; + }> { + const source = this.options.sources.get(workspaceId); + if (!source) { + throw new WorkspaceToolError( + 'Workspace does not allow conversation worktrees', + 'INVALID_REQUEST', + ); + } + let instance; + try { + instance = await this.options.manager.resolve( + workspaceId, + instanceId, + signal, + ); + } catch (error) { + if (error instanceof WorkspaceToolError) throw error; + if ( + signal?.aborted || + (error instanceof Error && error.name === 'AbortError') + ) { + throw new WorkspaceToolError( + 'Conversation worktree provisioning aborted', + 'EXECUTION_ABORTED', + ); + } + throw new WorkspaceToolError( + error instanceof Error + ? error.message + : 'Conversation worktree provisioning failed', + 'WRITE_UNAVAILABLE', + ); + } + this.options.onResolve?.(workspaceId, instance.root); + const internalId = internalWorkspaceId(workspaceId, instanceId); + const key = `${workspaceId}\0${instanceId}`; + let cached = this.executors.get(key); + if ( + cached == null || + cached.identity.dev !== instance.identity.dev || + cached.identity.ino !== instance.identity.ino || + cached.identity.path !== instance.identity.path + ) { + cached = { + identity: instance.identity, + value: LocalWorkspaceTools.create({ + repositoryInstructions: source.repositoryInstructions, + workspaces: [ + { + id: internalId, + identity: instance.identity, + root: instance.root, + writable: source.writable, + }, + ], + }), + }; + this.executors.set(key, cached); + } + return { + executor: await cached.value, + identity: instance.identity, + internalId, + root: instance.root, + }; + } + + async execute( + request: WorkspaceToolRequest, + signal?: AbortSignal, + ): Promise { + if (!request.workspaceInstanceId) { + return await this.options.delegate.execute(request, signal); + } + if ( + request.operation === 'read_file' && + request.instructionSha256 !== undefined + ) { + const { workspaceInstanceId: _workspaceInstanceId, ...sourceRequest } = + request; + return await this.options.delegate.execute(sourceRequest, signal); + } + const { workspaceInstanceId, ...baseRequest } = request; + const source = this.options.sources.get(request.workspaceId); + const resolved = await this.executor( + request.workspaceId, + workspaceInstanceId, + signal, + ); + const isolatedRequest = { + ...baseRequest, + workspaceId: resolved.internalId, + } as WorkspaceToolRequest; + if (request.operation === 'execute_command') { + if (!source?.command || !this.options.commandPool) { + throw new WorkspaceToolError( + 'Conversation worktree commands are unavailable', + 'COMMAND_DISABLED', + ); + } + await this.options.commandPool.registerRoot(resolved.internalId, { + ...source.command, + workspaceIdentity: resolved.identity, + workspaceRoot: resolved.root, + }); + return publicResult( + await this.options.commandPool.execute( + isolatedRequest as WorkspaceExecuteCommandRequest, + signal, + ), + request.workspaceId, + ); + } + return publicResult( + await resolved.executor.execute(isolatedRequest, signal), + request.workspaceId, + ); + } + + async executeProgrammatic( + workspaceId: string, + request: BridgeWorkspaceProgrammaticRequest, + signal?: AbortSignal, + ): Promise { + const instanceId = request.body.workspace_instance_id; + if (!instanceId) { + if (!this.options.commandPool) { + throw new WorkspaceToolError( + 'Workspace programmatic execution is unavailable', + 'COMMAND_DISABLED', + ); + } + return await this.options.commandPool.executeProgrammatic( + workspaceId, + request, + signal, + ); + } + const source = this.options.sources.get(workspaceId); + if (!source?.command || !this.options.commandPool) { + throw new WorkspaceToolError( + 'Conversation worktree programmatic execution is unavailable', + 'COMMAND_DISABLED', + ); + } + const resolved = await this.executor(workspaceId, instanceId, signal); + await this.options.commandPool.registerRoot(resolved.internalId, { + ...source.command, + workspaceIdentity: resolved.identity, + workspaceRoot: resolved.root, + }); + return await this.options.commandPool.executeProgrammatic( + resolved.internalId, + request, + signal, + ); + } +} diff --git a/packages/code/src/workspace-worker.test.ts b/packages/code/src/workspace-worker.test.ts index d97735dc..96a5d1f0 100644 --- a/packages/code/src/workspace-worker.test.ts +++ b/packages/code/src/workspace-worker.test.ts @@ -7,6 +7,19 @@ import { SandboxWorkspaceTools, WorkspaceToolError } from './workspace.js'; const incarnationId = 'incarnation-00000001'; +test('instance advertisement requires a guard resolver even for reads and with base guards', () => { + for (const operation of ['read_file', 'write_file'] as const) { + const workspaceTools = { protocolVersion: 1 as const, operations: [operation], workspaces: [{ id: 'primary', workspaceInstances: ['git_worktree'] as ['git_worktree'] }] }; + assert.throws(() => new BridgeWorker({ + codeApiUrl: 'https://code.example/v1', token: 'worker-secret', workerId: 'vm-1', incarnationId, + sandboxEndpoint: 'http://127.0.0.1:2000/api/v2', + capabilities: { statefulWorkspace: false, sandboxProfile: 'anthropic-srt', runtimes: [], workspaceTools }, + workspaceQuarantines: new Map([['primary', mutationQuarantine()]]), + workspaceTools: { capabilities: workspaceTools, async execute() { throw new Error('must not execute'); } }, + }), /instance capabilities require a durable quarantine resolver/); + } +}); + test('worker clears named actions when command execution is not negotiated', async () => { const workspaceTools = { protocolVersion: 1 as const, @@ -974,6 +987,93 @@ test('worker executes a workspace tool assignment locally without acquiring a sa }); }); +test('worker isolates dynamic worktree guards from collision-shaped root IDs', async () => { + const instanceId = 'a'.repeat(64); + const collisionRoot = `foo:git-worktree:${instanceId}`; + const lifecycle: string[] = []; + const workspaceCapabilities = { + protocolVersion: 1 as const, + operations: ['write_file' as const], + workspaces: [ + { id: 'foo', workspaceInstances: ['git_worktree'] as ['git_worktree'] }, + { id: collisionRoot }, + ], + }; + const worker = new BridgeWorker({ + codeApiUrl: 'https://code.example/v1', + token: 'worker-secret', + workerId: 'vm-1', + incarnationId, + sandboxEndpoint: 'http://127.0.0.1:2000/api/v2', + capabilities: { + statefulWorkspace: false, + sandboxProfile: 'anthropic-srt', + runtimes: [], + workspaceTools: workspaceCapabilities, + }, + workspaceTools: { + capabilities: workspaceCapabilities, + mutationFailuresAreAtomic: true, + async execute(request) { + return { + protocolVersion: 1, + operation: 'write_file', + workspaceId: request.workspaceId, + path: 'result.txt', + created: true, + bytesWritten: 2, + }; + }, + }, + workspaceQuarantines: new Map([ + [ + collisionRoot, + mutationQuarantine( + undefined, + () => lifecycle.push('root:arm'), + () => lifecycle.push('root:clear'), + ), + ], + ]), + workspaceQuarantineResolver: async () => + mutationQuarantine( + undefined, + () => lifecycle.push('instance:arm'), + () => lifecycle.push('instance:clear'), + ), + fetchImpl: async () => + Response.json({ protocolVersion: 1, accepted: true }), + }); + const assignment = (workspaceId: string, suffix: string) => ({ + protocolVersion: 1 as const, + assignmentId: `assignment-${suffix}`, + workerId: 'vm-1', + incarnationId, + generation: 4, + leaseToken: `lease-token-that-is-long-enough-${suffix}`, + expiresAt: new Date(Date.now() + 5_000).toISOString(), + executionKind: 'workspace_tool' as const, + request: { + protocolVersion: 1 as const, + operation: 'write_file' as const, + workspaceId, + path: 'result.txt', + content: 'ok', + ...(workspaceId === 'foo' ? { workspaceInstanceId: instanceId } : {}), + }, + }); + + await worker.executeAndSettle(assignment('foo', 'instance')); + await worker.executeAndSettle(assignment(collisionRoot, 'root')); + + assert.deepEqual(lifecycle, [ + 'instance:arm', + 'instance:clear', + 'root:arm', + 'root:clear', + ]); +}); + test('worker executes programmatic Bash in the selected workspace and preserves its fence', async () => { const programmaticRequests: object[] = []; const quarantineEvents: string[] = []; diff --git a/packages/code/src/worktrees.test.ts b/packages/code/src/worktrees.test.ts new file mode 100644 index 00000000..c353c24e --- /dev/null +++ b/packages/code/src/worktrees.test.ts @@ -0,0 +1,988 @@ +import assert from 'node:assert/strict'; +import { execFile, spawn } from 'node:child_process'; +import { once } from 'node:events'; +import { + mkdir, + mkdtemp, + readFile, + realpath, + rename, + rm, + stat, + symlink, + writeFile, +} from 'node:fs/promises'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { promisify } from 'node:util'; +import test from 'node:test'; +import { setTimeout as delay } from 'node:timers/promises'; + +import { GitWorktreeManager } from './worktrees.js'; +import { captureWorkspaceRootIdentity } from './root-identity.js'; +import { GitSourceSnapshot } from './git-snapshot.js'; + +const execFileAsync = promisify(execFile); + +test('aborting private staging releases its reservation only after cleanup', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + const controller = new AbortController(); + const copy = GitSourceSnapshot.prototype.copyTo; + const mocked = t.mock.method( + GitSourceSnapshot.prototype, + 'copyTo', + async function ( + this: GitSourceSnapshot, + destination: string, + signal?: AbortSignal + ) { + await copy.call(this, destination, signal); + controller.abort(); + } + ); + const id = 'b'.repeat(64); + const path = await manager.plannedRoot('primary', id); + await assert.rejects(manager.resolve('primary', id, controller.signal), { + name: 'AbortError', + }); + for (const name of [path, `${path}.source`, `${path}.complete`]) + await assert.rejects(stat(name), { code: 'ENOENT' }); + mocked.mock.restore(); + assert.ok(await manager.resolve('primary', 'c'.repeat(64))); +}); + +test('setup cannot publish a checkout that redirects its Git metadata', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + prepareInstance: async (instance) => { + await writeFile( + join(instance.root, '.git', 'commondir'), + join(fixture.root, '.git') + ); + }, + }); + await assert.rejects( + manager.resolve('primary', 'd'.repeat(64)), + /must not redirect/ + ); +}); + +test('private snapshots preserve SHA-256 repositories and the configured origin', async (t) => { + const fixture = await repository('sha256'); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + await git( + fixture.root, + 'remote', + 'add', + 'origin', + 'https://github.com/example/repo.git' + ); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + const instance = await manager.resolve('primary', 'a'.repeat(64)); + assert.equal( + await git(instance.root, 'rev-parse', '--show-object-format'), + 'sha256' + ); + assert.equal( + await git(instance.root, 'remote', 'get-url', 'origin'), + 'https://github.com/example/repo.git' + ); + assert.equal( + await readFile(join(instance.root, 'README.md'), 'utf8'), + 'source\n' + ); +}); + +test('sidecar-only crash reservations consume quota after restart', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const admitted = await source(fixture.root); + const options = { + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', admitted]]), + }; + const manager = new GitWorktreeManager(options); + const path = await manager.plannedRoot('primary', 'a'.repeat(64)); + await mkdir(join(path, '..'), { recursive: true }); + await writeFile( + `${path}.complete`, + JSON.stringify({ + version: 2, + source: admitted.identity, + sourceGit: (await GitSourceSnapshot.admit(admitted.identity)).fingerprint, + provisioningFailed: true, + }) + ); + const restarted = new GitWorktreeManager(options); + await assert.rejects( + restarted.resolve('primary', 'b'.repeat(64)), + /capacity is exhausted/ + ); + await assert.rejects( + restarted.resolve('primary', 'a'.repeat(64)), + /operator recovery required/ + ); + await assert.rejects(stat(path), { code: 'ENOENT' }); +}); + +test('linked-worktree sources retain their own HEAD and admitted common objects', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const linked = join(fixture.parent, 'linked'); + await git(fixture.root, 'worktree', 'add', '-b', 'linked', linked); + await writeFile(join(linked, 'linked.txt'), 'linked\n'); + await git(linked, 'add', 'linked.txt'); + await git( + linked, + '-c', + 'user.name=Test', + '-c', + 'user.email=test@example.com', + 'commit', + '-m', + 'linked' + ); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(linked)]]), + }); + const instance = await manager.resolve('primary', 'e'.repeat(64)); + assert.equal( + await readFile(join(instance.root, 'linked.txt'), 'utf8'), + 'linked\n' + ); +}); + +test('restart cannot rebind a completed checkout to replacement source Git metadata', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const options = { + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }; + const id = 'f'.repeat(64); + const instance = await new GitWorktreeManager(options).resolve('primary', id); + await writeFile(join(instance.root, 'uncommitted.txt'), 'keep'); + await rename(join(fixture.root, '.git'), join(fixture.root, '.git-original')); + await git(fixture.root, 'init'); + await assert.rejects( + new GitWorktreeManager(options).resolve('primary', id), + /source identity changed/ + ); + assert.equal( + await readFile(join(instance.root, 'uncommitted.txt'), 'utf8'), + 'keep' + ); +}); + +test('rejects source Git redirection without changing the admitted working-directory inode', async (t) => { + const fixture = await repository(); + const other = await repository(); + t.after(() => + Promise.all( + [fixture, other].map(({ parent }) => + rm(parent, { recursive: true, force: true }) + ) + ) + ); + const manager = new GitWorktreeManager({ + maxCount: 2, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + await manager.prepare(); + await rename(join(fixture.root, '.git'), join(fixture.root, '.git-original')); + await writeFile( + join(fixture.root, '.git'), + `gitdir: ${join(other.root, '.git')}\n` + ); + await assert.rejects( + manager.resolve('primary', 'c'.repeat(64)), + /Git metadata changed/ + ); +}); + +test('source replacement after validation cannot redirect the private snapshot', async (t) => { + const fixture = await repository(); + const other = await repository(); + t.after(() => + Promise.all( + [fixture, other].map(({ parent }) => + rm(parent, { recursive: true, force: true }) + ) + ) + ); + const snapshot = await GitSourceSnapshot.admit( + ( + await source(fixture.root) + ).identity + ); + const validate = snapshot.validate.bind(snapshot); + t.mock.method(snapshot, 'validate', async () => { + await validate(); + await rename( + join(fixture.root, '.git'), + join(fixture.root, '.git-original') + ); + await symlink(join(other.root, '.git'), join(fixture.root, '.git')); + }); + await assert.rejects(snapshot.copyTo(join(fixture.parent, 'snapshot'))); + await assert.rejects( + stat(join(fixture.parent, 'snapshot', 'objects', 'pack')), + { code: 'ENOENT' } + ); +}); + +test('cached instances reject object-directory redirection before executor recreation', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + const id = 'd'.repeat(64); + const instance = await manager.resolve('primary', id); + await rename( + join(instance.root, '.git', 'objects'), + join(instance.root, '.git', 'objects-original') + ); + await symlink( + join(fixture.root, '.git', 'objects'), + join(instance.root, '.git', 'objects') + ); + await assert.rejects( + manager.resolve('primary', id), + /does not own its Git objects/ + ); +}); + +test('snapshot rejects symlinked refs and never copies source hooks or config includes', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + await git(fixture.root, 'config', 'include.path', '/not-readable/config'); + const snapshot = await GitSourceSnapshot.admit( + ( + await source(fixture.root) + ).identity + ); + await snapshot.copyTo(join(fixture.parent, 'snapshot')); + assert.equal( + await readFile(join(fixture.parent, 'snapshot', 'config'), 'utf8'), + '[core]\nrepositoryformatversion = 0\nbare = true\n' + ); + await assert.rejects(stat(join(fixture.parent, 'snapshot', 'hooks')), { + code: 'ENOENT', + }); + await symlink( + join(fixture.root, 'README.md'), + join(fixture.root, '.git', 'refs', 'bad') + ); + await assert.rejects( + snapshot.copyTo(join(fixture.parent, 'snapshot-bad')), + /symbolic link/ + ); +}); + +test( + 'restart preserves a checkout reserved by a crashed provisioning process', + { timeout: 10_000 }, + async (t) => { + const fixture = await repository(); + const admitted = await source(fixture.root); + const storage = join(fixture.parent, 'instances'); + const id = '9'.repeat(64); + const child = spawn( + process.execPath, + [ + '--input-type=module', + '-e', + ` + import { GitWorktreeManager } from ${JSON.stringify( + new URL('./worktrees.js', import.meta.url).href + )}; + const manager = new GitWorktreeManager({ + maxCount: 1, root: ${JSON.stringify(storage)}, + sources: new Map([['primary', ${JSON.stringify(admitted)}]]), + prepareInstance: async () => { + process.stdout.write('setup-started'); + await new Promise(() => { setInterval(() => {}, 1000); }); + }, + }); + await manager.resolve('primary', ${JSON.stringify(id)}); + `, + ], + { stdio: ['ignore', 'pipe', 'pipe'] } + ); + const closed = once(child, 'close'); + t.after(async () => { + child.kill('SIGKILL'); + await closed; + await rm(fixture.parent, { recursive: true, force: true }); + }); + await once(child.stdout!, 'data'); + child.kill('SIGKILL'); + await closed; + const restarted = new GitWorktreeManager({ + maxCount: 1, + root: storage, + sources: new Map([['primary', admitted]]), + }); + await assert.rejects( + restarted.resolve('primary', id), + /operator recovery required/ + ); + await assert.rejects( + restarted.resolve('primary', '8'.repeat(64)), + /capacity is exhausted/ + ); + assert.equal( + await readFile( + join(await restarted.plannedRoot('primary', id), 'README.md'), + 'utf8' + ), + 'source\n' + ); + } +); + +test('cached checkouts revalidate their admitted source without deleting user work', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + const id = 'f'.repeat(64); + const instance = await manager.resolve('primary', id); + await writeFile(join(instance.root, 'pending.txt'), 'user work'); + await rename(fixture.root, `${fixture.root}.original`); + await mkdir(fixture.root); + await assert.rejects( + manager.resolve('primary', id), + /source changed after admission/ + ); + assert.equal( + await readFile(join(instance.root, 'pending.txt'), 'utf8'), + 'user work' + ); +}); + +test('preserves a failed setup checkout until executor cleanup is confirmed', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const options = { + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + prepareInstance: async () => { + throw new Error('setup failed'); + }, + discardInstance: async () => { + throw new Error('child cleanup unconfirmed'); + }, + }; + const manager = new GitWorktreeManager(options); + const id = 'a'.repeat(64); + await assert.rejects(manager.resolve('primary', id), /cleanup unconfirmed/); + const root = await manager.plannedRoot('primary', id); + assert.equal(await readFile(join(root, 'README.md'), 'utf8'), 'source\n'); + await assert.rejects( + new GitWorktreeManager(options).resolve('primary', id), + /operator recovery required/ + ); + await assert.rejects( + new GitWorktreeManager(options).resolve('primary', 'b'.repeat(64)), + /capacity is exhausted/ + ); + assert.equal(await readFile(join(root, 'README.md'), 'utf8'), 'source\n'); +}); + +test('cancellation waits for setup cleanup before releasing provisioning ownership', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + let started!: () => void; + const setupStarted = new Promise((resolve) => { + started = resolve; + }); + let cleanupFinished = false; + let setupRoot = ''; + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + prepareInstance: async (instance, signal) => { + const reservation = JSON.parse( + await readFile(`${instance.root}.complete`, 'utf8') + ); + assert.equal(reservation.provisioningFailed, true); + setupRoot = instance.root; + started(); + try { + await delay(60_000, undefined, { signal }); + await writeFile(join(instance.root, 'LATE'), 'should never happen'); + } finally { + await delay(20); + cleanupFinished = true; + } + }, + }); + const controller = new AbortController(); + const pending = manager.resolve('primary', 'a'.repeat(64), controller.signal); + const rejected = assert.rejects(pending, { name: 'AbortError' }); + await setupStarted; + controller.abort(); + await rejected; + assert.equal(cleanupFinished, true); + await assert.rejects(stat(setupRoot), { code: 'ENOENT' }); + await assert.rejects(stat(`${setupRoot}.complete`), { code: 'ENOENT' }); +}); + +test('cancels lock wait without provisioning while another caller continues', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + let started!: () => void; + const setupStarted = new Promise((resolve) => { + started = resolve; + }); + let release!: () => void; + const released = new Promise((resolve) => { + release = resolve; + }); + t.after(release); + let setups = 0; + const options = { + maxCount: 2, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + prepareInstance: async () => { + setups++; + started(); + await released; + }, + }; + const active = new GitWorktreeManager(options).resolve( + 'primary', + 'a'.repeat(64) + ); + await setupStarted; + const controller = new AbortController(); + const manager = new GitWorktreeManager(options); + const waiting = manager.resolve('primary', 'b'.repeat(64), controller.signal); + const rejected = assert.rejects(waiting, { name: 'AbortError' }); + await delay(75); + controller.abort(); + await rejected; + assert.equal(setups, 1); + release(); + await active; + await assert.rejects( + stat(await manager.plannedRoot('primary', 'b'.repeat(64))), + { code: 'ENOENT' } + ); +}); + +test('recovery preserves unknown directories and malformed completion records', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const options = { + maxCount: 4, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }; + const manager = new GitWorktreeManager(options); + const first = await manager.resolve('primary', 'a'.repeat(64)); + const unrelated = join(options.root, 'operator-backups', 'important'); + await mkdir(unrelated, { recursive: true }); + await writeFile(join(unrelated, 'notes'), 'keep'); + await manager.resolve('primary', 'b'.repeat(64)); + assert.equal(await readFile(join(unrelated, 'notes'), 'utf8'), 'keep'); + await writeFile(`${first.root}.complete`, '{"version":0}'); + await assert.rejects( + new GitWorktreeManager(options).resolve('primary', 'a'.repeat(64)), + /completion record is invalid/ + ); + await assert.rejects( + new GitWorktreeManager(options).resolve('primary', 'c'.repeat(64)), + /completion record is invalid/ + ); + assert.equal( + await readFile(join(first.root, 'README.md'), 'utf8'), + 'source\n' + ); +}); + +async function git(root: string, ...args: string[]): Promise { + const result = await execFileAsync('git', ['-C', root, ...args], { + encoding: 'utf8', + env: { + PATH: process.env.PATH, + GIT_CONFIG_NOSYSTEM: '1', + GIT_CONFIG_GLOBAL: '/dev/null', + GIT_TERMINAL_PROMPT: '0', + LC_ALL: 'C', + }, + }); + return result.stdout.trim(); +} + +async function repository( + objectFormat = 'sha1' +): Promise<{ parent: string; root: string }> { + const parent = await mkdtemp(join(tmpdir(), 'librechat-worktrees-')); + const root = join(parent, 'source'); + await execFileAsync('git', ['init', `--object-format=${objectFormat}`, root]); + await writeFile(join(root, 'README.md'), 'source\n'); + await git(root, 'add', 'README.md'); + await git( + root, + '-c', + 'user.name=Test', + '-c', + 'user.email=test@example.com', + 'commit', + '-m', + 'initial' + ); + return { parent, root: await realpath(root) }; +} + +async function source(root: string) { + return { root, identity: await captureWorkspaceRootIdentity(root) }; +} + +test('creates and reuses an isolated worktree for one conversation identity', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const manager = new GitWorktreeManager({ + maxCount: 4, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + const id = 'a'.repeat(64); + + const [first, concurrent] = await Promise.all([ + manager.resolve('primary', id), + manager.resolve('primary', id), + ]); + assert.deepEqual(concurrent, first); + assert.notEqual(first.root, fixture.root); + assert.equal( + (await realpath(join(first.root, '.git', 'objects'))).startsWith( + first.root + ), + true + ); + const instanceCommon = await realpath( + await git( + first.root, + 'rev-parse', + '--path-format=absolute', + '--git-common-dir' + ) + ); + assert.equal(instanceCommon.startsWith(first.root), true); + assert.equal( + await readFile(join(first.root, 'README.md'), 'utf8'), + 'source\n' + ); + + await writeFile(join(first.root, 'README.md'), 'conversation\n'); + assert.equal( + await readFile(join(fixture.root, 'README.md'), 'utf8'), + 'source\n' + ); + + const restarted = new GitWorktreeManager({ + maxCount: 4, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + assert.equal((await restarted.resolve('primary', id)).root, first.root); +}); + +test('replaces an incomplete checkout before admitting it after restart', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const storage = join(fixture.parent, 'instances'); + const id = 'c'.repeat(64); + const manager = new GitWorktreeManager({ + maxCount: 4, + root: storage, + sources: new Map([['primary', await source(fixture.root)]]), + }); + const first = await manager.resolve('primary', id); + await writeFile(join(first.root, 'README.md'), 'partial mutation\n'); + await rm(`${first.root}.complete`); + + const restarted = new GitWorktreeManager({ + maxCount: 4, + root: storage, + sources: new Map([['primary', await source(fixture.root)]]), + }); + const recovered = await restarted.resolve('primary', id); + assert.equal( + await readFile(join(recovered.root, 'README.md'), 'utf8'), + 'source\n' + ); +}); + +test('does not count an incomplete checkout against capacity after restart', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const storage = join(fixture.parent, 'instances'); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: storage, + sources: new Map([['primary', await source(fixture.root)]]), + }); + const abandoned = await manager.resolve('primary', 'c'.repeat(64)); + await rm(`${abandoned.root}.complete`); + const staleMarker = `${abandoned.root}.complete.1.tmp`; + await writeFile(staleMarker, '1\n'); + + const restarted = new GitWorktreeManager({ + maxCount: 1, + root: storage, + sources: new Map([['primary', await source(fixture.root)]]), + }); + const replacement = await restarted.resolve('primary', 'd'.repeat(64)); + assert.equal((await stat(replacement.root)).isDirectory(), true); + await assert.rejects(stat(abandoned.root), { code: 'ENOENT' }); + await assert.rejects(stat(staleMarker), { code: 'ENOENT' }); +}); + +test('keeps a conversation checkout independent of source object pruning', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + await writeFile(join(fixture.root, 'SECOND.md'), 'second\n'); + await git(fixture.root, 'add', 'SECOND.md'); + await git( + fixture.root, + '-c', + 'user.name=Test', + '-c', + 'user.email=test@example.com', + 'commit', + '-m', + 'second' + ); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + const instance = await manager.resolve('primary', 'e'.repeat(64)); + const retainedHead = await git(instance.root, 'rev-parse', 'HEAD'); + + await git(fixture.root, 'reset', '--hard', 'HEAD~1'); + await git(fixture.root, 'reflog', 'expire', '--expire=now', '--all'); + await git(fixture.root, 'gc', '--prune=now'); + + assert.equal(await git(instance.root, 'rev-parse', 'HEAD'), retainedHead); + assert.equal( + await readFile(join(instance.root, 'SECOND.md'), 'utf8'), + 'second\n' + ); + await assert.rejects( + readFile(join(instance.root, '.git', 'objects', 'info', 'alternates')), + { code: 'ENOENT' } + ); +}); + +test('dissociates a checkout from inherited source alternates', async (t) => { + const upstream = await repository(); + const sharedParent = await mkdtemp( + join(tmpdir(), 'librechat-shared-source-') + ); + const sharedRoot = join(sharedParent, 'source'); + t.after(() => + Promise.all([ + rm(upstream.parent, { recursive: true, force: true }), + rm(sharedParent, { recursive: true, force: true }), + ]) + ); + await execFileAsync('git', ['clone', '--shared', upstream.root, sharedRoot]); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(sharedParent, 'instances'), + sources: new Map([['primary', await source(await realpath(sharedRoot))]]), + }); + const instance = await manager.resolve('primary', 'f'.repeat(64)); + await rm(upstream.root, { recursive: true, force: true }); + + assert.equal( + await git(instance.root, 'rev-parse', 'HEAD^{commit}'), + await git(instance.root, 'rev-parse', 'HEAD') + ); + await assert.rejects( + readFile(join(instance.root, '.git', 'objects', 'info', 'alternates')), + { code: 'ENOENT' } + ); +}); + +test('provisions an orphan branch for a repository with an unborn HEAD', async (t) => { + const parent = await mkdtemp(join(tmpdir(), 'librechat-empty-source-')); + const root = join(parent, 'source'); + await execFileAsync('git', ['init', root]); + t.after(() => rm(parent, { recursive: true, force: true })); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(parent, 'instances'), + sources: new Map([['primary', await source(await realpath(root))]]), + }); + + const instance = await manager.resolve('primary', '0'.repeat(64)); + assert.match( + await git(instance.root, 'branch', '--show-current'), + /^librechat\/conversation-/ + ); + await assert.rejects(git(instance.root, 'rev-parse', '--verify', 'HEAD')); +}); + +test('rejects replacement of the admitted worktree storage root', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const storage = join(fixture.parent, 'instances'); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: storage, + sources: new Map([['primary', await source(fixture.root)]]), + }); + await manager.resolve('primary', '1'.repeat(64)); + await rename(storage, `${storage}.original`); + await mkdir(storage, { mode: 0o700 }); + + await assert.rejects( + manager.resolve('primary', '1'.repeat(64)), + /storage changed after admission/ + ); +}); + +test('keeps conversations and source repositories isolated', async (t) => { + const first = await repository(); + const second = await repository(); + t.after(() => + Promise.all([ + rm(first.parent, { recursive: true, force: true }), + rm(second.parent, { recursive: true, force: true }), + ]) + ); + const storage = await mkdtemp(join(tmpdir(), 'librechat-worktree-storage-')); + t.after(() => rm(storage, { recursive: true, force: true })); + const manager = new GitWorktreeManager({ + maxCount: 4, + root: storage, + sources: new Map([ + ['first', await source(first.root)], + ['second', await source(second.root)], + ]), + }); + + const firstConversation = await manager.resolve('first', '1'.repeat(64)); + const secondConversation = await manager.resolve('first', '2'.repeat(64)); + const otherRepository = await manager.resolve('second', '1'.repeat(64)); + assert.equal( + new Set([ + firstConversation.root, + secondConversation.root, + otherRepository.root, + ]).size, + 3 + ); +}); + +test('rejects invalid identities, overlapping storage and exhausted capacity', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + assert.throws( + () => + new GitWorktreeManager({ + cloneTimeoutMs: 29_999, + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([ + [ + 'primary', + { + root: fixture.root, + identity: { + path: fixture.root, + dev: '1', + ino: '1', + }, + }, + ], + ]), + }), + /clone timeout/ + ); + const overlapping = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.root, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + await assert.rejects( + overlapping.resolve('primary', 'a'.repeat(64)), + /must not overlap/ + ); + + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }); + await assert.rejects(manager.resolve('primary', '../escape'), /SHA-256/); + const first = await manager.resolve('primary', 'a'.repeat(64)); + assert.equal((await stat(first.root)).isDirectory(), true); + await assert.rejects( + manager.resolve('primary', 'b'.repeat(64)), + /capacity is exhausted/ + ); +}); + +test('serializes provisioning across manager instances sharing storage', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const options = { + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + }; + const results = await Promise.allSettled([ + new GitWorktreeManager(options).resolve('primary', 'a'.repeat(64)), + new GitWorktreeManager(options).resolve('primary', 'b'.repeat(64)), + ]); + assert.equal( + results.filter((result) => result.status === 'fulfilled').length, + 1 + ); + assert.equal( + results.filter((result) => result.status === 'rejected').length, + 1 + ); + assert.match( + ( + results.find( + (result) => result.status === 'rejected' + ) as PromiseRejectedResult + ).reason.message, + /capacity is exhausted/ + ); +}); + +test('prepares a new checkout before publishing its completion marker', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + let attempts = 0; + const options = { + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([['primary', await source(fixture.root)]]), + prepareInstance: async (instance: { root: string }) => { + attempts += 1; + if (attempts === 1) throw new Error('setup failed'); + await writeFile(join(instance.root, 'prepared'), 'yes\n'); + }, + }; + const id = 'c'.repeat(64); + await assert.rejects( + new GitWorktreeManager(options).resolve('primary', id), + /setup failed/ + ); + const instance = await new GitWorktreeManager(options).resolve('primary', id); + assert.equal( + await readFile(join(instance.root, 'prepared'), 'utf8'), + 'yes\n' + ); + assert.equal(attempts, 2); +}); + +test('preserves a completed checkout when its admitted source changes', async (t) => { + const first = await repository(); + const second = await repository(); + t.after(() => rm(first.parent, { recursive: true, force: true })); + t.after(() => rm(second.parent, { recursive: true, force: true })); + await writeFile(join(second.root, 'README.md'), 'replacement\n'); + await git(second.root, 'add', 'README.md'); + await git( + second.root, + '-c', + 'user.name=Test', + '-c', + 'user.email=test@example.com', + 'commit', + '-m', + 'replacement' + ); + const storage = join(first.parent, 'instances'); + const id = 'e'.repeat(64); + const original = await new GitWorktreeManager({ + maxCount: 1, + root: storage, + sources: new Map([['primary', await source(first.root)]]), + }).resolve('primary', id); + await writeFile(join(original.root, 'UNCOMMITTED.md'), 'user work\n'); + await assert.rejects( + new GitWorktreeManager({ + maxCount: 1, + root: storage, + sources: new Map([['primary', await source(second.root)]]), + }).resolve('primary', id), + /source identity changed/ + ); + assert.equal( + await readFile(join(original.root, 'UNCOMMITTED.md'), 'utf8'), + 'user work\n' + ); + assert.equal( + await readFile(join(original.root, 'README.md'), 'utf8'), + 'source\n' + ); +}); + +test('rejects a source whose admitted filesystem identity changed', async (t) => { + const fixture = await repository(); + t.after(() => rm(fixture.parent, { recursive: true, force: true })); + const metadata = await stat(fixture.root, { bigint: true }); + const manager = new GitWorktreeManager({ + maxCount: 1, + root: join(fixture.parent, 'instances'), + sources: new Map([ + [ + 'primary', + { + root: fixture.root, + identity: { + path: fixture.root, + dev: metadata.dev.toString(), + ino: (metadata.ino + 1n).toString(), + }, + }, + ], + ]), + }); + + await assert.rejects( + manager.resolve('primary', 'd'.repeat(64)), + /source changed after admission/ + ); +}); diff --git a/packages/code/src/worktrees.ts b/packages/code/src/worktrees.ts new file mode 100644 index 00000000..ab19aa38 --- /dev/null +++ b/packages/code/src/worktrees.ts @@ -0,0 +1,718 @@ +import { execFile } from 'node:child_process'; +import { createHash, randomUUID } from 'node:crypto'; +import { + lstat, + mkdir, + readFile, + readdir, + realpath, + rename, + rm, + stat, + writeFile, +} from 'node:fs/promises'; +import { isAbsolute, join, relative, resolve, sep } from 'node:path'; +import { promisify } from 'node:util'; + +import { matchesWorkspaceRoot } from './root-identity.js'; +import type { WorkspaceRootIdentity } from './root-identity.js'; +import { assertPrivateStorageAncestors } from './private-storage.js'; +import { withProcessLock } from './process-lock.js'; +import { GitSourceSnapshot } from './git-snapshot.js'; + +const execFileAsync = promisify(execFile); +const WORKTREE_INSTANCE_PATTERN = /^[a-f0-9]{64}$/; +const COMPLETION_TEMP_PATTERN = /^[a-f0-9]{64}\.complete\.[a-f0-9-]+\.tmp$/; +const GIT_TIMEOUT_MS = 30_000; +const DEFAULT_CLONE_TIMEOUT_MS = 5 * 60_000; + +export interface GitWorktreeSource { + identity: WorkspaceRootIdentity; + root: string; +} + +export interface GitWorktreeInstance { + id: string; + identity: WorkspaceRootIdentity; + root: string; + sourceWorkspaceId: string; +} + +export interface GitWorktreeManagerOptions { + cloneTimeoutMs?: number; + maxCount: number; + root: string; + sources: ReadonlyMap; + prepareInstance?: ( + instance: GitWorktreeInstance, + signal?: AbortSignal + ) => Promise; + discardInstance?: (instance: GitWorktreeInstance) => Promise | void; +} + +const PROVISIONING_LOCK = '.provision.lock'; + +function isInside(parent: string, candidate: string): boolean { + const path = relative(parent, candidate); + return ( + path === '' || + (path !== '..' && !path.startsWith(`..${sep}`) && !isAbsolute(path)) + ); +} + +function gitEnvironment(): NodeJS.ProcessEnv { + return { + PATH: process.env.PATH, + SYSTEMROOT: process.env.SYSTEMROOT, + GIT_CONFIG_NOSYSTEM: '1', + GIT_CONFIG_GLOBAL: '/dev/null', + GIT_TERMINAL_PROMPT: '0', + GIT_OPTIONAL_LOCKS: '0', + LC_ALL: 'C', + }; +} + +async function git( + root: string, + args: string[], + signal?: AbortSignal, + timeout = GIT_TIMEOUT_MS +): Promise { + const execution = execFileAsync( + 'git', + ['--no-optional-locks', '-C', root, ...args], + { + encoding: 'utf8', + env: gitEnvironment(), + maxBuffer: 16 * 1024, + signal, + timeout, + } + ); + const closed = new Promise((resolve) => + execution.child.once('close', () => resolve()) + ); + try { + return (await execution).stdout.trim(); + } finally { + // execFile's AbortError callback can run before its child exits. Retain the + // provisioning lock and directory until the writer is actually gone. + const killTimer = setTimeout(() => execution.child.kill('SIGKILL'), 1000); + killTimer.unref(); + try { + await closed; + } finally { + clearTimeout(killTimer); + } + } +} + +async function sourceConfig( + root: string, + key: string, + signal?: AbortSignal +): Promise { + try { + const remote = await git( + root, + [ + 'config', + '--no-includes', + '--file', + join(root, 'source-config'), + '--get', + key, + ], + signal + ); + return remote || undefined; + } catch { + signal?.throwIfAborted(); + return undefined; + } +} + +async function hasCommittedHead( + root: string, + signal?: AbortSignal +): Promise { + try { + await git(root, ['rev-parse', '--verify', 'HEAD'], signal); + return true; + } catch (error) { + signal?.throwIfAborted(); + if (error instanceof Error && 'code' in error && error.code === 128) { + return false; + } + throw error; + } +} + +async function directoryIdentity(path: string): Promise { + const metadata = await lstat(path, { bigint: true }); + if (!metadata.isDirectory() || metadata.isSymbolicLink()) { + throw new Error('Conversation worktree must be a real directory'); + } + return { + path, + dev: metadata.dev.toString(), + ino: metadata.ino.toString(), + }; +} + +export class GitWorktreeManager { + private readonly instances = new Map(); + private readonly sourceSnapshots = new Map< + string, + Promise + >(); + private canonicalRoot?: Promise<{ + identity: WorkspaceRootIdentity; + path: string; + }>; + + constructor(private readonly options: GitWorktreeManagerOptions) { + if ( + !Number.isSafeInteger(options.maxCount) || + options.maxCount < 1 || + options.maxCount > 1024 || + options.sources.size === 0 + ) { + throw new Error( + 'Conversation worktree capacity must be between 1 and 1024' + ); + } + if ( + options.cloneTimeoutMs !== undefined && + (!Number.isSafeInteger(options.cloneTimeoutMs) || + options.cloneTimeoutMs < GIT_TIMEOUT_MS || + options.cloneTimeoutMs > 30 * 60_000) + ) { + throw new Error( + 'Conversation worktree clone timeout must be between 30000 and 1800000 milliseconds' + ); + } + } + + private async root(): Promise { + this.canonicalRoot ??= (async () => { + const configuredRoot = resolve(this.options.root); + await assertPrivateStorageAncestors(configuredRoot, true); + await mkdir(configuredRoot, { + mode: 0o700, + recursive: true, + }); + await assertPrivateStorageAncestors(configuredRoot); + const root = await realpath(this.options.root); + await assertPrivateStorageAncestors(root); + const metadata = await stat(root); + if ( + !metadata.isDirectory() || + (process.platform !== 'win32' && (metadata.mode & 0o022) !== 0) + ) { + throw new Error( + 'Conversation worktree root must not be group or world writable' + ); + } + for (const source of this.options.sources.values()) { + const sourceRoot = await realpath(source.root); + if (isInside(sourceRoot, root) || isInside(root, sourceRoot)) { + throw new Error( + 'Conversation worktree storage must not overlap a source workspace' + ); + } + } + return { + identity: await directoryIdentity(root), + path: root, + }; + })(); + const root = await this.canonicalRoot; + if (!(await matchesWorkspaceRoot(root.path, root.identity))) { + throw new Error('Conversation worktree storage changed after admission'); + } + return root.path; + } + + private key(sourceWorkspaceId: string, instanceId: string): string { + return `${sourceWorkspaceId}\0${instanceId}`; + } + + private branch(sourceWorkspaceId: string, instanceId: string): string { + const source = createHash('sha256') + .update(sourceWorkspaceId) + .digest('hex') + .slice(0, 8); + return `librechat/conversation-${source}-${instanceId.slice(0, 31)}`; + } + + private async instancePath( + sourceWorkspaceId: string, + instanceId: string + ): Promise { + const sourceDirectory = createHash('sha256') + .update(sourceWorkspaceId) + .digest('hex') + .slice(0, 24); + return join(await this.root(), sourceDirectory, instanceId); + } + + async plannedRoot( + sourceWorkspaceId: string, + instanceId: string + ): Promise { + if (!WORKTREE_INSTANCE_PATTERN.test(instanceId)) { + throw new Error( + 'Conversation worktree identity must be a SHA-256 digest' + ); + } + if (!this.options.sources.has(sourceWorkspaceId)) { + throw new Error('Conversation worktree source is unavailable'); + } + return await this.instancePath(sourceWorkspaceId, instanceId); + } + + async prepare(): Promise { + await this.root(); + await Promise.all( + [...this.options.sources].map(async ([_workspaceId, source]) => { + const sourceRoot = await this.admittedSourceRoot(source); + await this.sourceSnapshot(sourceRoot, source); + }) + ); + } + + private async admittedSourceRoot(source: GitWorktreeSource): Promise { + const sourceRoot = await realpath(source.root); + if (!(await matchesWorkspaceRoot(sourceRoot, source.identity))) { + throw new Error('Conversation worktree source changed after admission'); + } + return sourceRoot; + } + + private async sourceSnapshot( + root: string, + source: GitWorktreeSource + ): Promise { + let snapshot = this.sourceSnapshots.get(root); + if (!snapshot) { + snapshot = GitSourceSnapshot.admit(source.identity); + this.sourceSnapshots.set(root, snapshot); + } + const admitted = await snapshot; + await admitted.validate(); + return admitted; + } + + private async countInstances(): Promise { + const root = await this.root(); + const sourceDirectories = await readdir(root, { withFileTypes: true }); + let count = 0; + for (const sourceDirectory of sourceDirectories) { + if (sourceDirectory.name.startsWith(PROVISIONING_LOCK)) continue; + if ( + !/^[a-f0-9]{24}$/.test(sourceDirectory.name) || + !sourceDirectory.isDirectory() || + sourceDirectory.isSymbolicLink() + ) + continue; + const entries = await readdir(join(root, sourceDirectory.name), { + withFileTypes: true, + }); + const reserved = new Set(); + for (const entry of entries) { + const id = entry.name.endsWith('.complete') + ? entry.name.slice(0, -9) + : ''; + if (!WORKTREE_INSTANCE_PATTERN.test(id)) continue; + if (!entry.isFile() || entry.isSymbolicLink()) + throw new Error('Invalid worktree reservation'); + if ( + await this.hasCompletionMarker(join(root, sourceDirectory.name, id)) + ) { + reserved.add(id); + count += 1; + } + } + for (const entry of entries) { + if (entry.isFile() && COMPLETION_TEMP_PATTERN.test(entry.name)) { + await rm(join(root, sourceDirectory.name, entry.name), { + force: true, + }); + continue; + } + if ( + !WORKTREE_INSTANCE_PATTERN.test(entry.name) || + !entry.isDirectory() || + entry.isSymbolicLink() + ) + continue; + const path = join(root, sourceDirectory.name, entry.name); + if (!reserved.has(entry.name)) { + await rm(path, { recursive: true, force: true }); + await rm(this.completionMarker(path), { force: true }); + } + } + } + return count; + } + + private async withProvisioningLock( + operation: () => Promise, + signal?: AbortSignal + ): Promise { + return await withProcessLock( + join(await this.root(), PROVISIONING_LOCK), + operation, + signal + ); + } + + private completionMarker(path: string): string { + return `${path}.complete`; + } + + private async hasCompletionMarker( + path: string, + source?: WorkspaceRootIdentity, + sourceGit?: string + ): Promise { + try { + const record = JSON.parse( + await readFile(this.completionMarker(path), 'utf8') + ) as { + version?: unknown; + source?: Partial; + provisioningFailed?: boolean; + sourceGit?: string; + }; + const valid = + record.version === 2 && + typeof record.sourceGit === 'string' && + WORKTREE_INSTANCE_PATTERN.test(record.sourceGit) && + typeof record.source?.path === 'string' && + typeof record.source.dev === 'string' && + typeof record.source.ino === 'string'; + if (!valid) + throw new Error( + 'Conversation worktree completion record is invalid; existing checkout preserved' + ); + if ( + source != null && + (record.source!.path !== source.path || + record.source!.dev !== source.dev || + record.source!.ino !== source.ino || + record.sourceGit !== sourceGit) + ) { + throw new Error( + 'Conversation worktree source identity changed; existing checkout preserved' + ); + } + if (source != null && record.provisioningFailed) { + throw new Error( + 'Conversation worktree setup cleanup is unconfirmed; operator recovery required' + ); + } + return true; + } catch (error) { + if ( + error instanceof Error && + 'code' in error && + error.code === 'ENOENT' + ) { + return false; + } + throw error; + } + } + + private async writeCompletionMarker( + path: string, + source: WorkspaceRootIdentity, + sourceGit: string, + provisioningFailed = false + ): Promise { + const marker = this.completionMarker(path); + const temporary = `${marker}.${randomUUID()}.tmp`; + try { + await writeFile( + temporary, + `${JSON.stringify({ + version: 2, + source, + sourceGit, + ...(provisioningFailed ? { provisioningFailed: true } : {}), + })}\n`, + { mode: 0o600, flag: 'wx' } + ); + await rename(temporary, marker); + } finally { + await rm(temporary, { force: true }); + } + } + + private async validateRepository( + sourceWorkspaceId: string, + instanceId: string, + path: string, + signal?: AbortSignal + ): Promise { + const canonicalPath = await realpath(path); + if (canonicalPath !== path || !isInside(await this.root(), canonicalPath)) { + throw new Error( + 'Conversation worktree escaped its configured storage root' + ); + } + signal?.throwIfAborted(); + const instanceCommon = join(canonicalPath, '.git'); + const gitMetadata = await lstat(instanceCommon); + if ( + !gitMetadata.isDirectory() || + gitMetadata.isSymbolicLink() || + (await realpath(instanceCommon)) !== instanceCommon + ) { + throw new Error('Conversation worktree does not own its Git metadata'); + } + try { + await lstat(join(instanceCommon, 'commondir')); + throw new Error( + 'Conversation worktree must not redirect its Git metadata' + ); + } catch (error) { + if ( + !(error instanceof Error) || + !('code' in error) || + error.code !== 'ENOENT' + ) + throw error; + } + const instanceObjects = await realpath(join(instanceCommon, 'objects')); + if (!isInside(canonicalPath, instanceObjects)) { + throw new Error('Conversation worktree does not own its Git objects'); + } + try { + await lstat(join(instanceObjects, 'info', 'alternates')); + throw new Error( + 'Conversation worktree must not use external Git objects' + ); + } catch (error) { + if ( + !(error instanceof Error) || + !('code' in error) || + error.code !== 'ENOENT' + ) { + throw error; + } + } + return { + id: instanceId, + identity: await directoryIdentity(canonicalPath), + root: canonicalPath, + sourceWorkspaceId, + }; + } + + private async validateExisting( + sourceWorkspaceId: string, + instanceId: string, + path: string, + source: WorkspaceRootIdentity, + sourceGit: string, + signal?: AbortSignal + ): Promise { + if (!(await this.hasCompletionMarker(path, source, sourceGit))) { + const error = new Error('Conversation worktree is incomplete'); + Object.assign(error, { code: 'EINCOMPLETE' }); + throw error; + } + return await this.validateRepository( + sourceWorkspaceId, + instanceId, + path, + signal + ); + } + + private async createLocked( + sourceWorkspaceId: string, + instanceId: string, + signal?: AbortSignal + ): Promise { + if (!WORKTREE_INSTANCE_PATTERN.test(instanceId)) { + throw new Error( + 'Conversation worktree identity must be a SHA-256 digest' + ); + } + const source = this.options.sources.get(sourceWorkspaceId); + if (!source) throw new Error('Conversation worktree source is unavailable'); + const sourceRoot = await this.admittedSourceRoot(source); + const snapshot = await this.sourceSnapshot(sourceRoot, source); + const path = await this.instancePath(sourceWorkspaceId, instanceId); + try { + return await this.validateExisting( + sourceWorkspaceId, + instanceId, + path, + source.identity, + snapshot.fingerprint, + signal + ); + } catch (error) { + if (!(error instanceof Error) || !('code' in error)) { + throw error; + } + if (error.code === 'EINCOMPLETE') { + await rm(path, { recursive: true, force: true }); + await rm(this.completionMarker(path), { force: true }); + } else if (error.code !== 'ENOENT') { + throw error; + } + } + if ((await this.countInstances()) >= this.options.maxCount) { + throw new Error('Conversation worktree capacity is exhausted'); + } + await mkdir(resolve(path, '..'), { mode: 0o700, recursive: true }); + const branch = this.branch(sourceWorkspaceId, instanceId); + let instance: GitWorktreeInstance | undefined; + const staging = `${path}.source`; + try { + // Reserve before launching any writer. A worker crash may leave a Git + // child or setup executor alive after the parent's kernel lock releases. + // Recovery must not sweep or reuse that uncertain directory. + await this.writeCompletionMarker( + path, + source.identity, + snapshot.fingerprint, + true + ); + const cloneSignal = AbortSignal.any([ + ...(signal ? [signal] : []), + AbortSignal.timeout( + this.options.cloneTimeoutMs ?? DEFAULT_CLONE_TIMEOUT_MS + ), + ]); + await snapshot.copyTo(staging, cloneSignal); + const remote = await sourceConfig( + staging, + 'remote.origin.url', + cloneSignal + ); + const objectFormat = await sourceConfig( + staging, + 'extensions.objectformat', + cloneSignal + ); + if ( + objectFormat && + objectFormat !== 'sha1' && + objectFormat !== 'sha256' + ) { + throw new Error('Unsupported source Git object format'); + } + if (objectFormat === 'sha256') { + await writeFile( + join(staging, 'config'), + '[core]\nrepositoryformatversion = 1\nbare = true\n[extensions]\nobjectformat = sha256\n', + { mode: 0o600 } + ); + } + await git( + resolve(path, '..'), + ['clone', '--local', '--no-checkout', '--no-tags', staging, path], + cloneSignal, + this.options.cloneTimeoutMs ?? DEFAULT_CLONE_TIMEOUT_MS + ); + await rm(staging, { recursive: true, force: true }); + const sourceHasHead = await hasCommittedHead(path, signal); + if (remote) { + await git(path, ['remote', 'set-url', 'origin', remote], signal); + } else { + await git(path, ['remote', 'remove', 'origin'], signal); + } + await git( + path, + sourceHasHead + ? ['checkout', '--force', '-b', branch, 'HEAD'] + : ['checkout', '--orphan', branch], + signal + ); + instance = await this.validateRepository( + sourceWorkspaceId, + instanceId, + path, + signal + ); + await this.options.prepareInstance?.(instance, signal); + signal?.throwIfAborted(); + if (!(await matchesWorkspaceRoot(instance.root, instance.identity))) { + throw new Error('Conversation worktree changed during setup'); + } + await this.validateRepository( + sourceWorkspaceId, + instanceId, + path, + signal + ); + await this.admittedSourceRoot(source); + await snapshot.validate(); + await this.writeCompletionMarker( + path, + source.identity, + snapshot.fingerprint + ); + return instance; + } catch (error) { + if (instance) { + // The reservation remains until executor cleanup is confirmed. + await this.options.discardInstance?.(instance); + } + await rm(path, { recursive: true, force: true }); + await rm(staging, { recursive: true, force: true }); + await rm(this.completionMarker(path), { force: true }); + throw error; + } + } + + private async create( + sourceWorkspaceId: string, + instanceId: string, + signal?: AbortSignal + ): Promise { + return await this.withProvisioningLock( + () => this.createLocked(sourceWorkspaceId, instanceId, signal), + signal + ); + } + + async resolve( + sourceWorkspaceId: string, + instanceId: string, + signal?: AbortSignal + ): Promise { + signal?.throwIfAborted(); + const key = this.key(sourceWorkspaceId, instanceId); + const cached = this.instances.get(key); + if (cached) { + await this.root(); + const source = this.options.sources.get(sourceWorkspaceId)!; + await this.sourceSnapshot(await this.admittedSourceRoot(source), source); + if (!(await matchesWorkspaceRoot(cached.root, cached.identity))) { + this.instances.delete(key); + throw new Error('Conversation worktree changed after admission'); + } + await this.validateRepository( + sourceWorkspaceId, + instanceId, + cached.root, + signal + ); + return cached; + } + // The same kernel lock coordinates callers and processes. Keep cancellation + // attached through setup and cleanup; never release a lease while detached + // provisioning is still mutating the checkout. + const instance = await this.create(sourceWorkspaceId, instanceId, signal); + this.instances.set(key, instance); + return instance; + } +} diff --git a/service/src/bridge/concurrent-store.test.ts b/service/src/bridge/concurrent-store.test.ts index 736b7216..33f3d082 100644 --- a/service/src/bridge/concurrent-store.test.ts +++ b/service/src/bridge/concurrent-store.test.ts @@ -2,7 +2,10 @@ import { afterEach, expect, test } from 'bun:test'; import RedisMock from 'ioredis-mock'; import type Redis from 'ioredis'; import { RedisBridgeStore } from './store'; -import { BRIDGE_PROTOCOL_VERSION } from '../../../packages/code/src/protocol'; +import { + BRIDGE_PROTOCOL_VERSION, + workspaceIsolationKey, +} from '../../../packages/code/src/protocol'; import type { CodeBridgeAssignment } from './store'; const redis = new RedisMock() as unknown as Redis; @@ -26,13 +29,20 @@ async function register(workspaceLeaseSlots = 2) { workspaceTools: { protocolVersion: BRIDGE_PROTOCOL_VERSION, operations: ['read_file'], - workspaces: [{ id: 'a' }, { id: 'b' }], + workspaces: [ + { id: 'a', workspaceInstances: ['git_worktree'] }, + { id: 'b' }, + ], }, }, }); await store.confirmReady(workerId, incarnationId, generation); } -function dispatch(workspaceId: string, signal = new AbortController().signal) { +function dispatch( + workspaceId: string, + signal = new AbortController().signal, + workspaceInstanceId?: string, +) { const promise = store.dispatchWorkspaceTool({ workerId, signal, @@ -41,6 +51,7 @@ function dispatch(workspaceId: string, signal = new AbortController().signal) { protocolVersion: BRIDGE_PROTOCOL_VERSION, operation: 'read_file', workspaceId, + ...(workspaceInstanceId == null ? {} : { workspaceInstanceId }), path: 'file.txt', }, }); @@ -346,6 +357,52 @@ test('same-root work waits while another root progresses', async () => { await Promise.all([nextA, b]); }); +test('conversation worktrees on one source use independent capacity lanes', async () => { + await register(); + const firstId = 'a'.repeat(64); + const secondId = 'b'.repeat(64); + const firstPending = dispatch('a', undefined, firstId); + const first = (await store.lease( + workerId, + incarnationId, + 1000, + undefined, + undefined, + 0, + ))!; + const samePending = dispatch('a', undefined, firstId); + const secondPending = dispatch('a', undefined, secondId); + const second = (await store.lease( + workerId, + incarnationId, + 1000, + undefined, + undefined, + 1, + ))!; + expect(second.request).toMatchObject({ + workspaceId: 'a', + workspaceInstanceId: secondId, + }); + await settle(first); + await firstPending; + const same = (await store.lease( + workerId, + incarnationId, + 1000, + undefined, + undefined, + 0, + ))!; + expect(same.request).toMatchObject({ + workspaceId: 'a', + workspaceInstanceId: firstId, + }); + await settle(second); + await settle(same); + await Promise.all([samePending, secondPending]); +}); + test('queued cancellation never leases and does not block another root', async () => { await register(); const a = dispatch('a'); @@ -548,7 +605,11 @@ test('post-settlement fences are authenticated, idempotent, and invalidated by r await expect(dispatch('a')).rejects.toMatchObject({ code: 'WORKSPACE_QUARANTINED', }); - await store.resetWorkspace(workerId, incarnationId, 'native-workspace:a'); + await store.resetWorkspace( + workerId, + incarnationId, + `native-workspace:${workspaceIsolationKey('a')}`, + ); await expect( store.settle( workerId, diff --git a/service/src/bridge/router.ts b/service/src/bridge/router.ts index 73870772..5b319733 100644 --- a/service/src/bridge/router.ts +++ b/service/src/bridge/router.ts @@ -430,6 +430,7 @@ router.post( supportedWorkspaceEditFileFeatures: ['expected_base_sha256'], supportedWorkspaceListFileFeatures: ['after_path'], supportedWorkspaceProgrammaticLanguages: ['bash'], + supportedWorkspaceInstanceTypes: ['git_worktree'], }); } catch (error) { if (error instanceof BridgeStoreError) { diff --git a/service/src/bridge/store.test.ts b/service/src/bridge/store.test.ts index 9d271551..354108af 100644 --- a/service/src/bridge/store.test.ts +++ b/service/src/bridge/store.test.ts @@ -4,7 +4,7 @@ import RedisMock from 'ioredis-mock'; import type Redis from 'ioredis'; import type * as t from '../types'; import { BRIDGE_PROTOCOL_VERSION } from '../../../packages/code/src/protocol'; -import { RedisBridgeStore } from './store'; +import { RedisBridgeStore, workspaceAdmissionId } from './store'; import type { RegisteredBridgeWorker } from './store'; @@ -27,6 +27,16 @@ afterEach(async () => { }); describe('RedisBridgeStore', () => { + test('uses disjoint admission identities for roots and worktree instances', () => { + const instanceId = 'a'.repeat(64); + expect( + workspaceAdmissionId(`foo:git-worktree:${instanceId}`), + ).not.toBe(workspaceAdmissionId('foo', instanceId)); + expect(workspaceAdmissionId('foo')).not.toBe( + workspaceAdmissionId('workspace:foo'), + ); + }); + test('reports an atomic, capability-limited worker status snapshot', async () => { const store = new RedisBridgeStore(redis); const capabilities = { diff --git a/service/src/bridge/store.ts b/service/src/bridge/store.ts index 4eaabd7f..2ffeec66 100644 --- a/service/src/bridge/store.ts +++ b/service/src/bridge/store.ts @@ -15,8 +15,9 @@ import { BRIDGE_PROTOCOL_VERSION, isValidBridgeWorkerCapabilities, isValidBridgeWorkerId, - isWorkspaceToolRequest, - isWorkspaceToolResult, + isWorkspaceToolRequest, + isWorkspaceToolResult, + workspaceIsolationKey, } from '../../../packages/code/src/protocol'; import type { BridgeWorkerBinding } from './pairing'; import { BridgeAdmissionQueue } from './admission'; @@ -124,6 +125,12 @@ function supportsWorkspaceTool( if (!supportsOperation) { return supportsOperation; } + if ( + request.workspaceInstanceId !== undefined && + workspace.workspaceInstances?.includes('git_worktree') !== true + ) { + return false; + } if (request.operation === 'list_files' && request.afterPath !== undefined) { return capabilities?.listFileFeatures?.includes('after_path') === true; } @@ -156,6 +163,7 @@ function supportsWorkspaceProgrammatic( registration: RegisteredBridgeWorker, workspaceId: string, language: string, + workspaceInstanceId?: string, ): boolean { const capabilities = registration.capabilities.workspaceTools; const workspace = capabilities?.workspaces.find( @@ -163,6 +171,8 @@ function supportsWorkspaceProgrammatic( ); return ( workspace != null && + (workspaceInstanceId === undefined || + workspace.workspaceInstances?.includes('git_worktree') === true) && capabilities?.operations.includes('execute_command') === true && (workspace.operations == null || workspace.operations.includes('execute_command')) && @@ -172,6 +182,25 @@ function supportsWorkspaceProgrammatic( ); } +function workspaceInstanceId(body: t.PayloadBody): string | undefined { + if ( + typeof body === 'object' && + body != null && + 'workspace_instance_id' in body && + typeof body.workspace_instance_id === 'string' + ) { + return body.workspace_instance_id; + } + return undefined; +} + +export function workspaceAdmissionId( + workspaceId: string, + instanceId?: string, +): string { + return workspaceIsolationKey(workspaceId, instanceId); +} + function workerKey(workerId: string): string { return `${PREFIX}:worker:${encodeURIComponent(workerId)}`; } @@ -807,6 +836,7 @@ export class RedisBridgeStore { registration, args.workspaceId, args.body.language, + workspaceInstanceId(args.body), ) ) { throw new BridgeStoreError( @@ -843,6 +873,15 @@ export class RedisBridgeStore { let workspaceLeaseSlot: number | undefined; const selectedWorkspaceId = args.workspaceRequest?.workspaceId ?? args.workspaceId; + const selectedWorkspaceInstanceId = + args.workspaceRequest?.workspaceInstanceId ?? workspaceInstanceId(args.body); + const selectedWorkspaceAdmissionId = + selectedWorkspaceId == null + ? undefined + : workspaceAdmissionId( + selectedWorkspaceId, + selectedWorkspaceInstanceId, + ); const workspaceSlots = selectedWorkspaceId != null && (registration.capabilities.workspaceLeaseSlots ?? 1) > 1 @@ -864,7 +903,7 @@ export class RedisBridgeStore { args.deadlineAtMs, workspaceSlots == null ? undefined - : selectedWorkspaceId, + : selectedWorkspaceAdmissionId, ), args, 'Bridge admission enqueue', @@ -899,7 +938,7 @@ export class RedisBridgeStore { workerId: args.workerId, incarnationId: lockIncarnationId, assignmentId, - workspaceId: selectedWorkspaceId!, + workspaceId: selectedWorkspaceAdmissionId!, capacity: registration.capabilities.workspaceLeaseSlots!, expiresAtMs: Date.now() + ttlSeconds * 1000, }), @@ -961,6 +1000,7 @@ export class RedisBridgeStore { current.registration, args.workspaceId, args.body.language, + workspaceInstanceId(args.body), )) ) { throw new BridgeStoreError( @@ -986,14 +1026,14 @@ export class RedisBridgeStore { generation, leaseToken, leaseTokenHash: tokenHash(leaseToken), - ...(selectedWorkspaceId == null ? {} : { - workspaceFence: `native-workspace:${selectedWorkspaceId}`, + ...(selectedWorkspaceAdmissionId == null ? {} : { + workspaceFence: `native-workspace:${selectedWorkspaceAdmissionId}`, }), ...(workspaceLeaseSlot === undefined ? {} : { workspaceLeaseSlot, - workspaceFence: `native-workspace:${selectedWorkspaceId!}`, + workspaceFence: `native-workspace:${selectedWorkspaceAdmissionId!}`, }), ...(registration.identityId != null ? { workerIdentityId: registration.identityId } @@ -1082,6 +1122,7 @@ export class RedisBridgeStore { replacement.registration, args.workspaceId, args.body.language, + workspaceInstanceId(args.body), ) ) { throw new BridgeStoreError( diff --git a/service/src/bridge/workspace-instance.test.ts b/service/src/bridge/workspace-instance.test.ts new file mode 100644 index 00000000..44e93300 --- /dev/null +++ b/service/src/bridge/workspace-instance.test.ts @@ -0,0 +1,40 @@ +import { describe, expect, it } from 'bun:test'; +import { principalWorkspaceInstanceId } from './workspace-instance'; + +describe('principalWorkspaceInstanceId', () => { + it('keeps principal components distinct even when identifiers contain delimiters', () => { + const instanceId = 'a'.repeat(64); + expect(principalWorkspaceInstanceId({ instanceId, tenantId: 'tenant\0user', principalId: 'a' })) + .not.toBe(principalWorkspaceInstanceId({ instanceId, tenantId: 'tenant', principalId: 'user\0a' })); + }); + it('is stable only within the same authenticated principal', () => { + const instanceId = 'a'.repeat(64); + const first = principalWorkspaceInstanceId({ + instanceId, + tenantId: 'tenant', + principalId: 'user-a', + }); + expect(first).toMatch(/^[a-f0-9]{64}$/); + expect( + principalWorkspaceInstanceId({ + instanceId, + tenantId: 'tenant', + principalId: 'user-a', + }) + ).toBe(first); + expect( + principalWorkspaceInstanceId({ + instanceId, + tenantId: 'tenant', + principalId: 'user-b', + }) + ).not.toBe(first); + expect( + principalWorkspaceInstanceId({ + instanceId, + tenantId: 'other', + principalId: 'user-a', + }) + ).not.toBe(first); + }); +}); diff --git a/service/src/bridge/workspace-instance.ts b/service/src/bridge/workspace-instance.ts new file mode 100644 index 00000000..4c49afbb --- /dev/null +++ b/service/src/bridge/workspace-instance.ts @@ -0,0 +1,17 @@ +import { createHash } from 'node:crypto'; + +/** Bind a caller-selected conversation identity to the authenticated principal. */ +export function principalWorkspaceInstanceId(args: { + instanceId: string; + tenantId: string; + principalId: string; +}): string { + return createHash('sha256') + .update(JSON.stringify([ + 'codeapi-workspace-instance-v1', + args.tenantId, + args.principalId, + args.instanceId, + ])) + .digest('hex'); +} diff --git a/service/src/service/programmatic-router.ts b/service/src/service/programmatic-router.ts index 1063fbfe..a646b9ad 100644 --- a/service/src/service/programmatic-router.ts +++ b/service/src/service/programmatic-router.ts @@ -44,6 +44,7 @@ import { SessionKeyResolutionError, } from '../session-key'; import { getCredentialId, getPrincipalOrReject } from '../auth/principal'; +import { principalWorkspaceInstanceId } from '../bridge/workspace-instance'; import { getExecutionIdentity } from '../execution-identity'; import { PROGRAMMATIC_RUNTIME_SESSION_EXEMPTION } from '../runtime-session/job-policy'; import { @@ -81,6 +82,7 @@ import { authorizeRequestedFiles, } from './file-authorization'; import { + bindReplayWorkspaceInstance, buildReplayExecutionState, resolveReplayStateSandboxBackend, } from './programmatic-state'; @@ -335,7 +337,7 @@ function buildReplayPayload( state: ExecutionState, history: Record, ): t.PayloadBody { - return createProgrammaticPayload({ + const payload = createProgrammaticPayload({ req, session_id: state.session_id, execution_id: state.execution_id, @@ -347,6 +349,7 @@ function buildReplayPayload( filesOverride: state.files, language: state.language ?? 'python', }); + return bindReplayWorkspaceInstance(payload, state); } async function runReplayIteration( @@ -501,10 +504,17 @@ async function handleReplayInitial( userId: string; bridgeWorkerId?: string; workspaceId?: string; + workspaceInstanceId?: string; }, cancellation: ReplayRequestCancellation, ): Promise { - const { apiKeyId, userId, bridgeWorkerId, workspaceId } = params; + const { + apiKeyId, + userId, + bridgeWorkerId, + workspaceId, + workspaceInstanceId, + } = params; const { code, tools, user_id, files } = req.body as t.ProgrammaticRequestBody; let timeout: number; @@ -660,6 +670,7 @@ async function handleReplayInitial( language, bridgeWorkerId, workspaceId, + workspaceInstanceId, executionProfile: env.EXECUTION_PROFILE, executionProfileSource: env.EXECUTION_PROFILE_SOURCE, sandboxBackend: resolveReplayStateSandboxBackend({ @@ -1279,6 +1290,7 @@ router.post( const requestedLanguage: unknown = rawBody.language ?? rawBody.lang; let bridgeWorkerId: string | undefined; let workspaceId: string | undefined; + let workspaceInstanceId: string | undefined; if (continuation_token == null || continuation_token === '') { try { const bridgeSelection = resolveBridgeWorkerSelection({ @@ -1312,6 +1324,23 @@ router.post( } workspaceId = requestedWorkspaceId; } + const requestedWorkspaceInstanceId = rawBody.workspace_instance_id; + if (requestedWorkspaceInstanceId !== undefined) { + if ( + workspaceId == null || + typeof requestedWorkspaceInstanceId !== 'string' || + !/^[a-f0-9]{64}$/.test(requestedWorkspaceInstanceId) + ) { + return res.status(400).json({ + error: 'Invalid code workspace instance ID', + }); + } + workspaceInstanceId = principalWorkspaceInstanceId({ + instanceId: requestedWorkspaceInstanceId, + tenantId: principal.tenantId, + principalId: principal.userId, + }); + } } catch (error) { if (error instanceof BridgeWorkerSelectionError) { return res @@ -1428,6 +1457,7 @@ router.post( userId, bridgeWorkerId, workspaceId, + workspaceInstanceId, }, cancellation); } if (workspaceId != null) { diff --git a/service/src/service/programmatic-state.test.ts b/service/src/service/programmatic-state.test.ts index 81405021..b0bff413 100644 --- a/service/src/service/programmatic-state.test.ts +++ b/service/src/service/programmatic-state.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from 'bun:test'; import type { CodeApiAuthContext, RequestFile } from '../types'; import type { LCTool } from '../preamble'; import { + bindReplayWorkspaceInstance, buildReplayExecutionState, resolveReplayStateSandboxBackend, } from './programmatic-state'; @@ -84,6 +85,7 @@ describe('buildReplayExecutionState', () => { authContext, bridgeWorkerId: 'code-user_123', workspaceId: 'project-a', + workspaceInstanceId: 'a'.repeat(64), sandboxBackend: 'remote-bridge', executionProfile: 'stateful', executionProfileSource: 'explicit', @@ -104,6 +106,7 @@ describe('buildReplayExecutionState', () => { apiKeyId: 'key_legacy', bridgeWorkerId: 'code-user_123', workspaceId: 'project-a', + workspaceInstanceId: 'a'.repeat(64), sandboxBackend: 'remote-bridge', executionProfile: 'stateful', executionProfileSource: 'explicit', @@ -119,6 +122,19 @@ describe('buildReplayExecutionState', () => { }); }); + test('binds a selected conversation checkout into every replay payload', () => { + const payload = { language: 'bash', version: '5.2', files: [] }; + expect( + bindReplayWorkspaceInstance(payload, { + workspaceInstanceId: 'b'.repeat(64), + }), + ).toEqual({ + ...payload, + workspace_instance_id: 'b'.repeat(64), + }); + expect(bindReplayWorkspaceInstance(payload, {})).toBe(payload); + }); + test('falls back to JWT identity only when no managed auth context exists', () => { const state = build({ authContext: undefined, userId: 'user_api_key' }); diff --git a/service/src/service/programmatic-state.ts b/service/src/service/programmatic-state.ts index e6bda59a..38c72486 100644 --- a/service/src/service/programmatic-state.ts +++ b/service/src/service/programmatic-state.ts @@ -39,6 +39,7 @@ export interface BuildReplayExecutionStateParams { language: 'python' | 'bash'; bridgeWorkerId?: string; workspaceId?: string; + workspaceInstanceId?: string; sandboxBackend?: SandboxBackendName; executionProfile: ExecutionProfile; executionProfileSource: ExecutionProfileSource; @@ -68,6 +69,7 @@ export function buildReplayExecutionState( apiKeyId: params.apiKeyId, bridgeWorkerId: params.bridgeWorkerId, workspaceId: params.workspaceId, + workspaceInstanceId: params.workspaceInstanceId, sandboxBackend: params.sandboxBackend, executionProfile: params.executionProfile, executionProfileSource: params.executionProfileSource, @@ -83,3 +85,13 @@ export function buildReplayExecutionState( language: params.language, }; } + +/** Bind the authenticated conversation checkout to every replay iteration. */ +export function bindReplayWorkspaceInstance( + payload: t.PayloadBody, + state: Pick, +): t.PayloadBody { + return state.workspaceInstanceId == null + ? payload + : { ...payload, workspace_instance_id: state.workspaceInstanceId }; +} diff --git a/service/src/service/replay-state.ts b/service/src/service/replay-state.ts index 3b65cedf..fd4e90e3 100644 --- a/service/src/service/replay-state.ts +++ b/service/src/service/replay-state.ts @@ -118,6 +118,8 @@ export interface ExecutionState { bridgeWorkerId?: string; /** Selected workspace retained and bound across every replay iteration. */ workspaceId?: string; + /** Selected conversation checkout retained across every replay iteration. */ + workspaceInstanceId?: string; /** Original queue/backend target retained across replay continuations. */ sandboxBackend?: SandboxBackendName; /** Original producer profile retained so continuations use the same queue. */ diff --git a/service/src/types/service.ts b/service/src/types/service.ts index 0404ad16..91c3ed89 100644 --- a/service/src/types/service.ts +++ b/service/src/types/service.ts @@ -204,6 +204,8 @@ export type PayloadFileRef = { export interface PayloadBody { language: string; version: string; + /** Opaque conversation checkout selected and authenticated by the API. */ + workspace_instance_id?: string; /** Stable identity shared by all replay iterations of one execution. */ execution_id?: string; replay_tool_count?: number; @@ -393,6 +395,8 @@ export interface ProgrammaticRequestBody { * legacy `/exec` sandbox body), so the router accepts either key and * normalizes to `language`. If both are present, `language` wins. */ lang?: 'python' | 'bash'; + /** Opaque conversation checkout binding for a selected native workspace. */ + workspace_instance_id?: string; } export interface ProgrammaticToolCall { diff --git a/service/src/workspace-tools/router.test.ts b/service/src/workspace-tools/router.test.ts index 04738b8a..a52b3f0b 100644 --- a/service/src/workspace-tools/router.test.ts +++ b/service/src/workspace-tools/router.test.ts @@ -13,6 +13,7 @@ import { executionProfileMiddleware } from '../middleware/execution-profile'; import { hostedAppPreviewGateway } from '../hosted-app/preview-gateway'; import { applyPrincipal } from '../auth/principal'; import { BridgeStoreError } from '../bridge/store'; +import { principalWorkspaceInstanceId } from '../bridge/workspace-instance'; import { bridgeStoreStatus, createWorkspaceToolsRouter } from './router'; import type { WorkspaceToolRequest } from '../../../packages/code/src/protocol'; @@ -36,6 +37,36 @@ test('maps invalid worker results to an upstream failure', () => { expect(bridgeStoreStatus(new BridgeStoreError('WORKER_QUEUE_FULL', 'queue full'))).toBe(429); }); +test('binds instance admission to the authenticated tenant and user while preserving legacy requests', async () => { + const app = express(); + app.use(json()); + app.use((req, _res, next) => { + applyPrincipal(req, { userId: 'user-1', tenantId: 'tenant-1', principalSource: 'librechat_jwt', codeWorkerId: 'user-worker' }); + next(); + }); + const dispatched: WorkspaceToolRequest[] = []; + app.use(createWorkspaceToolsRouter({ + backend: 'remote-bridge', configuredWorkerId: 'user-worker', dynamicWorkers: false, + store: { async dispatchWorkspaceTool(args) { + dispatched.push(args.request); + return { protocolVersion: 1, generation: 1, leaseToken: 'lease', incarnationId: 'incarnation', status: 'rejected', error: 'fixture' }; + } }, + })); + server = createServer(app); + await new Promise(resolve => server!.listen(0, '127.0.0.1', resolve)); + const address = server.address(); + if (address == null || typeof address === 'string') throw new Error('Missing listener'); + for (const workspaceInstanceId of ['a'.repeat(64), undefined]) { + const response = await fetch(`http://127.0.0.1:${address.port}/workspace-tools/execute`, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ protocolVersion: 1, operation: 'read_file', workspaceId: 'primary', workspaceInstanceId, path: 'README.md' }), + }); + await response.json(); + } + expect(dispatched[0]?.workspaceInstanceId).toBe(principalWorkspaceInstanceId({ instanceId: 'a'.repeat(64), tenantId: 'tenant-1', principalId: 'user-1' })); + expect(dispatched[1]?.workspaceInstanceId).toBeUndefined(); +}); + test.each<[WorkspaceToolRequest, number, number?]>([ [{ protocolVersion: 1, operation: 'read_file', workspaceId: 'primary', path: 'README.md' }, 30_000, undefined], [{ protocolVersion: 1, operation: 'execute_command', workspaceId: 'primary', command: 'echo ready' }, 35_000, undefined], diff --git a/service/src/workspace-tools/router.ts b/service/src/workspace-tools/router.ts index eb89370e..3e1f9fee 100644 --- a/service/src/workspace-tools/router.ts +++ b/service/src/workspace-tools/router.ts @@ -18,6 +18,7 @@ import { BridgeWorkerSelectionError, resolveBridgeWorkerSelection, } from '../bridge/selection'; +import { principalWorkspaceInstanceId } from '../bridge/workspace-instance'; interface WorkspaceToolsRouterOptions { store: Pick; @@ -82,12 +83,22 @@ export function createWorkspaceToolsRouter(options: WorkspaceToolsRouterOptions) return; } outcome.operation = req.body.operation; - const request: WorkspaceToolRequest = req.body.operation === 'execute_command' - ? { ...req.body, timeoutMs: Math.min( - req.body.timeoutMs ?? BRIDGE_WORKSPACE_COMMAND_DEFAULT_TIMEOUT_MS, + const principalRequest: WorkspaceToolRequest = req.body.workspaceInstanceId == null + ? req.body + : { + ...req.body, + workspaceInstanceId: principalWorkspaceInstanceId({ + instanceId: req.body.workspaceInstanceId, + tenantId: principal.tenantId, + principalId: principal.userId, + }), + }; + const request: WorkspaceToolRequest = principalRequest.operation === 'execute_command' + ? { ...principalRequest, timeoutMs: Math.min( + principalRequest.timeoutMs ?? BRIDGE_WORKSPACE_COMMAND_DEFAULT_TIMEOUT_MS, options.timeoutMs ?? Number.MAX_SAFE_INTEGER, ) } - : req.body; + : principalRequest; const executionBudgetMs = request.operation === 'execute_command' ? request.timeoutMs! + 5_000 : Math.min(options.timeoutMs ?? 30_000, 30_000);