From b61bc5cd8da6f4ebd49f240069a703bc1d13142a Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 9 Aug 2026 03:51:24 -0700 Subject: [PATCH] fix: sync Cloud Worker workspaces from macOS gateways (#121027) * fix(cloud-workers): support Apple openrsync receivers * build: register worker receiver entry * fix(cloud-workers): narrow receiver invocation types * test(cloud-workers): resolve real rsync before wrapping --- config/knip.config.ts | 2 + package.json | 2 +- scripts/check-changed.mjs | 2 +- scripts/ci-changed-scope.mjs | 3 + ...ver.sessions.worker-original-order.test.ts | 12 +- .../worker-environments/service.test.ts | 5 +- src/gateway/worker-environments/service.ts | 4 +- .../tunnel.test-support.ts | 95 ++++++++++- .../worker-environments/tunnel.test.ts | 5 + src/gateway/worker-environments/tunnel.ts | 4 + .../workspace-accepted-sync.test.ts | 47 +++--- .../workspace-accepted-sync.ts | 5 +- .../workspace-mutation-remote-script.test.ts | 37 +++-- .../workspace-rsync-path.test.ts | 156 ++++++++++++++++++ .../workspace-sync-helpers.ts | 59 ++++--- .../workspace-sync-scripts.ts | 10 +- .../workspace-sync-tunnel.test.ts | 97 ++++++----- .../workspace-sync.test.ts | 1 + .../worker-environments/workspace-sync.ts | 37 +++-- ...ci-changed-scope.contract-fixtures.test.ts | 15 ++ src/worker/workspace-rsync-receiver.ts | 58 +++++++ test/scripts/changed-lanes.test.ts | 25 +++ test/scripts/package-mac-app.test.ts | 1 + tsdown.config.ts | 1 + 24 files changed, 530 insertions(+), 153 deletions(-) create mode 100644 src/gateway/worker-environments/workspace-rsync-path.test.ts create mode 100644 src/worker/workspace-rsync-receiver.ts diff --git a/config/knip.config.ts b/config/knip.config.ts index 2705bdee70cf..c687a8e684fb 100644 --- a/config/knip.config.ts +++ b/config/knip.config.ts @@ -108,6 +108,8 @@ const rootEntries = [ "src/entry.ts!", // Built as the official image's Docker HEALTHCHECK entrypoint. "src/docker-healthcheck.ts!", + // Uploaded in the worker bundle and launched by rsync; no static host import exists. + "src/worker/workspace-rsync-receiver.ts!", // Shipped compatibility facade for statusCommand and getStatusSummary. "src/commands/status.ts!", "src/cli/daemon-cli.ts!", diff --git a/package.json b/package.json index 04a8afae4661..f0eb9b4ca39b 100644 --- a/package.json +++ b/package.json @@ -1906,7 +1906,7 @@ "test:live:media:music": "node --import tsx test/e2e/qa-lab/media/hosted-media-provider-live.ts music", "test:live:media:video": "node --import tsx test/e2e/qa-lab/media/hosted-media-provider-live.ts video", "test:live:models-profiles": "node scripts/test-live.mjs -- src/agents/models.profiles.live.test.ts", - "test:macos:ci": "node scripts/test-projects.mjs src/daemon/launchd.test.ts src/daemon/runtime-paths.test.ts src/daemon/runtime-binary.test.ts src/infra/brew.test.ts src/infra/stable-node-path.test.ts test/scripts/vitest-process-group.test.ts test/scripts/package-mac-app.test.ts test/scripts/package-mac-dist.test.ts test/scripts/create-dmg.test.ts test/scripts/codesign-mac-app.test.ts test/scripts/notarize-mac-artifact.test.ts", + "test:macos:ci": "node scripts/test-projects.mjs src/daemon/launchd.test.ts src/daemon/runtime-paths.test.ts src/daemon/runtime-binary.test.ts src/gateway/worker-environments/workspace-rsync-path.test.ts src/infra/brew.test.ts src/infra/stable-node-path.test.ts test/scripts/vitest-process-group.test.ts test/scripts/package-mac-app.test.ts test/scripts/package-mac-dist.test.ts test/scripts/create-dmg.test.ts test/scripts/codesign-mac-app.test.ts test/scripts/notarize-mac-artifact.test.ts", "test:max": "node scripts/test-projects-max.mjs", "test:parallels:linux": "bash scripts/e2e/parallels-linux-smoke.sh", "test:parallels:macos": "bash scripts/e2e/parallels-macos-smoke.sh", diff --git a/scripts/check-changed.mjs b/scripts/check-changed.mjs index 2a433423572d..5db0bacfe9c7 100644 --- a/scripts/check-changed.mjs +++ b/scripts/check-changed.mjs @@ -79,7 +79,7 @@ const ANDROID_VERSION_SYNC_PATHS = new Set([ "apps/android/version.json", ]); const MACOS_APP_CI_PATH_RE = - /^(?:apps\/(?:macos|macos-mlx-tts|shared|swabble)\/|Swabble\/|scripts\/(?:codesign-mac-app|create-dmg|notarize-mac-artifact|package-mac-app|package-mac-dist)\.sh$|scripts\/lib\/(?:plistbuddy|swift-toolchain)\.sh$|test\/scripts\/(?:codesign-mac-app|create-dmg|notarize-mac-artifact|package-mac-app|package-mac-dist)\.test\.ts$)/u; + /^(?:apps\/(?:macos|macos-mlx-tts|shared|swabble)\/|Swabble\/|src\/(?:worker\/workspace-rsync-receiver\.ts|gateway\/worker-environments\/workspace-(?:accepted-(?:remote-script|sync)|mutation-remote-script|rsync-path\.test|sync(?:-helpers)?)\.ts)$|scripts\/(?:codesign-mac-app|create-dmg|notarize-mac-artifact|package-mac-app|package-mac-dist)\.sh$|scripts\/lib\/(?:plistbuddy|swift-toolchain)\.sh$|test\/scripts\/(?:codesign-mac-app|create-dmg|notarize-mac-artifact|package-mac-app|package-mac-dist)\.test\.ts$)/u; let corepackPnpmShimDir; let corepackPnpmShimCleanupRegistered = false; let cachedGeneratedExtensionAssetPaths; diff --git a/scripts/ci-changed-scope.mjs b/scripts/ci-changed-scope.mjs index 40ecfb97da9b..bbef1b9b7cee 100644 --- a/scripts/ci-changed-scope.mjs +++ b/scripts/ci-changed-scope.mjs @@ -45,6 +45,8 @@ const MACOS_NATIVE_RE = /^(apps\/macos\/|apps\/macos-mlx-tts\/|apps\/ios\/|apps\/shared\/|apps\/swabble\/|Swabble\/)/; const MACOS_SCRIPT_SCOPE_RE = /^(?:scripts\/(?:check-swift-tools|codesign-mac-app|create-dmg|format-swift|install-swift-tools|install-xcodegen|lint-swift|notarize-mac-artifact|package-mac-app|package-mac-dist)\.sh|scripts\/lib\/(?:plistbuddy|swift-toolchain)\.sh|test\/scripts\/(?:codesign-mac-app|create-dmg|notarize-mac-artifact|package-mac-app|package-mac-dist)\.test\.ts)$/; +const WORKSPACE_RSYNC_RECEIVER_SCOPE_RE = + /^src\/(?:worker\/workspace-rsync-receiver\.ts|gateway\/worker-environments\/workspace-(?:accepted-(?:remote-script|sync)|mutation-remote-script|rsync-path\.test|sync(?:-helpers)?)\.ts)$/; const IOS_BUILD_RE = /^(apps\/ios\/|apps\/shared\/|apps\/swabble\/|Swabble\/|scripts\/(?:check-swift-tools|format-swift|install-swift-tools|install-xcodegen|lint-swift)\.sh$|scripts\/(?:ios-(?:configure-signing|team-id|write-version-xcconfig)\.sh|ios-write-swift-filelist\.mjs|ios-version\.ts)$|scripts\/lib\/(?:ios-version\.ts|release-version\.mjs|version-script-args\.ts)$)/; const ANDROID_NATIVE_RE = /^(apps\/android\/|apps\/shared\/)/; @@ -146,6 +148,7 @@ export function detectChangedScope(changedPaths) { !NATIVE_PROTOCOL_GEN_RE.test(path) && (MACOS_NATIVE_RE.test(path) || MACOS_SCRIPT_SCOPE_RE.test(path) || + WORKSPACE_RSYNC_RECEIVER_SCOPE_RE.test(path) || APPLE_SHARED_CONTRACT_FIXTURE_RE.test(path) || isAppleSwiftConfig) ) { diff --git a/src/gateway/server.sessions.worker-original-order.test.ts b/src/gateway/server.sessions.worker-original-order.test.ts index bb23212f2370..1d58ca088fe6 100644 --- a/src/gateway/server.sessions.worker-original-order.test.ts +++ b/src/gateway/server.sessions.worker-original-order.test.ts @@ -29,6 +29,7 @@ import { import { createWorkerEnvironmentStore } from "./worker-environments/store.js"; import type { WorkerSshProcess, WorkerSshRunner } from "./worker-environments/tunnel-ssh-runner.js"; import { createWorkerTunnelManager } from "./worker-environments/tunnel.js"; +import { prepareLocalWorkspaceRsyncBoundary } from "./worker-environments/tunnel.test-support.js"; import { rsyncArgvPort, sshArgvPort } from "./worker-environments/worker-ssh-argv.test-support.js"; import { createWorkerWorkspaceOperationCoordinator } from "./worker-environments/workspace-operation-coordinator.js"; @@ -189,15 +190,18 @@ class OriginalOrderSshRunner implements WorkerSshRunner { } if (argv[0] === "rsync") { this.events.push(`workspace:transfer:${port}`); + if (argv.some((arg) => arg.startsWith("--rsync-path="))) { + const boundary = await prepareLocalWorkspaceRsyncBoundary(this.remoteHome, argv); + return await runCommandWithTimeout(boundary.argv, { + ...options, + baseEnv: { ...options.baseEnv, HOME: this.remoteHome }, + }); + } const localArgv = [...argv]; const remoteShellIndex = localArgv.indexOf("-e"); if (remoteShellIndex >= 0) { localArgv.splice(remoteShellIndex, 2); } - const remoteReceiverIndex = localArgv.findIndex((arg) => arg.startsWith("--rsync-path=")); - if (remoteReceiverIndex >= 0) { - localArgv.splice(remoteReceiverIndex, 1); - } for (let index = 1; index < localArgv.length; index += 1) { const candidate = localArgv[index]; const separator = candidate?.indexOf(":") ?? -1; diff --git a/src/gateway/worker-environments/service.test.ts b/src/gateway/worker-environments/service.test.ts index 0edb753ded20..9f698dafd63e 100644 --- a/src/gateway/worker-environments/service.test.ts +++ b/src/gateway/worker-environments/service.test.ts @@ -2195,7 +2195,10 @@ describe("worker environment service", () => { ownerEpoch: 1, }); expect(tunnelManager.start).toHaveBeenCalledWith( - expect.objectContaining({ gateway: { host: "127.0.0.1", port: 18_789 } }), + expect.objectContaining({ + bundleHash: BUNDLE_HASH, + gateway: { host: "127.0.0.1", port: 18_789 }, + }), ); expect(workerService.get("worker-tunnel")).toMatchObject({ tunnelStatus: "connected" }); diff --git a/src/gateway/worker-environments/service.ts b/src/gateway/worker-environments/service.ts index 34b798462f6a..58ae9b2d21c9 100644 --- a/src/gateway/worker-environments/service.ts +++ b/src/gateway/worker-environments/service.ts @@ -1082,7 +1082,8 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService !inState(record, "ready", "idle", "attached") || record.destroyRequestedAtMs !== null || !record.leaseId || - !record.sshEndpoint + !record.sshEndpoint || + !record.bootstrapReceipt ) { throw serviceError("invalid_state", `Cannot start tunnel in state: ${record.state}`); } @@ -1103,6 +1104,7 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService // lock while SSH connects so drain/destroy can fence an indefinitely reconnecting start. startup = tunnels.start({ ...request, + bundleHash: record.bootstrapReceipt.bundleHash, gateway, ssh: record.sshEndpoint, resolveIdentity: identityResolverFor(record, provider, record.leaseId), diff --git a/src/gateway/worker-environments/tunnel.test-support.ts b/src/gateway/worker-environments/tunnel.test-support.ts index 2950ed859bd6..0fe0c823d69c 100644 --- a/src/gateway/worker-environments/tunnel.test-support.ts +++ b/src/gateway/worker-environments/tunnel.test-support.ts @@ -1,5 +1,7 @@ +import { createHash } from "node:crypto"; import fs from "node:fs/promises"; import path from "node:path"; +import { pathToFileURL } from "node:url"; import { expect, vi } from "vitest"; import type { WorkerSshEndpoint } from "../../plugins/types.js"; import { @@ -32,6 +34,7 @@ export const SSH: WorkerSshEndpoint = { hostKey: HOST_KEY, keyRef: { source: "file", provider: "workers", id: "/identity" }, }; +export const BUNDLE_HASH = "a".repeat(64); export const PWD_COMMAND = { transportRetry: "idempotent", argv: ["pwd"] } as const; export function success(stdout = "", stderr = ""): SpawnResult { @@ -47,7 +50,75 @@ export function success(stdout = "", stderr = ""): SpawnResult { export function rsyncReceiverNonce(argv: readonly string[]): string | undefined { const remotePath = argv.find((arg) => arg.startsWith("--rsync-path=")); - return remotePath ? /'([a-f0-9]{32})' '[^']+'$/u.exec(remotePath)?.[1] : undefined; + const words = remotePath?.slice("--rsync-path=".length).split(" "); + return words?.length === 5 && /^[a-f0-9]{32}$/u.test(words[4] ?? "") ? words[4] : undefined; +} + +function rsyncReceiverInvocation(argv: readonly string[]) { + const remotePath = argv.find((arg) => arg.startsWith("--rsync-path=")); + const words = remotePath?.slice("--rsync-path=".length).split(" "); + if (!words || words.length !== 5 || words[0] !== "node") { + return undefined; + } + const [node, receiverEntryPath, mode, encodedContext, nonce] = words; + if ( + !node || + !receiverEntryPath || + !["workspace-root", "git-pack", "accepted-next"].includes(mode ?? "") || + !encodedContext || + !/^[a-f0-9]{32}$/u.test(nonce ?? "") + ) { + return undefined; + } + const [workspace] = JSON.parse(Buffer.from(encodedContext, "base64url").toString("utf8")) as [ + string, + string, + string, + ]; + const target = + mode === "git-pack" + ? path.join(workspace, ".openclaw-base.pack") + : mode === "accepted-next" + ? path.join( + path.dirname(workspace), + `.openclaw-accepted-${createHash("sha256").update(workspace).digest("hex")}-${nonce}`, + "next", + ) + : workspace; + return { receiverEntryPath, target }; +} + +export async function prepareLocalWorkspaceRsyncBoundary( + remoteHome: string, + argv: readonly string[], +): Promise<{ argv: string[]; receiverTarget: string }> { + const invocation = rsyncReceiverInvocation(argv); + if (!invocation) { + throw new Error("test rsync transfer is missing its bundled receiver invocation"); + } + const receiverEntry = path.join(remoteHome, invocation.receiverEntryPath); + const installRoot = path.join(remoteHome, ".openclaw-worker", BUNDLE_HASH); + await fs.mkdir(path.dirname(receiverEntry), { recursive: true }); + await fs.writeFile(path.join(installRoot, "package.json"), '{"type":"module"}\n'); + const tsxApi = import.meta.resolve("tsx/esm/api"); + const sourceEntry = pathToFileURL(path.resolve("src/worker/workspace-rsync-receiver.ts")).href; + await fs.writeFile( + receiverEntry, + `import { tsImport } from ${JSON.stringify(tsxApi)};\nawait tsImport(${JSON.stringify(sourceEntry)}, import.meta.url);\n`, + ); + const fakeSsh = path.join(remoteHome, ".openclaw-test-ssh"); + await fs.writeFile( + fakeSsh, + '#!/bin/sh\nset -eu\nwhile [ "$#" -gt 0 ]; do\n case "$1" in -l|-p) shift 2 ;; -*) shift ;; *) shift; break ;; esac\ndone\ncd "$HOME"\nif [ -n "${OPENCLAW_TEST_RECEIVER_PATH:-}" ]; then PATH=$OPENCLAW_TEST_RECEIVER_PATH; export PATH; fi\nexec sh -c "$*"\n', + { mode: 0o755 }, + ); + const localArgv = [...argv]; + const remoteShellIndex = localArgv.indexOf("-e"); + if (remoteShellIndex < 0) { + throw new Error("test rsync transfer is missing its remote shell"); + } + localArgv[remoteShellIndex + 1] = fakeSsh; + return { argv: localArgv, receiverTarget: invocation.target }; } function shellQuoted(value: string): string { @@ -183,6 +254,7 @@ export function localWorkspaceRunner( argv: string[], localArgv: string[], options: CommandOptions, + receiverTarget?: string, ) => Promise, onCommandCompleted?: (argv: readonly string[], result: SpawnResult) => void, ) { @@ -200,17 +272,25 @@ export function localWorkspaceRunner( return await runCommandWithTimeout(argv, options); } if (argv[0] === "rsync") { + if (argv.some((arg) => arg.startsWith("--rsync-path="))) { + const boundary = await prepareLocalWorkspaceRsyncBoundary(remoteHome, argv); + const boundaryOptions = { + ...options, + baseEnv: { ...options.baseEnv, HOME: remoteHome }, + }; + const intercepted = await onRsync?.( + argv, + boundary.argv, + boundaryOptions, + boundary.receiverTarget, + ); + return intercepted ?? (await runCommandWithTimeout(boundary.argv, boundaryOptions)); + } const localArgv = [...argv]; const remoteShellIndex = localArgv.indexOf("-e"); if (remoteShellIndex >= 0) { localArgv.splice(remoteShellIndex, 2); } - const remoteReceiverIndex = localArgv.findIndex((arg) => arg.startsWith("--rsync-path=")); - if (remoteReceiverIndex >= 0) { - // Local transfers do not have a remote shell; boundary cases that need - // receiver ownership launch the production wrapper explicitly below. - localArgv.splice(remoteReceiverIndex, 1); - } for (let index = localArgv.indexOf("--") + 1; index < localArgv.length; index += 1) { const candidate = localArgv[index]; const separator = candidate?.indexOf(":") ?? -1; @@ -292,6 +372,7 @@ export function startTestTunnel( return manager.start({ environmentId, ownerEpoch, + bundleHash: BUNDLE_HASH, ssh, gateway: { host: "127.0.0.1", port: 18789 }, resolveIdentity, diff --git a/src/gateway/worker-environments/tunnel.test.ts b/src/gateway/worker-environments/tunnel.test.ts index 1281b9cd20cc..bded2000cb45 100644 --- a/src/gateway/worker-environments/tunnel.test.ts +++ b/src/gateway/worker-environments/tunnel.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, vi } from "vitest"; import { createWorkerSshRunner } from "./tunnel-ssh-runner.js"; import { createWorkerTunnelManager } from "./tunnel.js"; import { + BUNDLE_HASH, PWD_COMMAND, SSH, deferred, @@ -137,6 +138,7 @@ describe("worker tunnel manager", () => { sleep: async () => {}, }); const request = { + bundleHash: BUNDLE_HASH, environmentId: "worker:port-reconnect", ownerEpoch: 1, ssh: { ...SSH, port: 2222, fallbackPorts: [22] }, @@ -280,6 +282,7 @@ describe("worker tunnel manager", () => { await sleepStarted.promise; const reconnecting = manager.start({ + bundleHash: BUNDLE_HASH, environmentId: "worker:drain", ownerEpoch: 8, ssh: SSH, @@ -316,6 +319,7 @@ describe("worker tunnel manager", () => { const fake = fakeRunner(); const manager = createWorkerTunnelManager({ runner: fake.runner, sleep: async () => {} }); const initialRequest = { + bundleHash: BUNDLE_HASH, environmentId: "worker:replacement", ownerEpoch: 1, ssh: SSH, @@ -335,6 +339,7 @@ describe("worker tunnel manager", () => { const releaseStop = deferred(); staleReconnect.blockStopUntil(releaseStop.promise); const replacement = manager.start({ + bundleHash: BUNDLE_HASH, environmentId: "worker:replacement", ownerEpoch: 2, ssh: SSH, diff --git a/src/gateway/worker-environments/tunnel.ts b/src/gateway/worker-environments/tunnel.ts index 5ad24356cd73..1d58ba8e3aef 100644 --- a/src/gateway/worker-environments/tunnel.ts +++ b/src/gateway/worker-environments/tunnel.ts @@ -77,12 +77,14 @@ rmdir -- "$directory" 2>/dev/null || true `; type WorkerTunnelStartRequest = WorkerTunnelRequest & { + bundleHash: string; gateway: { host: "127.0.0.1" | "::1"; port: number }; ssh: WorkerSshEndpoint; resolveIdentity: WorkerSshIdentityResolver; }; type TunnelEntry = { + bundleHash: string; environmentId: string; ownerEpoch: number; gateway: WorkerTunnelStartRequest["gateway"]; @@ -229,6 +231,7 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = getPrepared: () => entry.prepared, runner, tasks: entry.workspaceTasks, + bundleHash: entry.bundleHash, }), stop: () => stop(entry.environmentId, entry.ownerEpoch), }); @@ -413,6 +416,7 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = void readiness.promise.catch(() => undefined); const entry: TunnelEntry = { environmentId: request.environmentId, + bundleHash: request.bundleHash, ownerEpoch: request.ownerEpoch, gateway: request.gateway, remoteDirectory, diff --git a/src/gateway/worker-environments/workspace-accepted-sync.test.ts b/src/gateway/worker-environments/workspace-accepted-sync.test.ts index 489897b2130d..b09b2c47e020 100644 --- a/src/gateway/worker-environments/workspace-accepted-sync.test.ts +++ b/src/gateway/worker-environments/workspace-accepted-sync.test.ts @@ -11,6 +11,7 @@ import { WorkerTunnelOwnerDisconnectedError, type WorkerWorkspaceCommand, } from "./tunnel-contract.js"; +import { BUNDLE_HASH, prepareLocalWorkspaceRsyncBoundary } from "./tunnel.test-support.js"; import { AcceptedWorkspacePublicationIndeterminateError, isAcceptedWorkspacePublicationIndeterminateError, @@ -23,12 +24,14 @@ import { serializeWorkerWorkspaceManifest, type WorkerWorkspaceManifest, } from "./workspace-manifest.js"; +import { workerWorkspaceRsyncReceiverEntryPath } from "./workspace-sync-helpers.js"; import { REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS, REMOTE_WORKSPACE_MANIFEST_JS, } from "./workspace-sync-scripts.js"; const tempDirs = useAutoCleanupTempDirTracker(afterEach); +const RECEIVER_ENTRY_PATH = workerWorkspaceRsyncReceiverEntryPath(BUNDLE_HASH); function result(overrides: Partial = {}): SpawnResult { return { @@ -66,10 +69,6 @@ function settlement(outcome: "begun" | "rolled-back" | "applied" | "committed"): return result({ stdout: `${JSON.stringify({ version: 1, outcome })}\n` }); } -function shellQuoted(value: string): string { - return `'${value.replaceAll("'", `'"'"'`)}'`; -} - describe("accepted workspace publication", () => { it.skipIf(process.platform === "win32")( "waits for the staging receiver group before promoting its inodes live", @@ -118,7 +117,7 @@ describe("accepted workspace publication", () => { const env = { ...process.env, HOME: home, - PATH: `${bin}:${process.env.PATH ?? ""}`, + OPENCLAW_TEST_RECEIVER_PATH: `${bin}:${process.env.PATH ?? ""}`, OPENCLAW_TEST_RECEIVER_GATE: gate, OPENCLAW_TEST_RECEIVER_MARKER: receiverMarker, }; @@ -138,27 +137,15 @@ describe("accepted workspace publication", () => { }); }; const publisher = createAcceptedWorkspacePublisherFactory({ + receiverEntryPath: RECEIVER_ENTRY_PATH, runWorkspaceCommand, runRsync: async (argvForSsh) => { const argv = argvForSsh("ssh"); - const receiverPath = argv.find((entry) => entry.startsWith("--rsync-path=")); - if (!receiverPath) { - throw new Error("accepted staging transfer is missing its receiver wrapper"); - } - const stagingRoot = argv.at(-1)?.slice("test:".length).replace(/\/$/u, ""); - if (!stagingRoot) { - throw new Error("accepted staging transfer is missing its destination"); - } - receiverChild = spawn( - "sh", - [ - "-c", - `${receiverPath.slice("--rsync-path=".length)} '--server' '.' ${shellQuoted( - stagingRoot, - )}`, - ], - { env, stdio: ["ignore", "ignore", "pipe"] }, - ); + const boundary = await prepareLocalWorkspaceRsyncBoundary(home, argv); + receiverChild = spawn(boundary.argv[0]!, boundary.argv.slice(1), { + env, + stdio: ["ignore", "ignore", "pipe"], + }); const receiverStderr = receiverChild.stderr; if (!receiverStderr) { throw new Error("accepted staging receiver has no stderr pipe"); @@ -217,7 +204,12 @@ describe("accepted workspace publication", () => { ); await releaseReceiver("release"); - await expect(receiverExited).resolves.toMatchObject({ code: 0, signal: null, stderr: "" }); + if (!receiverExited) { + throw new Error("accepted staging receiver did not start"); + } + const receiverExit = await receiverExited; + expect(receiverExit.signal).toBeNull(); + expect(receiverExit.code).not.toBe(0); await expect(publishing).resolves.toBeUndefined(); expect(actions).toEqual(["begin", "apply", "commit"]); await expect(fs.readFile(path.join(workspace, "result.txt"), "utf8")).resolves.toBe( @@ -341,6 +333,7 @@ fs.renameSync = function(source, destination) { return result(); }; const publisher = createAcceptedWorkspacePublisherFactory({ + receiverEntryPath: RECEIVER_ENTRY_PATH, runWorkspaceCommand, runRsync, scpTarget: "test", @@ -519,6 +512,7 @@ Atomics.wait = function(waitArray, index, value, timeout) { return commandResult; }; const publisher = createAcceptedWorkspacePublisherFactory({ + receiverEntryPath: RECEIVER_ENTRY_PATH, runWorkspaceCommand, runRsync: async () => { if (!stagingRoot) { @@ -582,6 +576,7 @@ Atomics.wait = function(waitArray, index, value, timeout) { const accepted = manifest("local\n"); const actions: string[] = []; const factory = createAcceptedWorkspacePublisherFactory({ + receiverEntryPath: RECEIVER_ENTRY_PATH, runWorkspaceCommand: async (command) => { if (command.argv[2] !== REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS) { return result(); @@ -648,6 +643,7 @@ Atomics.wait = function(waitArray, index, value, timeout) { const accepted = manifest("local\n"); const actions: string[] = []; const publisher = createAcceptedWorkspacePublisherFactory({ + receiverEntryPath: RECEIVER_ENTRY_PATH, runWorkspaceCommand: async (command) => { if (command.argv[2] !== REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS) { return result(); @@ -689,6 +685,7 @@ Atomics.wait = function(waitArray, index, value, timeout) { const accepted = manifest("local\n"); const actions: string[] = []; const publisher = createAcceptedWorkspacePublisherFactory({ + receiverEntryPath: RECEIVER_ENTRY_PATH, runWorkspaceCommand: async (command) => { if (command.argv[2] !== REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS) { return result(); @@ -774,6 +771,7 @@ Atomics.wait = function(waitArray, index, value, timeout) { const transactionCalls: Array<{ action: string; nonce: string }> = []; let commitCount = 0; const publisher = createAcceptedWorkspacePublisherFactory({ + receiverEntryPath: RECEIVER_ENTRY_PATH, runWorkspaceCommand: async (command) => { if (command.argv[2] !== REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS) { return result({ stdout: command.argv[5] === "publish" ? "" : `${acceptedRef}\n` }); @@ -836,6 +834,7 @@ Atomics.wait = function(waitArray, index, value, timeout) { const acceptedRef = manifestRef(accepted); const actions: string[] = []; const publisher = createAcceptedWorkspacePublisherFactory({ + receiverEntryPath: RECEIVER_ENTRY_PATH, runWorkspaceCommand: async (command) => { if (command.argv[2] !== REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS) { return result({ stdout: command.argv[5] === "publish" ? "" : `${acceptedRef}\n` }); diff --git a/src/gateway/worker-environments/workspace-accepted-sync.ts b/src/gateway/worker-environments/workspace-accepted-sync.ts index 20deafd08870..a4abc636bf5d 100644 --- a/src/gateway/worker-environments/workspace-accepted-sync.ts +++ b/src/gateway/worker-environments/workspace-accepted-sync.ts @@ -17,6 +17,7 @@ import { import { changedPaths, manifestNodes } from "./workspace-reconcile.js"; import { parseManifestRef, + WORKER_WORKSPACE_RSYNC_DESTINATION, workerAcceptedWorkspaceRsyncReceiverPath, workerWorkspaceCommandSucceeded, workspaceSyncError, @@ -62,6 +63,7 @@ function createAcceptedWorkspacePublisher(params: { runWorkspaceCommand: (command: WorkerWorkspaceCommand) => Promise; runRsync: (argv: (rsyncSsh: string) => string[]) => Promise; scpTarget: string; + receiverEntryPath: string; localPath: string; remoteWorkspaceDir: string; remoteManifest: WorkerWorkspaceManifest; @@ -246,6 +248,7 @@ function createAcceptedWorkspacePublisher(params: { "--from0", `--files-from=${transferListPath}`, `--rsync-path=${workerAcceptedWorkspaceRsyncReceiverPath({ + receiverEntryPath: params.receiverEntryPath, remoteWorkspaceDir: params.remoteWorkspaceDir, nonce: transactionNonce, })}`, @@ -253,7 +256,7 @@ function createAcceptedWorkspacePublisher(params: { rsyncSsh, "--", localSource, - `${params.scpTarget}:${remoteStagingRoot}/`, + `${params.scpTarget}:${WORKER_WORKSPACE_RSYNC_DESTINATION}`, ]); if (!workerWorkspaceCommandSucceeded(transferred)) { throw workspaceSyncError(transferred); diff --git a/src/gateway/worker-environments/workspace-mutation-remote-script.test.ts b/src/gateway/worker-environments/workspace-mutation-remote-script.test.ts index 8831d3ba9d98..128e7287885d 100644 --- a/src/gateway/worker-environments/workspace-mutation-remote-script.test.ts +++ b/src/gateway/worker-environments/workspace-mutation-remote-script.test.ts @@ -6,10 +6,13 @@ import { afterEach, describe, expect, it } from "vitest"; import { waitForChildClose, waitForDead, waitForFile } from "../../../test/helpers/process-wait.js"; import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; import { runCommandWithTimeout } from "../../process/exec.js"; +import { BUNDLE_HASH, prepareLocalWorkspaceRsyncBoundary } from "./tunnel.test-support.js"; +import { REMOTE_GIT_WORKSPACE_RETRY_RESET_JS } from "./workspace-mutation-remote-script.js"; import { - REMOTE_GIT_WORKSPACE_RETRY_RESET_JS, - REMOTE_WORKSPACE_RSYNC_RECEIVER_JS, -} from "./workspace-mutation-remote-script.js"; + createWorkerWorkspaceRsyncReceiverPathFactory, + WORKER_WORKSPACE_RSYNC_DESTINATION, + workerWorkspaceRsyncReceiverEntryPath, +} from "./workspace-sync-helpers.js"; const tempDirs = useAutoCleanupTempDirTracker(afterEach); @@ -79,18 +82,32 @@ process.kill = function(pid, signal) { OPENCLAW_TEST_RESET_NONCE: resetNonce, OPENCLAW_TEST_CONTENDER_MARKER: contenderMarker, }; + const receiverCommand = createWorkerWorkspaceRsyncReceiverPathFactory({ + receiverEntryPath: workerWorkspaceRsyncReceiverEntryPath(BUNDLE_HASH), + remoteWorkspaceDir: workspace, + canonicalHome: home, + remoteRelative: relative, + })("workspace-root"); + const boundary = await prepareLocalWorkspaceRsyncBoundary(home, [ + "rsync", + `--rsync-path=${receiverCommand}`, + "-e", + "ssh", + "--", + "source", + `test:${WORKER_WORKSPACE_RSYNC_DESTINATION}`, + ]); + const [node, receiverEntry, mode, context] = receiverCommand.split(" "); + expect(node).toBe("node"); const receiver = spawnTransaction( [ - "-e", - REMOTE_WORKSPACE_RSYNC_RECEIVER_JS, - workspace, - home, - relative, + path.join(home, receiverEntry!), + mode!, + context!, receiverNonce, - workspace, "--server", ".", - `${workspace}/`, + boundary.argv.at(-1)!.slice("test:".length), ], env, ); diff --git a/src/gateway/worker-environments/workspace-rsync-path.test.ts b/src/gateway/worker-environments/workspace-rsync-path.test.ts new file mode 100644 index 000000000000..7298307d3ef3 --- /dev/null +++ b/src/gateway/worker-environments/workspace-rsync-path.test.ts @@ -0,0 +1,156 @@ +import { createHash } from "node:crypto"; +import fs from "node:fs/promises"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; +import { afterEach, describe, expect, it } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; +import { runCommandWithTimeout } from "../../process/exec.js"; +import { + createWorkerWorkspaceRsyncReceiverPathFactory, + WORKER_WORKSPACE_RSYNC_DESTINATION, + workerAcceptedWorkspaceRsyncReceiverPath, + workerWorkspaceRsyncReceiverEntryPath, +} from "./workspace-sync-helpers.js"; + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); +const BUNDLE_HASH = "a".repeat(64); + +describe.skipIf(process.platform === "win32")("workspace rsync receiver path", () => { + it.each([ + { mode: "workspace-root", sourceKind: "directory" }, + { mode: "git-pack", sourceKind: "file" }, + { mode: "accepted-next", sourceKind: "file" }, + ] as const)("crosses the real rsync and OpenSSH argv boundary for $mode", async (testCase) => { + const root = path.join(tempDirs.make("openclaw-rsync-path-"), "paths with spaces"); + const home = path.join(root, "remote home"); + const workspace = path.join(home, ".openclaw-worker/workspaces/env/session/1"); + const source = path.join( + root, + testCase.sourceKind === "directory" ? "source dir" : "input.bin", + ); + const tools = tempDirs.make("openclaw-rsync-tools-"); + const sshArgvPath = path.join(root, "ssh-argv"); + const receiverArgvPath = path.join(root, "receiver-argv"); + await fs.mkdir(workspace, { recursive: true }); + await fs.mkdir(tools, { recursive: true }); + const canonicalHome = await fs.realpath(home); + const canonicalWorkspace = await fs.realpath(workspace); + const remoteRelative = path.posix.relative(canonicalHome, canonicalWorkspace); + const nonce = "b".repeat(32); + const receiverEntryPath = workerWorkspaceRsyncReceiverEntryPath(BUNDLE_HASH); + const installRoot = path.join(canonicalHome, ".openclaw-worker", BUNDLE_HASH); + const receiverEntry = path.join(canonicalHome, receiverEntryPath); + await fs.mkdir(path.dirname(receiverEntry), { recursive: true }); + await fs.writeFile(path.join(installRoot, "package.json"), '{"type":"module"}\n'); + const tsxApi = import.meta.resolve("tsx/esm/api"); + const sourceEntry = pathToFileURL(path.resolve("src/worker/workspace-rsync-receiver.ts")).href; + await fs.writeFile( + receiverEntry, + `import { tsImport } from ${JSON.stringify(tsxApi)};\nawait tsImport(${JSON.stringify(sourceEntry)}, import.meta.url);\n`, + ); + + const resolvedRsync = await runCommandWithTimeout(["sh", "-c", "command -v rsync"], { + timeoutMs: 10_000, + }); + expect(resolvedRsync).toMatchObject({ termination: "exit", code: 0 }); + const rsync = resolvedRsync.stdout.trim(); + await fs.writeFile( + path.join(tools, "rsync"), + '#!/bin/sh\nset -eu\nprintf "%s\\0" "$@" > "$OPENCLAW_TEST_RECEIVER_ARGV"\nexec "$OPENCLAW_TEST_REAL_RSYNC" "$@"\n', + { mode: 0o755 }, + ); + const fakeSsh = path.join(tools, "ssh"); + await fs.writeFile( + fakeSsh, + '#!/bin/sh\nset -eu\nshift\nprintf "%s\\0" "$@" > "$OPENCLAW_TEST_SSH_ARGV"\ncd "$HOME"\nexec sh -c "$*"\n', + { mode: 0o755 }, + ); + + let receiverCommand: string; + let receiverTarget: string; + if (testCase.mode === "accepted-next") { + const workspaceKey = createHash("sha256").update(canonicalWorkspace).digest("hex"); + const transaction = path.join( + path.dirname(canonicalWorkspace), + `.openclaw-accepted-${workspaceKey}-${nonce}`, + ); + receiverTarget = path.join(transaction, "next"); + await fs.mkdir(receiverTarget, { recursive: true }); + await fs.writeFile( + path.join(transaction, "phase.json"), + JSON.stringify({ version: 1, nonce, phase: "begun" }), + ); + receiverCommand = workerAcceptedWorkspaceRsyncReceiverPath({ + receiverEntryPath, + remoteWorkspaceDir: canonicalWorkspace, + nonce, + }); + } else { + receiverTarget = + testCase.mode === "git-pack" + ? path.join(canonicalWorkspace, ".openclaw-base.pack") + : canonicalWorkspace; + receiverCommand = createWorkerWorkspaceRsyncReceiverPathFactory({ + receiverEntryPath, + remoteWorkspaceDir: canonicalWorkspace, + canonicalHome, + remoteRelative, + })(testCase.mode); + } + + const contents = `received through ${testCase.mode}\n`; + if (testCase.sourceKind === "directory") { + await fs.mkdir(source, { recursive: true }); + await fs.writeFile(path.join(source, "payload.txt"), contents); + } else { + await fs.writeFile(source, contents); + } + const result = await runCommandWithTimeout( + [ + rsync, + "--archive", + "--checksum", + `--rsync-path=${receiverCommand}`, + "-e", + fakeSsh, + "--", + testCase.sourceKind === "directory" ? `${source}/` : source, + `test:${WORKER_WORKSPACE_RSYNC_DESTINATION}`, + ], + { + timeoutMs: 30_000, + baseEnv: { + ...process.env, + HOME: canonicalHome, + PATH: `${tools}:${process.env.PATH ?? ""}`, + OPENCLAW_TEST_REAL_RSYNC: rsync, + OPENCLAW_TEST_RECEIVER_ARGV: receiverArgvPath, + OPENCLAW_TEST_SSH_ARGV: sshArgvPath, + }, + }, + ); + expect(result).toMatchObject({ termination: "exit", code: 0 }); + const receivedPath = + testCase.mode === "workspace-root" + ? path.join(receiverTarget, "payload.txt") + : testCase.mode === "accepted-next" + ? path.join(receiverTarget, path.basename(source)) + : receiverTarget; + await expect(fs.readFile(receivedPath, "utf8")).resolves.toBe(contents); + + const sshArgv = (await fs.readFile(sshArgvPath)).toString().split("\0").filter(Boolean); + expect(sshArgv.join(" ")).toMatch( + /^node [A-Za-z0-9_./-]+ (?:workspace-root|git-pack|accepted-next) [A-Za-z0-9_-]+ [a-f0-9]{32} --server /u, + ); + expect(sshArgv.at(-1)).toBe(WORKER_WORKSPACE_RSYNC_DESTINATION); + expect(sshArgv.join(" ")).not.toContain(canonicalWorkspace); + const receiverArgv = (await fs.readFile(receiverArgvPath)) + .toString() + .split("\0") + .filter(Boolean); + expect(receiverArgv.at(-1)).toBe(receiverTarget); + await expect( + fs.access(path.join(canonicalHome, WORKER_WORKSPACE_RSYNC_DESTINATION)), + ).rejects.toMatchObject({ code: "ENOENT" }); + }); +}); diff --git a/src/gateway/worker-environments/workspace-sync-helpers.ts b/src/gateway/worker-environments/workspace-sync-helpers.ts index d2dbc2f8f678..c762c8982d3b 100644 --- a/src/gateway/worker-environments/workspace-sync-helpers.ts +++ b/src/gateway/worker-environments/workspace-sync-helpers.ts @@ -12,13 +12,10 @@ import { workerSshRemoteCommand, } from "./ssh.js"; import type { WorkerWorkspaceCommand, WorkerWorkspaceSyncRequest } from "./tunnel-contract.js"; -import { - REMOTE_WORKSPACE_ACCEPTED_RSYNC_RECEIVER_JS, - REMOTE_WORKSPACE_MANIFEST_JS, - REMOTE_WORKSPACE_RSYNC_RECEIVER_JS, -} from "./workspace-sync-scripts.js"; +import { REMOTE_WORKSPACE_MANIFEST_JS } from "./workspace-sync-scripts.js"; const MANIFEST_REF_PATTERN = /^sha256:[a-f0-9]{64}$/u; +export const WORKER_WORKSPACE_RSYNC_DESTINATION = "openclaw-rsync-destination"; export type WorkerWorkspaceActionsOptions = { environmentId: string; @@ -27,6 +24,7 @@ export type WorkerWorkspaceActionsOptions = { getPrepared: () => PreparedWorkerSsh | undefined; runner: { run(argv: string[], options: CommandOptions): Promise }; tasks: Set>; + bundleHash: string; }; export function waitForQuiescenceRenewal( @@ -77,39 +75,42 @@ export function workerWorkspaceRsyncRemoteCommand( ]); } +type WorkerWorkspaceRsyncReceiverMode = "accepted-next" | "git-pack" | "workspace-root"; + function workerWorkspaceRsyncReceiverPath(params: { + receiverEntryPath: string; remoteWorkspaceDir: string; canonicalHome: string; remoteRelative: string; - remoteTarget: string; + mode: WorkerWorkspaceRsyncReceiverMode; nonce: string; }): string { - return workerSshRemoteCommand([ - "node", - "-e", - REMOTE_WORKSPACE_RSYNC_RECEIVER_JS, - params.remoteWorkspaceDir, - params.canonicalHome, - params.remoteRelative, - params.nonce, - params.remoteTarget, - ]); + const context = Buffer.from( + JSON.stringify([params.remoteWorkspaceDir, params.canonicalHome, params.remoteRelative]), + ).toString("base64url"); + const command = ["node", params.receiverEntryPath, params.mode, context, params.nonce]; + if (command.some((word) => !/^[A-Za-z0-9_./-]+$/u.test(word))) { + throw new Error("Worker workspace rsync receiver command is not shell-safe"); + } + return command.join(" "); } export function createWorkerWorkspaceRsyncReceiverPathFactory(params: { + receiverEntryPath: string; remoteWorkspaceDir: string; canonicalHome: string; remoteRelative: string; -}): (remoteTarget: string) => string { - return (remoteTarget) => +}): (mode: "git-pack" | "workspace-root") => string { + return (mode) => workerWorkspaceRsyncReceiverPath({ ...params, - remoteTarget, + mode, nonce: randomBytes(16).toString("hex"), }); } export function workerAcceptedWorkspaceRsyncReceiverPath(params: { + receiverEntryPath: string; remoteWorkspaceDir: string; nonce: string; }): string { @@ -120,15 +121,21 @@ export function workerAcceptedWorkspaceRsyncReceiverPath(params: { } const canonicalHome = params.remoteWorkspaceDir.slice(0, markerIndex); const remoteRelative = params.remoteWorkspaceDir.slice(markerIndex + 1); - return workerSshRemoteCommand([ - "node", - "-e", - REMOTE_WORKSPACE_ACCEPTED_RSYNC_RECEIVER_JS, - params.remoteWorkspaceDir, + return workerWorkspaceRsyncReceiverPath({ + receiverEntryPath: params.receiverEntryPath, + remoteWorkspaceDir: params.remoteWorkspaceDir, canonicalHome, remoteRelative, - params.nonce, - ]); + mode: "accepted-next", + nonce: params.nonce, + }); +} + +export function workerWorkspaceRsyncReceiverEntryPath(bundleHash: string): string { + if (!/^[a-f0-9]{64}$/u.test(bundleHash)) { + throw new Error("Worker workspace rsync receiver bundle hash is invalid"); + } + return `.openclaw-worker/${bundleHash}/dist/worker/workspace-rsync-receiver.js`; } export function workerWorkspaceSshArgv( diff --git a/src/gateway/worker-environments/workspace-sync-scripts.ts b/src/gateway/worker-environments/workspace-sync-scripts.ts index 98b5027981ba..4c35d698372a 100644 --- a/src/gateway/worker-environments/workspace-sync-scripts.ts +++ b/src/gateway/worker-environments/workspace-sync-scripts.ts @@ -2,14 +2,8 @@ import { REMOTE_WORKSPACE_MANIFEST_CANONICAL_JS, REMOTE_WORKSPACE_MANIFEST_REGISTRY_JS, } from "./workspace-manifest-remote-script.js"; -export { - REMOTE_WORKSPACE_ACCEPTED_RSYNC_RECEIVER_JS, - REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS, -} from "./workspace-accepted-remote-script.js"; -export { - REMOTE_GIT_WORKSPACE_RETRY_RESET_JS, - REMOTE_WORKSPACE_RSYNC_RECEIVER_JS, -} from "./workspace-mutation-remote-script.js"; +export { REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS } from "./workspace-accepted-remote-script.js"; +export { REMOTE_GIT_WORKSPACE_RETRY_RESET_JS } from "./workspace-mutation-remote-script.js"; import { DERIVED_WORKSPACE_DIRECTORY_NAMES, DERIVED_WORKSPACE_FILE_NAMES, diff --git a/src/gateway/worker-environments/workspace-sync-tunnel.test.ts b/src/gateway/worker-environments/workspace-sync-tunnel.test.ts index 604e83d4dbf3..aabda52a7bc4 100644 --- a/src/gateway/worker-environments/workspace-sync-tunnel.test.ts +++ b/src/gateway/worker-environments/workspace-sync-tunnel.test.ts @@ -8,6 +8,7 @@ import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js" import { runCommandWithTimeout } from "../../process/exec.js"; import { createWorkerTunnelManager } from "./tunnel.js"; import { + BUNDLE_HASH, PWD_COMMAND, SSH, fakeRunner, @@ -24,7 +25,6 @@ import { workspaceSetup, } from "./tunnel.test-support.js"; import { rsyncArgvPort, sshArgvPort } from "./worker-ssh-argv.test-support.js"; -import { REMOTE_WORKSPACE_RSYNC_RECEIVER_JS } from "./workspace-mutation-remote-script.js"; import { parseWorkerWorkspaceManifest } from "./workspace-reconcile.js"; import { stableWorkerPathComponent } from "./workspace-sync.js"; @@ -86,7 +86,7 @@ describe("worker tunnel manager", () => { const transfer = outboundTransfers.at(-1); expect(transfer?.argv).toContain("--checksum"); expect(transfer?.argv).toContain(`${localPath}/`); - expect(transfer?.argv.at(-1)).toBe(`worker@worker.example.test:${remoteWorkspaceDir}/`); + expect(transfer?.argv.at(-1)).toBe("worker@worker.example.test:openclaw-rsync-destination"); expect(transfer?.argv).not.toContain("--protect-args"); expect(transfer?.argv.some((arg) => arg.startsWith("--files-from="))).toBe(true); const remoteShell = transfer?.argv[transfer.argv.indexOf("-e") + 1]; @@ -305,13 +305,13 @@ describe("worker tunnel manager", () => { const lifecycle: string[] = []; const fake = localWorkspaceRunner( remoteHome, - async (argv, localArgv, options) => { + async (argv, localArgv, options, receiverTarget) => { const isWorkspaceTransfer = argv.some((arg) => arg.startsWith("--files-from=")); if (!primaryTransfer || !isWorkspaceTransfer || rsyncArgvPort(argv) !== 2222) { return undefined; } primaryTransfer = false; - const remoteWorkspaceDir = localArgv.at(-1); + const remoteWorkspaceDir = receiverTarget; if (!remoteWorkspaceDir) { throw new Error("missing test rsync destination"); } @@ -332,32 +332,17 @@ describe("worker tunnel manager", () => { .split(path.sep) .join("/"); receiverRelative = remoteRelative.replace(/\/$/u, ""); - receiverChild = spawn( - process.execPath, - [ - "-e", - REMOTE_WORKSPACE_RSYNC_RECEIVER_JS, - canonicalReceiverWorkspace, - canonicalRemoteHome, - receiverRelative, - "d".repeat(32), - canonicalReceiverWorkspace, - "--server", - ".", - remoteWorkspaceDir, - ], - { - env: { - ...process.env, - HOME: canonicalRemoteHome, - PATH: `${bin}:${process.env.PATH ?? ""}`, - OPENCLAW_TEST_RECEIVER_GATE: receiverGate, - OPENCLAW_TEST_RECEIVER_MARKER: receiverMarker, - OPENCLAW_TEST_RECEIVER_WORKSPACE: remoteWorkspaceDir, - }, - stdio: ["ignore", "ignore", "pipe"], + receiverChild = spawn(localArgv[0]!, localArgv.slice(1), { + env: { + ...process.env, + HOME: canonicalRemoteHome, + OPENCLAW_TEST_RECEIVER_PATH: `${bin}:${process.env.PATH ?? ""}`, + OPENCLAW_TEST_RECEIVER_GATE: receiverGate, + OPENCLAW_TEST_RECEIVER_MARKER: receiverMarker, + OPENCLAW_TEST_RECEIVER_WORKSPACE: remoteWorkspaceDir, }, - ); + stdio: ["ignore", "ignore", "pipe"], + }); receiverChild.stderr?.setEncoding("utf8"); receiverChild.stderr?.on("data", (chunk: string) => { receiverStderr += chunk; @@ -383,6 +368,7 @@ describe("worker tunnel manager", () => { ); const manager = createWorkerTunnelManager({ runner: fake.runner }); const starting = manager.start({ + bundleHash: BUNDLE_HASH, environmentId: "worker:convergent-sync", ownerEpoch: 1, ssh: { ...SSH, port: 2222, fallbackPorts: [22] }, @@ -471,7 +457,12 @@ describe("worker tunnel manager", () => { const gateWriter = await fs.open(receiverGate, "w"); await gateWriter.write("release\n"); await gateWriter.close(); - expect(await receiverExited).toMatchObject({ code: 0, signal: null }); + if (!receiverExited) { + throw new Error("workspace receiver did not start"); + } + const receiverExit = await receiverExited; + expect(receiverExit.signal).toBeNull(); + expect(receiverExit.code).not.toBe(0); const result = await syncing; expect(lifecycle).toEqual(["receiver-exit", `reset-complete:${resetCommands[0]!.nonce}`]); expect(result.mode).toBe("git"); @@ -547,29 +538,33 @@ describe("worker tunnel manager", () => { await git(localPath, "commit", "-m", "base"); let primaryTransfer = true; - const fake = localWorkspaceRunner(remoteHome, async (argv, localArgv, options) => { - if ( - !primaryTransfer || - !argv.some((arg) => arg.startsWith("--files-from=")) || - rsyncArgvPort(argv) !== 2222 - ) { - return undefined; - } - primaryTransfer = false; - const workspace = localArgv.at(-1)?.replace(/\/$/u, ""); - if (!workspace) { - throw new Error("missing test rsync destination"); - } - const transferred = await runCommandWithTimeout(localArgv, options); - if (transferred.termination !== "exit" || transferred.code !== 0) { - throw new Error(transferred.stderr || "test rsync transfer failed"); - } - await fs.rm(workspace, { recursive: true }); - await fs.symlink(unrelated, workspace, "dir"); - return { ...transferred, code: 255, stderr: "primary transport disconnected" }; - }); + const fake = localWorkspaceRunner( + remoteHome, + async (argv, localArgv, options, receiverTarget) => { + if ( + !primaryTransfer || + !argv.some((arg) => arg.startsWith("--files-from=")) || + rsyncArgvPort(argv) !== 2222 + ) { + return undefined; + } + primaryTransfer = false; + const workspace = receiverTarget?.replace(/\/$/u, ""); + if (!workspace) { + throw new Error("missing test rsync destination"); + } + const transferred = await runCommandWithTimeout(localArgv, options); + if (transferred.termination !== "exit" || transferred.code !== 0) { + throw new Error(transferred.stderr || "test rsync transfer failed"); + } + await fs.rm(workspace, { recursive: true }); + await fs.symlink(unrelated, workspace, "dir"); + return { ...transferred, code: 255, stderr: "primary transport disconnected" }; + }, + ); const manager = createWorkerTunnelManager({ runner: fake.runner }); const starting = manager.start({ + bundleHash: BUNDLE_HASH, environmentId: "worker:retry-owner", ownerEpoch: 1, ssh: { ...SSH, port: 2222, fallbackPorts: [22] }, diff --git a/src/gateway/worker-environments/workspace-sync.test.ts b/src/gateway/worker-environments/workspace-sync.test.ts index 02a9f31c6c8f..7eeb4d0abdc4 100644 --- a/src/gateway/worker-environments/workspace-sync.test.ts +++ b/src/gateway/worker-environments/workspace-sync.test.ts @@ -45,6 +45,7 @@ function createWorkspaceActions( ) { const prepared = createPreparedSsh(); return createWorkerWorkspaceActions({ + bundleHash: "a".repeat(64), environmentId: "worker:test", ownerSignal: new AbortController().signal, isConnected: () => true, diff --git a/src/gateway/worker-environments/workspace-sync.ts b/src/gateway/worker-environments/workspace-sync.ts index ed9b6dbc51c5..be89d4a020f9 100644 --- a/src/gateway/worker-environments/workspace-sync.ts +++ b/src/gateway/worker-environments/workspace-sync.ts @@ -44,8 +44,10 @@ import { validateWorkspaceSyncRequest, verifyRemoteWorkspaceManifest, waitForQuiescenceRenewal, + WORKER_WORKSPACE_RSYNC_DESTINATION, workerWorkspaceCommandSucceeded as success, workerWorkspaceRsyncRemoteCommand, + workerWorkspaceRsyncReceiverEntryPath, workerWorkspaceSshArgv, workspaceSyncError, type WorkerWorkspaceActionsOptions, @@ -108,6 +110,7 @@ export function createWorkerWorkspaceActions( runTask, timeoutMs: WORKSPACE_TIMEOUT_MS, }); + const receiverEntryPath = workerWorkspaceRsyncReceiverEntryPath(options.bundleHash); const runWorkspaceCommand = async (command: WorkerWorkspaceCommand): Promise => { const prepared = requirePrepared(); @@ -246,12 +249,10 @@ export function createWorkerWorkspaceActions( ): Promise => { validateWorkspaceSyncRequest(request); const prepared = requirePrepared(); - const environmentKey = stableWorkerPathComponent(options.environmentId, 16); - const sessionKey = stableWorkerPathComponent(request.sessionId, 32); const remoteRelative = [ REMOTE_WORKSPACE_ROOT, - environmentKey, - sessionKey, + stableWorkerPathComponent(options.environmentId, 16), + stableWorkerPathComponent(request.sessionId, 32), String(request.generation), ].join("/"); const setup = await runWorkspaceCommand({ @@ -279,7 +280,12 @@ export function createWorkerWorkspaceActions( path.join(os.tmpdir(), "openclaw-worker-workspace-sync-"), ); try { - const receiverContext = { remoteWorkspaceDir, canonicalHome, remoteRelative }; + const receiverContext = { + receiverEntryPath, + remoteWorkspaceDir, + canonicalHome, + remoteRelative, + }; const mutationReceiverPath = createWorkerWorkspaceRsyncReceiverPathFactory(receiverContext); let prepareGitTransferList: (() => Promise) | undefined; if (mode === "git") { @@ -331,14 +337,12 @@ export function createWorkerWorkspaceActions( "rsync", "--archive", "--checksum", - `--rsync-path=${mutationReceiverPath( - path.posix.join(remoteWorkspaceDir, REMOTE_GIT_PACK_NAME), - )}`, + `--rsync-path=${mutationReceiverPath("git-pack")}`, "-e", rsyncSsh, "--", packPath, - `${prepared.scpTarget}:${remoteWorkspaceDir}/${REMOTE_GIT_PACK_NAME}`, + `${prepared.scpTarget}:${WORKER_WORKSPACE_RSYNC_DESTINATION}`, ]); if (!success(packTransfer)) { throw workspaceSyncError(packTransfer); @@ -383,12 +387,12 @@ export function createWorkerWorkspaceActions( "--exclude=.git", ...DERIVED_WORKSPACE_RSYNC_EXCLUDES.map((pattern) => `--exclude=${pattern}`), ...(fileListPath ? ["--recursive", "--from0", `--files-from=${fileListPath}`] : []), - `--rsync-path=${mutationReceiverPath(remoteWorkspaceDir)}`, + `--rsync-path=${mutationReceiverPath("workspace-root")}`, "-e", rsyncSsh, "--", localSource, - `${prepared.scpTarget}:${remoteWorkspaceDir}/`, + `${prepared.scpTarget}:${WORKER_WORKSPACE_RSYNC_DESTINATION}`, ]; let retryingGitTransfer = false; const transfer = prepareGitTransferList @@ -496,6 +500,7 @@ export function createWorkerWorkspaceActions( runWorkspaceCommand, runRsync: async (argv) => await runRsync(prepared, argv), scpTarget: prepared.scpTarget, + receiverEntryPath, localPath: request.localPath, remoteWorkspaceDir: request.remoteWorkspaceDir, }); @@ -713,13 +718,9 @@ export function createWorkerWorkspaceActions( return { quiesceWorkspace, - reconcileWorkspace(request) { - return track(reconcileWorkspaceImpl(request)); - }, + reconcileWorkspace: (request) => track(reconcileWorkspaceImpl(request)), runWorkspaceCommand, - syncWorkspace(request) { - // Keep the outer task registered across local-file phases so tunnel stop drains all owner work. - return track(syncWorkspaceImpl(request)); - }, + // Keep the outer task registered across local-file phases so tunnel stop drains all owner work. + syncWorkspace: (request) => track(syncWorkspaceImpl(request)), }; } diff --git a/src/scripts/ci-changed-scope.contract-fixtures.test.ts b/src/scripts/ci-changed-scope.contract-fixtures.test.ts index db5336f569ed..dac916d672c9 100644 --- a/src/scripts/ci-changed-scope.contract-fixtures.test.ts +++ b/src/scripts/ci-changed-scope.contract-fixtures.test.ts @@ -18,4 +18,19 @@ describe("shared Apple contract fixture CI scope", () => { runUiTests: false, }); }); + + it.each([ + "src/worker/workspace-rsync-receiver.ts", + "src/gateway/worker-environments/workspace-sync.ts", + "src/gateway/worker-environments/workspace-sync-helpers.ts", + "src/gateway/worker-environments/workspace-accepted-sync.ts", + "src/gateway/worker-environments/workspace-accepted-remote-script.ts", + "src/gateway/worker-environments/workspace-mutation-remote-script.ts", + "src/gateway/worker-environments/workspace-rsync-path.test.ts", + ])("routes workspace rsync receiver owner %s through macOS CI", (ownerPath) => { + expect(detectChangedScope([ownerPath])).toMatchObject({ + runNode: true, + runMacos: true, + }); + }); }); diff --git a/src/worker/workspace-rsync-receiver.ts b/src/worker/workspace-rsync-receiver.ts new file mode 100644 index 000000000000..c1ca77899530 --- /dev/null +++ b/src/worker/workspace-rsync-receiver.ts @@ -0,0 +1,58 @@ +import { createHash } from "node:crypto"; +import { createRequire } from "node:module"; +import path from "node:path"; +import { compileFunction } from "node:vm"; +import { REMOTE_WORKSPACE_ACCEPTED_RSYNC_RECEIVER_JS } from "../gateway/worker-environments/workspace-accepted-remote-script.js"; +import { REMOTE_WORKSPACE_RSYNC_RECEIVER_JS } from "../gateway/worker-environments/workspace-mutation-remote-script.js"; + +const FIXED_DESTINATION = "openclaw-rsync-destination"; +const [mode, encodedContext, nonce, ...receiverArgs] = process.argv.slice(2); +if ( + !/^(?:workspace-root|git-pack|accepted-next)$/u.test(mode ?? "") || + !/^[A-Za-z0-9_-]+$/u.test(encodedContext ?? "") || + !/^[a-f0-9]{32}$/u.test(nonce ?? "") || + receiverArgs.at(-1) !== FIXED_DESTINATION +) { + throw new Error("invalid worker workspace rsync receiver invocation"); +} +const contextBytes = Buffer.from(encodedContext!, "base64url"); +const context: unknown = JSON.parse(contextBytes.toString("utf8")); +if ( + contextBytes.toString("base64url") !== encodedContext || + !Array.isArray(context) || + context.length !== 3 || + !context.every((value) => typeof value === "string") +) { + throw new Error("invalid worker workspace rsync receiver context"); +} +const [workspace, canonicalHome, remoteRelative] = context as [string, string, string]; +const receiverMode = mode!; +const receiverNonce = nonce!; + +const receiverTarget = + receiverMode === "git-pack" + ? path.posix.join(workspace, ".openclaw-base.pack") + : receiverMode === "accepted-next" + ? path.posix.join( + path.posix.dirname(workspace), + `.openclaw-accepted-${createHash("sha256").update(workspace).digest("hex")}-${receiverNonce}`, + "next", + ) + : workspace; +// rsync implementations disagree on quote preservation in --rsync-path. This entry accepts only +// safe words, then restores the attested destination before the shared locked receiver runs. +process.argv = [ + process.argv[0]!, + workspace, + canonicalHome, + remoteRelative, + receiverNonce, + ...(receiverMode === "accepted-next" ? [] : [receiverTarget]), + ...receiverArgs.with(receiverArgs.length - 1, receiverTarget), +]; +compileFunction( + receiverMode === "accepted-next" + ? REMOTE_WORKSPACE_ACCEPTED_RSYNC_RECEIVER_JS + : REMOTE_WORKSPACE_RSYNC_RECEIVER_JS, + ["require"], +)(createRequire(import.meta.url)); diff --git a/test/scripts/changed-lanes.test.ts b/test/scripts/changed-lanes.test.ts index c5ba55634eca..d65368daa34a 100644 --- a/test/scripts/changed-lanes.test.ts +++ b/test/scripts/changed-lanes.test.ts @@ -2021,6 +2021,31 @@ describe("scripts/changed-lanes", () => { } }); + it("runs macOS CI tests for workspace rsync receiver owners", () => { + for (const changedPath of [ + "src/worker/workspace-rsync-receiver.ts", + "src/gateway/worker-environments/workspace-sync.ts", + "src/gateway/worker-environments/workspace-sync-helpers.ts", + "src/gateway/worker-environments/workspace-accepted-sync.ts", + "src/gateway/worker-environments/workspace-accepted-remote-script.ts", + "src/gateway/worker-environments/workspace-mutation-remote-script.ts", + "src/gateway/worker-environments/workspace-rsync-path.test.ts", + ]) { + const plan = createChangedCheckPlan(detectChangedLanes([changedPath]), { + env: { PATH: "/usr/bin" }, + platform: "linux", + swiftlintAvailable: false, + }); + + expect(plan.commands).toContainEqual( + expect.objectContaining({ + name: "macOS app CI tests", + args: ["test:macos:ci"], + }), + ); + } + }); + it("runs the native state schema guard for either contract owner", () => { for (const changedPath of [ "apps/shared/OpenClawKit/Sources/OpenClawNativeState/OpenClawNativeStateSQLite.swift", diff --git a/test/scripts/package-mac-app.test.ts b/test/scripts/package-mac-app.test.ts index 222876b2690d..7c024a3d58e1 100644 --- a/test/scripts/package-mac-app.test.ts +++ b/test/scripts/package-mac-app.test.ts @@ -965,6 +965,7 @@ describe("package-mac-app plist stamping", () => { }; const macosCi = pkg.scripts?.["test:macos:ci"] ?? ""; + expect(macosCi).toContain("src/gateway/worker-environments/workspace-rsync-path.test.ts"); expect(macosCi).toContain("test/scripts/package-mac-app.test.ts"); expect(macosCi).toContain("test/scripts/package-mac-dist.test.ts"); expect(macosCi).toContain("test/scripts/create-dmg.test.ts"); diff --git a/tsdown.config.ts b/tsdown.config.ts index e6a5a1db3fef..bde2e9386fce 100644 --- a/tsdown.config.ts +++ b/tsdown.config.ts @@ -302,6 +302,7 @@ function buildCoreDistEntries(): Record { "provider-dispatcher.runtime": "src/auto-reply/reply/provider-dispatcher.runtime.ts", "server-close.runtime": "src/gateway/server-close.runtime.ts", "gateway/worker-environments/runtime": "src/gateway/worker-environments/runtime.ts", + "worker/workspace-rsync-receiver": "src/worker/workspace-rsync-receiver.ts", "plugins/hook-runner-global": "src/plugins/hook-runner-global.ts", "plugins/memory-state": "src/plugins/memory-state.ts", "plugins/synthetic-auth.runtime": "src/plugins/synthetic-auth.runtime.ts",