diff --git a/src/gateway/worker-environments/node-workspace-transfer-snapshot.ts b/src/gateway/worker-environments/node-workspace-transfer-snapshot.ts index d0c247c305d2..dd4d4792989b 100644 --- a/src/gateway/worker-environments/node-workspace-transfer-snapshot.ts +++ b/src/gateway/worker-environments/node-workspace-transfer-snapshot.ts @@ -8,10 +8,10 @@ import { } from "./workspace-manifest.js"; import { readActualWorkspaceManifest } from "./workspace-reconcile.js"; import { - createGitTransferList, + createWorkspaceGitTransferList, readWorkspaceTransferPaths, - runLocalCommandToFile, -} from "./workspace-sync-local.js"; + runWorkspaceInventoryCommandToFile, +} from "./workspace-sync-inventory.js"; const TRANSFER_TIMEOUT_MS = 10 * 60_000; @@ -59,7 +59,7 @@ export async function prepareNodeWorkspaceTransferSnapshot(params: { if (!/^[a-f0-9]{40}(?:[a-f0-9]{24})?$/u.test(baseCommit)) { throw new Error("Worker workspace Git base is not a commit id"); } - const transferList = await createGitTransferList({ + const transferList = await createWorkspaceGitTransferList({ gitRoot: root, temporaryDirectory: path.join(params.temporaryRoot, "inventory"), signal: params.signal ?? AbortSignal.timeout(TRANSFER_TIMEOUT_MS), @@ -81,7 +81,7 @@ export async function prepareNodeWorkspaceTransferSnapshot(params: { const signal = params.signal ?? AbortSignal.timeout(TRANSFER_TIMEOUT_MS); const objectListPath = path.join(params.temporaryRoot, "base-objects"); packPath = path.join(params.temporaryRoot, "base.pack"); - await runLocalCommandToFile({ + await runWorkspaceInventoryCommandToFile({ argv: [ "git", "-C", @@ -96,7 +96,7 @@ export async function prepareNodeWorkspaceTransferSnapshot(params: { timeoutMs: TRANSFER_TIMEOUT_MS, }); await fsp.appendFile(objectListPath, `${baseCommit}\n`); - await runLocalCommandToFile({ + await runWorkspaceInventoryCommandToFile({ argv: ["git", "-C", root, "pack-objects", "--stdout"], inputPath: objectListPath, outputPath: packPath, diff --git a/src/gateway/worker-environments/workspace-sync-local.test.ts b/src/gateway/worker-environments/workspace-sync-inventory.test.ts similarity index 96% rename from src/gateway/worker-environments/workspace-sync-local.test.ts rename to src/gateway/worker-environments/workspace-sync-inventory.test.ts index 41e21dae848d..0955d1070dc6 100644 --- a/src/gateway/worker-environments/workspace-sync-local.test.ts +++ b/src/gateway/worker-environments/workspace-sync-inventory.test.ts @@ -9,10 +9,10 @@ import { MAX_WORKSPACE_INVENTORY_ENTRIES, } from "./workspace-inventory-limits.js"; import { - createGitTransferList, + createWorkspaceGitTransferList, filterExistingGitTransferList, - runLocalCommandToFile, -} from "./workspace-sync-local.js"; + runWorkspaceInventoryCommandToFile, +} from "./workspace-sync-inventory.js"; import { preflightWorkerWorkspace } from "./workspace-sync-preflight.js"; const tempDirs = useAutoCleanupTempDirTracker(afterEach); @@ -99,14 +99,14 @@ async function waitForFile(filePath: string): Promise { throw new Error(`Timed out waiting for ${filePath}`); } -describe("runLocalCommandToFile", () => { +describe("runWorkspaceInventoryCommandToFile", () => { it("fully persists bounded stdout after a positive short write", async () => { const root = tempDirs.make("openclaw-workspace-command-short-write-"); const outputPath = path.join(root, "output"); const expected = Buffer.from("bounded workspace inventory output\n"); const shortWriteObserved = injectPositiveShortWrite(outputPath); - await runLocalCommandToFile({ + await runWorkspaceInventoryCommandToFile({ argv: [process.execPath, "-e", "process.stdout.write(process.argv[1])", expected.toString()], outputPath, signal: new AbortController().signal, @@ -130,7 +130,7 @@ describe("runLocalCommandToFile", () => { await git(root, "init", "--quiet"); const shortWriteObserved = injectPositiveShortWrite(outputPath); - await createGitTransferList({ + await createWorkspaceGitTransferList({ gitRoot: root, temporaryDirectory, signal: new AbortController().signal, @@ -168,7 +168,7 @@ describe("runLocalCommandToFile", () => { const outputPath = path.join(root, "output"); const readyPath = path.join(root, "ready"); const controller = new AbortController(); - const operation = runLocalCommandToFile({ + const operation = runWorkspaceInventoryCommandToFile({ argv: [ process.execPath, "-e", @@ -203,7 +203,7 @@ describe("runLocalCommandToFile", () => { const outputPath = path.join(root, "pack"); await expect( - runLocalCommandToFile({ + runWorkspaceInventoryCommandToFile({ argv: [process.execPath, "-e", 'process.stdout.write("x".repeat(1024))'], outputPath, signal: new AbortController().signal, @@ -238,14 +238,14 @@ describe("runLocalCommandToFile", () => { await fs.writeFile(path.join(root, file), file); }), ); - await runLocalCommandToFile({ + await runWorkspaceInventoryCommandToFile({ argv: ["git", "-C", root, "init", "--quiet"], outputPath: initOutputPath, signal: new AbortController().signal, timeoutMs: 10_000, }); - const outputPath = await createGitTransferList({ + const outputPath = await createWorkspaceGitTransferList({ gitRoot: root, temporaryDirectory, signal: new AbortController().signal, @@ -292,7 +292,7 @@ process.stdout.write("eligible.txt\\0".repeat(count)); vi.stubEnv("PATH", `${bin}${path.delimiter}${process.env.PATH ?? ""}`); await writeMockGit(MAX_WORKSPACE_INVENTORY_ENTRIES + 1); - const acceptedPath = await createGitTransferList({ + const acceptedPath = await createWorkspaceGitTransferList({ gitRoot: root, temporaryDirectory: firstTransfer, signal: new AbortController().signal, @@ -304,7 +304,7 @@ process.stdout.write("eligible.txt\\0".repeat(count)); await writeMockGit(MAX_WORKSPACE_GIT_CANDIDATES + 1); await expect( - createGitTransferList({ + createWorkspaceGitTransferList({ gitRoot: root, temporaryDirectory: secondTransfer, signal: new AbortController().signal, @@ -362,7 +362,7 @@ describe("preflightWorkerWorkspace", () => { await fs.writeFile(path.join(root, "nested", "private.txt"), "nested\n"); await preflightWorkerWorkspace({ localPath: root, timeoutMs: 10_000 }); - const transferPath = await createGitTransferList({ + const transferPath = await createWorkspaceGitTransferList({ gitRoot: root, temporaryDirectory: transferDirectory, signal: new AbortController().signal, diff --git a/src/gateway/worker-environments/workspace-sync-inventory.ts b/src/gateway/worker-environments/workspace-sync-inventory.ts index 302060348fb9..e75c7bfe2242 100644 --- a/src/gateway/worker-environments/workspace-sync-inventory.ts +++ b/src/gateway/worker-environments/workspace-sync-inventory.ts @@ -181,6 +181,38 @@ async function* readBoundedGitPathCandidates(filePath: string): AsyncGenerator> { + const paths = new Set(); + for await (const entry of readBoundedGitPathCandidates(filePath)) { + paths.add(entry); + } + return paths; +} + +export async function filterExistingGitTransferList(params: { + gitRoot: string; + preparedListPath: string; + outputPath: string; +}): Promise { + const output = await fs.open(params.outputPath, "wx", 0o600); + try { + for await (const file of readBoundedGitPathCandidates(params.preparedListPath)) { + const stats = await fs.lstat(path.join(params.gitRoot, file)).catch((error: unknown) => { + if (hasNodeErrorCode(error, "ENOENT")) { + return undefined; + } + throw error; + }); + if (stats?.isFile() || stats?.isSymbolicLink()) { + await output.writeFile(`${file}\0`); + } + } + } finally { + await output.close(); + } + return params.outputPath; +} + export async function runWorkspaceInventoryCommandToFile(params: { argv: string[]; inputPath?: string; diff --git a/src/gateway/worker-environments/workspace-sync-local.ts b/src/gateway/worker-environments/workspace-sync-local.ts deleted file mode 100644 index 28cb483a636f..000000000000 --- a/src/gateway/worker-environments/workspace-sync-local.ts +++ /dev/null @@ -1,94 +0,0 @@ -import { createReadStream } from "node:fs"; -import fs from "node:fs/promises"; -import path from "node:path"; -import { hasNodeErrorCode } from "../../infra/path-guards.js"; -import { - createWorkspaceGitTransferList, - runWorkspaceInventoryCommandToFile, -} from "./workspace-sync-inventory.js"; - -function validateGitRelativePath(file: string): string { - if ( - !file || - path.posix.isAbsolute(file) || - path.posix.normalize(file) !== file || - file === ".." || - file.startsWith("../") - ) { - throw new Error("Worker workspace git file list contains an unsafe path"); - } - return file; -} - -async function* readNulFile(filePath: string): AsyncGenerator { - let pending = Buffer.alloc(0); - for await (const value of createReadStream(filePath)) { - const chunk = Buffer.isBuffer(value) ? value : Buffer.from(value); - const buffer = pending.length === 0 ? chunk : Buffer.concat([pending, chunk]); - let offset = 0; - for (;;) { - const separator = buffer.indexOf(0, offset); - if (separator < 0) { - break; - } - yield validateGitRelativePath(buffer.subarray(offset, separator).toString("utf8")); - offset = separator + 1; - } - pending = Buffer.from(buffer.subarray(offset)); - } - if (pending.length > 0) { - throw new Error("Worker workspace git file list is not NUL terminated"); - } -} - -export async function readWorkspaceTransferPaths(filePath: string): Promise> { - const paths = new Set(); - for await (const entry of readNulFile(filePath)) { - paths.add(entry); - } - return paths; -} - -export async function runLocalCommandToFile(params: { - argv: string[]; - inputPath?: string; - outputPath: string; - signal: AbortSignal; - timeoutMs: number; - maxOutputBytes?: number; -}): Promise { - await runWorkspaceInventoryCommandToFile(params); -} - -export async function createGitTransferList(params: { - gitRoot: string; - temporaryDirectory: string; - signal: AbortSignal; - timeoutMs: number; -}): Promise { - return await createWorkspaceGitTransferList(params); -} - -export async function filterExistingGitTransferList(params: { - gitRoot: string; - preparedListPath: string; - outputPath: string; -}): Promise { - const output = await fs.open(params.outputPath, "wx", 0o600); - try { - for await (const file of readNulFile(params.preparedListPath)) { - const stats = await fs.lstat(path.join(params.gitRoot, file)).catch((error: unknown) => { - if (hasNodeErrorCode(error, "ENOENT")) { - return undefined; - } - throw error; - }); - if (stats?.isFile() || stats?.isSymbolicLink()) { - await output.writeFile(`${file}\0`); - } - } - } finally { - await output.close(); - } - return params.outputPath; -} diff --git a/src/gateway/worker-environments/workspace-sync.ts b/src/gateway/worker-environments/workspace-sync.ts index 5d3564025666..bc5550ed8894 100644 --- a/src/gateway/worker-environments/workspace-sync.ts +++ b/src/gateway/worker-environments/workspace-sync.ts @@ -63,10 +63,10 @@ import { type WorkerWorkspaceActionsOptions, } from "./workspace-sync-helpers.js"; import { - createGitTransferList, + createWorkspaceGitTransferList, filterExistingGitTransferList, - runLocalCommandToFile, -} from "./workspace-sync-local.js"; + runWorkspaceInventoryCommandToFile, +} from "./workspace-sync-inventory.js"; import { REMOTE_GIT_WORKSPACE_RETRY_RESET_JS, REMOTE_GIT_WORKSPACE_SETUP_SCRIPT, @@ -253,7 +253,7 @@ export function createWorkerWorkspaceActions( throw new Error("Worker workspace git base is not a commit id"); } - gitTransferListPath = await createGitTransferList({ + gitTransferListPath = await createWorkspaceGitTransferList({ gitRoot, temporaryDirectory: path.join(temporaryDirectory, "transfer"), signal: options.ownerSignal, @@ -262,7 +262,7 @@ export function createWorkerWorkspaceActions( const objectListPath = path.join(temporaryDirectory, "base-objects"); const packPath = path.join(temporaryDirectory, "base.pack"); - await runLocalCommandToFile({ + await runWorkspaceInventoryCommandToFile({ argv: [ "git", "-C", @@ -277,7 +277,7 @@ export function createWorkerWorkspaceActions( timeoutMs: WORKSPACE_TIMEOUT_MS, }); await fs.appendFile(objectListPath, `${baseCommit}\n`); - await runLocalCommandToFile({ + await runWorkspaceInventoryCommandToFile({ argv: ["git", "-C", gitRoot, "pack-objects", "--stdout"], inputPath: objectListPath, outputPath: packPath,