From 792f5b7b74e433f9471eed5abcb0fc334d28bdeb Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 19 Jul 2026 20:36:50 -0700 Subject: [PATCH] fix(cloud-workers): reconcile workspace results across turns (#111481) * fix(cloud-workers): preserve accepted workspace manifests * fix(cloud-workers): keep manifest helpers private * fix(cloud-workers): harden accepted workspace publication * fix(cloud-workers): preserve ancestor modes during publication * fix(cloud-workers): pass manifest path filter to verifier * refactor(cloud-workers): extract manifest stability check --- .../worker-environments/tunnel.test.ts | 30 +- .../workspace-accepted-sync.ts | 233 ++++++++ .../workspace-manifest-remote-script.ts | 515 ++++++++++++++++++ .../worker-environments/workspace-manifest.ts | 6 +- .../workspace-reconcile-apply.ts | 24 +- .../workspace-reconcile-core.ts | 5 + .../workspace-reconcile.test.ts | 55 ++ .../workspace-result-staging.ts | 12 + .../workspace-sync-helpers.ts | 112 +++- .../workspace-sync-scripts.test.ts | 311 +++++++++++ .../workspace-sync-scripts.ts | 61 ++- .../worker-environments/workspace-sync.ts | 144 ++--- 12 files changed, 1395 insertions(+), 113 deletions(-) create mode 100644 src/gateway/worker-environments/workspace-accepted-sync.ts create mode 100644 src/gateway/worker-environments/workspace-manifest-remote-script.ts diff --git a/src/gateway/worker-environments/tunnel.test.ts b/src/gateway/worker-environments/tunnel.test.ts index 1be9a3d62ddd..30491faeafd1 100644 --- a/src/gateway/worker-environments/tunnel.test.ts +++ b/src/gateway/worker-environments/tunnel.test.ts @@ -59,14 +59,17 @@ function deferred() { return { promise, resolve, reject }; } -function memoryWorkspaceJournal(): WorkerWorkspaceReconciliationJournalAdapter { +function memoryWorkspaceJournal( + onCommit?: (manifestRef: string) => void, +): WorkerWorkspaceReconciliationJournalAdapter { let pending: WorkerWorkspaceReconciliationJournal | undefined; return { load: () => pending, begin: (journal) => { pending = journal; }, - commit: () => { + commit: (manifestRef) => { + onCommit?.(manifestRef); pending = undefined; }, abort: () => { @@ -506,6 +509,7 @@ describe("worker tunnel manager", () => { fs.writeFile(path.join(localPath, "gone.txt"), "delete me\n"), fs.writeFile(path.join(localPath, "rename-old.txt"), "rename me\n"), fs.writeFile(path.join(localPath, "modified.txt"), "before\n"), + fs.writeFile(path.join(localPath, "conflict.txt"), "base\n"), ]); const largeFiles = Array.from( { length: 1_800 }, @@ -592,6 +596,7 @@ describe("worker tunnel manager", () => { await fs.mkdir(path.join(result.remoteWorkspaceDir, "private")); await Promise.all([ fs.writeFile(path.join(result.remoteWorkspaceDir, "modified.txt"), "worker result\n"), + fs.writeFile(path.join(result.remoteWorkspaceDir, "conflict.txt"), "worker result\n"), fs.appendFile( path.join(result.remoteWorkspaceDir, ".gitignore"), "ordinary-untracked.txt\n", @@ -609,8 +614,12 @@ describe("worker tunnel manager", () => { fs.rm(path.join(result.remoteWorkspaceDir, "rename-new.txt")), fs.symlink("modified.txt", path.join(result.remoteWorkspaceDir, "worker-link")), ]); + await fs.writeFile(path.join(localPath, "conflict.txt"), "local result\n"); - const journal = memoryWorkspaceJournal(); + let acceptedManifestRef = result.manifestRef; + const journal = memoryWorkspaceJournal((manifestRef) => { + acceptedManifestRef = manifestRef; + }); const reconciled = await handle.reconcileWorkspace({ localPath, remoteWorkspaceDir: result.remoteWorkspaceDir, @@ -636,14 +645,23 @@ describe("worker tunnel manager", () => { ).resolves.toBe("allowed\n"); await expect(fs.access(path.join(localPath, "private/worker-secret.txt"))).rejects.toThrow(); await expect(fs.access(path.join(localPath, "rename-new.txt"))).rejects.toThrow(); + await expect(fs.readFile(path.join(localPath, "conflict.txt"), "utf8")).resolves.toBe( + "local result\n", + ); + await expect( + fs.readFile(path.join(result.remoteWorkspaceDir, "conflict.txt"), "utf8"), + ).resolves.toBe("local result\n"); + await expect( + fs.access(path.join(result.remoteWorkspaceDir, "private/ignored.txt")), + ).rejects.toThrow(); expect(await git(localPath, "rev-parse", "HEAD")).toBe(baseCommit); const unchanged = await handle.reconcileWorkspace({ localPath, remoteWorkspaceDir: result.remoteWorkspaceDir, - baseManifestRef: reconciled.manifestRef, + baseManifestRef: acceptedManifestRef, journal, }); - expect(unchanged).toMatchObject({ manifestRef: reconciled.manifestRef, changed: false }); + expect(unchanged).toMatchObject({ manifestRef: acceptedManifestRef, changed: false }); await unchanged.verifyStable(); await unchanged.verifyLocalStable(); await fs.writeFile(path.join(result.remoteWorkspaceDir, "modified.txt"), "late write\n"); @@ -652,7 +670,7 @@ describe("worker tunnel manager", () => { ); await fs.writeFile(path.join(localPath, "modified.txt"), "local late write\n"); await expect(unchanged.verifyLocalStable()).rejects.toThrow( - "Gateway workspace changed after cloud dispatch", + "Gateway workspace changed after cloud reconciliation", ); const manifestPath = path.join( diff --git a/src/gateway/worker-environments/workspace-accepted-sync.ts b/src/gateway/worker-environments/workspace-accepted-sync.ts new file mode 100644 index 000000000000..00c883d6b35d --- /dev/null +++ b/src/gateway/worker-environments/workspace-accepted-sync.ts @@ -0,0 +1,233 @@ +import { createHash, randomBytes } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import type { CommandOptions, SpawnResult } from "../../process/exec.js"; +import { workerSshCommandOptions } from "./ssh.js"; +import type { WorkerWorkspaceCommand } from "./tunnel-contract.js"; +import { + serializeWorkerWorkspaceManifest, + type WorkerWorkspaceManifest, +} from "./workspace-manifest.js"; +import { changedPaths, manifestNodes } from "./workspace-reconcile.js"; +import { + parseManifestRef, + workerWorkspaceCommandSucceeded, + workspaceSyncError, +} from "./workspace-sync-helpers.js"; +import { + REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS, + REMOTE_WORKSPACE_MANIFEST_JS, +} from "./workspace-sync-scripts.js"; + +const WORKSPACE_TIMEOUT_MS = 10 * 60_000; + +export async function recoverAcceptedWorkspacePublication(params: { + runWorkspaceCommand: (command: WorkerWorkspaceCommand) => Promise; + remoteWorkspaceDir: string; +}) { + const recovered = await params.runWorkspaceCommand({ + argv: [ + "node", + "-e", + REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS, + "recover", + params.remoteWorkspaceDir, + randomBytes(16).toString("hex"), + ], + }); + if (!workerWorkspaceCommandSucceeded(recovered)) { + throw workspaceSyncError(recovered); + } +} + +function createAcceptedWorkspacePublisher(params: { + runWorkspaceCommand: (command: WorkerWorkspaceCommand) => Promise; + runTask: (argv: string[], options: CommandOptions) => Promise; + ownerSignal: AbortSignal; + rsyncSsh: string; + scpTarget: string; + localPath: string; + remoteWorkspaceDir: string; + remoteManifest: WorkerWorkspaceManifest; +}) { + return async (accepted: { + manifestRef: string; + manifest: WorkerWorkspaceManifest; + conflictPaths: string[]; + }) => { + const acceptedRaw = serializeWorkerWorkspaceManifest(accepted.manifest); + const acceptedDigest = createHash("sha256").update(acceptedRaw).digest("hex"); + if (`sha256:${acceptedDigest}` !== accepted.manifestRef) { + throw new Error("Accepted workspace manifest does not match its reference"); + } + const published = await params.runWorkspaceCommand({ + argv: [ + "node", + "-e", + REMOTE_WORKSPACE_MANIFEST_JS, + params.remoteWorkspaceDir, + "", + "publish", + acceptedDigest, + ], + input: acceptedRaw, + }); + if (!workerWorkspaceCommandSucceeded(published)) { + throw workspaceSyncError(published); + } + + const verifyAcceptedWorkspace = async () => { + const verified = await params.runWorkspaceCommand({ + argv: [ + "node", + "-e", + REMOTE_WORKSPACE_MANIFEST_JS, + params.remoteWorkspaceDir, + accepted.manifest.baseCommit ?? "", + ...(accepted.manifest.baseCommit ? ["eligible", acceptedDigest] : []), + ], + }); + if (!workerWorkspaceCommandSucceeded(verified)) { + throw workspaceSyncError(verified); + } + const verifiedRef = parseManifestRef(verified.stdout.trim()); + if (verifiedRef !== accepted.manifestRef) { + throw new Error( + `Worker workspace does not match its accepted manifest: expected ${accepted.manifestRef}, got ${verifiedRef}`, + ); + } + }; + + // Git-ignored and derived worker scratch paths are intentionally outside the + // accepted manifest (for example dependency caches) and remain worker-local. + // Only accepted manifest members may be mirrored from the gateway. + const changed = changedPaths(params.remoteManifest, accepted.manifest); + if (changed.size === 0) { + await verifyAcceptedWorkspace(); + return; + } + + const transactionNonce = randomBytes(16).toString("hex"); + const transactionCommand = async (action: "apply" | "rollback" | "commit") => + await params.runWorkspaceCommand({ + argv: [ + "node", + "-e", + REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS, + action, + params.remoteWorkspaceDir, + transactionNonce, + ], + }); + let transactionBegun = false; + try { + const begun = await params.runWorkspaceCommand({ + argv: [ + "node", + "-e", + REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS, + "begin", + params.remoteWorkspaceDir, + transactionNonce, + ], + input: JSON.stringify([...changed]), + }); + if (!workerWorkspaceCommandSucceeded(begun)) { + throw workspaceSyncError(begun); + } + transactionBegun = true; + const remoteStagingRoot = begun.stdout.trim(); + if (!path.posix.isAbsolute(remoteStagingRoot) || remoteStagingRoot.includes("\n")) { + throw new Error("Worker returned an invalid accepted workspace staging path"); + } + + const acceptedNodes = manifestNodes(accepted.manifest); + const transferPaths = [...changed].filter((entryPath) => acceptedNodes.has(entryPath)); + if (transferPaths.length > 0) { + const temporaryDirectory = await fs.mkdtemp( + path.join(os.tmpdir(), "openclaw-worker-workspace-accepted-"), + ); + const transferListPath = path.join(temporaryDirectory, "transfer-list"); + try { + await fs.writeFile( + transferListPath, + Buffer.from(`${transferPaths.toSorted().join("\0")}\0`), + { mode: 0o600 }, + ); + const localSource = params.localPath.endsWith(path.sep) + ? params.localPath + : `${params.localPath}${path.sep}`; + const transferred = await params.runTask( + [ + "rsync", + "--archive", + "--checksum", + "--no-recursive", + "--from0", + `--files-from=${transferListPath}`, + "-e", + params.rsyncSsh, + "--", + localSource, + `${params.scpTarget}:${remoteStagingRoot}/`, + ], + workerSshCommandOptions({ + timeoutMs: WORKSPACE_TIMEOUT_MS, + signal: params.ownerSignal, + }), + ); + if (!workerWorkspaceCommandSucceeded(transferred)) { + throw workspaceSyncError(transferred); + } + } finally { + await fs.rm(temporaryDirectory, { recursive: true, force: true }); + } + } + + const applied = await transactionCommand("apply"); + if (!workerWorkspaceCommandSucceeded(applied)) { + throw workspaceSyncError(applied); + } + await verifyAcceptedWorkspace(); + const committed = await transactionCommand("commit"); + if (!workerWorkspaceCommandSucceeded(committed)) { + throw workspaceSyncError(committed); + } + } catch (error) { + if (transactionBegun) { + const rolledBack = await transactionCommand("rollback"); + if (!workerWorkspaceCommandSucceeded(rolledBack)) { + const rollbackError = new Error("Accepted workspace publication rollback failed", { + cause: error, + }); + Object.defineProperty(rollbackError, "rollbackFailure", { + value: workspaceSyncError(rolledBack), + }); + throw rollbackError; + } + } + throw error; + } + }; +} + +export function createAcceptedWorkspacePublisherFactory( + params: Omit[0], "remoteManifest">, +) { + return (remoteManifest: WorkerWorkspaceManifest, initialRemoteRef: string) => { + let expectedRemoteRef = initialRemoteRef; + const publish = createAcceptedWorkspacePublisher({ ...params, remoteManifest }); + return { + expectedRemoteRef: () => expectedRemoteRef, + publishAcceptedManifest: async (accepted: { + manifestRef: string; + manifest: WorkerWorkspaceManifest; + conflictPaths: string[]; + }) => { + await publish(accepted); + expectedRemoteRef = accepted.manifestRef; + }, + }; + }; +} diff --git a/src/gateway/worker-environments/workspace-manifest-remote-script.ts b/src/gateway/worker-environments/workspace-manifest-remote-script.ts new file mode 100644 index 000000000000..f9d00dec8897 --- /dev/null +++ b/src/gateway/worker-environments/workspace-manifest-remote-script.ts @@ -0,0 +1,515 @@ +export const REMOTE_WORKSPACE_MANIFEST_CANONICAL_JS = String.raw`function canonicalMode(type, mode) { + if (type === "directory") return 0o700; + if (type === "symlink") return 0o777; + return (mode & 0o111) === 0 ? 0o644 : 0o755; +} +function canonicalEntry(entry) { + if (entry.type === "directory") { + return { path: entry.path, type: entry.type, mode: canonicalMode(entry.type, entry.mode) }; + } + if (entry.type === "file") { + return { + path: entry.path, + type: entry.type, + mode: canonicalMode(entry.type, entry.mode), + size: entry.size, + sha256: entry.sha256, + }; + } + if (entry.type === "symlink") { + return { + path: entry.path, + type: entry.type, + mode: canonicalMode(entry.type, entry.mode), + target: entry.target, + }; + } + fail("unsupported worker workspace manifest entry"); +} +function compareManifestPaths(left, right) { + return left.path < right.path ? -1 : left.path > right.path ? 1 : 0; +} +function serializeManifest(baseCommit, entries, comparePaths = compareManifestPaths) { + return JSON.stringify({ + version: 1, + baseCommit, + entries: entries + .filter((entry) => !isDerivedWorkspacePath(entry.path)) + .map(canonicalEntry) + .sort(comparePaths), + }); +}`; + +export const REMOTE_WORKSPACE_MANIFEST_REGISTRY_JS = String.raw`function publishManifest(manifestRoot, manifest) { + const digest = crypto.createHash("sha256").update(manifest).digest("hex"); + const manifestPath = path.join(manifestRoot, digest + ".json"); + const temporaryPath = manifestPath + "." + process.pid + "." + crypto.randomBytes(4).toString("hex"); + fs.writeFileSync(temporaryPath, manifest, { encoding: "utf8", flag: "wx", mode: 0o600 }); + try { + try { + fs.linkSync(temporaryPath, manifestPath); + } catch (error) { + const existing = error && error.code === "EEXIST" ? fs.lstatSync(manifestPath) : null; + if ( + !existing || + existing.isSymbolicLink() || + !existing.isFile() || + fs.readFileSync(manifestPath, "utf8") !== manifest + ) { + throw error; + } + } + } finally { + fs.rmSync(temporaryPath, { force: true }); + } + return digest; +} +function readManifestFile(manifestPath) { + const descriptor = fs.openSync(manifestPath, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW); + try { + const stats = fs.fstatSync(descriptor); + if (!stats.isFile() || stats.size > 64 * 1024 * 1024) { + fail("unsafe worker workspace manifest file"); + } + return fs.readFileSync(descriptor, "utf8"); + } finally { + fs.closeSync(descriptor); + } +} +function resolveManifest(manifestRoot, requestedDigest) { + if (!/^[a-f0-9]{64}$/.test(requestedDigest || "")) fail("invalid workspace manifest digest"); + const requestedPath = path.join(manifestRoot, requestedDigest + ".json"); + try { + fs.lstatSync(requestedPath); + // The bounded inbound transfer remains authoritative for validating an + // already-addressable manifest's type, size, and content digest. + return requestedDigest; + } catch (error) { + if (!error || error.code !== "ENOENT") throw error; + } + + // Drain only the immediately preceding unshipped gateway format. The caller + // supplies that same profile's full locale; this is not a shipped migration. + let legacyCompare; + try { + legacyCompare = new Intl.Collator(legacyGatewayLocale).compare; + } catch { + fail("invalid legacy gateway locale"); + } + const candidates = fs + .readdirSync(manifestRoot) + .filter((name) => /^[a-f0-9]{64}\.json$/.test(name)) + .map((name) => { + try { + return { name, mtimeMs: fs.lstatSync(path.join(manifestRoot, name)).mtimeMs }; + } catch { + return null; + } + }) + .filter(Boolean) + .sort((left, right) => + right.mtimeMs - left.mtimeMs || (left.name < right.name ? -1 : left.name > right.name ? 1 : 0), + ) + .slice(0, 256); + let scannedBytes = 0; + for (const { name } of candidates) { + const candidatePath = path.join(manifestRoot, name); + let raw; + try { + raw = readManifestFile(candidatePath); + } catch { + continue; + } + scannedBytes += Buffer.byteLength(raw); + if (scannedBytes > 256 * 1024 * 1024) break; + if (crypto.createHash("sha256").update(raw).digest("hex") !== name.slice(0, -5)) continue; + let value; + try { + value = JSON.parse(raw); + } catch { + continue; + } + if (!value || value.version !== 1 || !Array.isArray(value.entries)) continue; + let canonical; + try { + canonical = serializeManifest(value.baseCommit ?? null, value.entries); + } catch { + continue; + } + if (crypto.createHash("sha256").update(canonical).digest("hex") !== requestedDigest) { + // Old gateways used their default locale collation for the accepted ref. + const legacySeed = [ + ...value.entries.filter((entry) => entry.type === "directory"), + ...value.entries.filter((entry) => entry.type !== "directory"), + ]; + canonical = serializeManifest(value.baseCommit ?? null, legacySeed, (left, right) => + legacyCompare(left.path, right.path), + ); + if (crypto.createHash("sha256").update(canonical).digest("hex") !== requestedDigest) continue; + } + if (publishManifest(manifestRoot, canonical) !== requestedDigest) { + fail("resolved workspace manifest digest mismatch"); + } + return requestedDigest; + } + fail("worker workspace manifest is unavailable: " + requestedDigest); +}`; + +export const REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS = String.raw`const crypto = require("node:crypto"); +const fs = require("node:fs"); +const path = require("node:path"); +const action = process.argv[1]; +const root = fs.realpathSync(process.argv[2]); +const nonce = process.argv[3]; +if (!/^[a-f0-9]{32}$/.test(nonce || "")) throw new Error("invalid accepted workspace transaction"); +// REMOTE_WORKSPACE_SETUP_SCRIPT creates and chmods every workspace parent for this worker. +// Keeping the transaction beside the workspace makes all live swaps same-filesystem renames. +const transactionRoot = path.dirname(root); +const transactionRootStats = fs.lstatSync(transactionRoot); +if (transactionRootStats.isSymbolicLink() || !transactionRootStats.isDirectory()) { + throw new Error("unsafe accepted workspace transaction directory"); +} +const workspaceKey = crypto.createHash("sha256").update(root).digest("hex"); +const transactionPrefix = ".openclaw-accepted-" + workspaceKey + "-"; +const cleanupPrefix = ".openclaw-accepted-cleanup-" + workspaceKey + "-"; +const transaction = path.join(transactionRoot, transactionPrefix + nonce); +const cleanup = path.join(transactionRoot, cleanupPrefix + nonce); +const nextRoot = path.join(transaction, "next"); +const backupRoot = path.join(transaction, "backup"); +const pathsFile = path.join(transaction, "paths.json"); +const stateFile = path.join(transaction, "state.json"); +const ancestorModesFile = path.join(transaction, "ancestor-modes.json"); +const appliedFile = path.join(transaction, "applied"); +function isSafeRelativePath(relative) { + return ( + typeof relative === "string" && + relative && + !relative.includes("\\") && + !path.posix.isAbsolute(relative) && + path.posix.normalize(relative) === relative && + relative !== "." && + relative !== ".." && + relative !== ".git" && + !relative.startsWith(".git/") && + !relative.startsWith("../") + ); +} +function parsePaths(raw) { + const values = JSON.parse(raw); + if (!Array.isArray(values) || values.length > 25_000) { + throw new Error("invalid accepted workspace paths"); + } + const paths = [...new Set(values)]; + for (const relative of paths) { + if (!isSafeRelativePath(relative)) { + throw new Error("unsafe accepted workspace path"); + } + } + const selected = new Set(paths); + // Directory modes are canonical, so a changed directory is added, removed, or + // replaced and all of its accepted descendants are changed and staged too. + return paths + .filter((relative) => { + const segments = relative.split("/"); + for (let index = 1; index < segments.length; index += 1) { + if (selected.has(segments.slice(0, index).join("/"))) return false; + } + return true; + }) + .sort(); +} +function targetPath(base, relative) { + return path.join(base, relative); +} +function livePath(relative) { + const segments = relative.split("/"); + let parent = root; + for (const segment of segments.slice(0, -1)) { + parent = path.join(parent, segment); + const stats = fs.lstatSync(parent); + if (stats.isSymbolicLink() || !stats.isDirectory()) { + throw new Error("unsafe accepted workspace parent"); + } + } + return path.join(root, relative); +} +function exists(target) { + try { + fs.lstatSync(target); + return true; + } catch (error) { + if (error && error.code === "ENOENT") return false; + throw error; + } +} +function removeTree(target) { + let stats; + try { + stats = fs.lstatSync(target); + } catch (error) { + if (error && error.code === "ENOENT") return; + throw error; + } + if (stats.isDirectory() && !stats.isSymbolicLink()) { + fs.chmodSync(target, 0o700); + for (const name of fs.readdirSync(target)) { + removeTree(path.join(target, name)); + } + fs.rmdirSync(target); + } else { + fs.unlinkSync(target); + } +} +function readPaths() { + return parsePaths(fs.readFileSync(pathsFile, "utf8")); +} +function readState(candidate) { + const value = JSON.parse(fs.readFileSync(path.join(candidate, "state.json"), "utf8")); + if (!Array.isArray(value) || value.length > 25_000) { + throw new Error("invalid accepted workspace transaction state"); + } + const relatives = parsePaths(JSON.stringify(value.map((entry) => entry && entry.relative))); + if ( + relatives.length !== value.length || + value.some( + (entry, index) => + !entry || + entry.relative !== relatives[index] || + typeof entry.hadLive !== "boolean" || + (entry.directoryMode !== undefined && + (!Number.isInteger(entry.directoryMode) || + entry.directoryMode < 0 || + entry.directoryMode > 0o7777)), + ) + ) { + throw new Error("invalid accepted workspace transaction state"); + } + return value; +} +function readAncestorModes(candidate) { + const candidateModes = path.join(candidate, "ancestor-modes.json"); + if (!exists(candidateModes)) return []; + const value = JSON.parse(fs.readFileSync(candidateModes, "utf8")); + if (!Array.isArray(value) || value.length > 250_000) { + throw new Error("invalid accepted workspace ancestor modes"); + } + const seen = new Set(); + for (const entry of value) { + if ( + !entry || + (entry.relative !== "" && !isSafeRelativePath(entry.relative)) || + seen.has(entry.relative) || + !Number.isInteger(entry.mode) || + entry.mode < 0 || + entry.mode > 0o7777 + ) { + throw new Error("invalid accepted workspace ancestor modes"); + } + seen.add(entry.relative); + } + return value; +} +function writeAncestorModes(value) { + const temporary = ancestorModesFile + ".tmp"; + fs.writeFileSync(temporary, JSON.stringify(value), { flag: "wx", mode: 0o600 }); + fs.renameSync(temporary, ancestorModesFile); +} +function ancestorPaths(paths) { + const ancestors = new Set(); + for (const relative of paths) { + const segments = relative.split("/"); + for (let index = 1; index < segments.length; index += 1) { + ancestors.add(segments.slice(0, index).join("/")); + } + } + if (ancestors.size + 1 > 250_000) { + throw new Error("accepted workspace transaction has too many ancestors"); + } + return [...ancestors].sort((left, right) => { + const depth = left.split("/").length - right.split("/").length; + return depth || (left < right ? -1 : left > right ? 1 : 0); + }); +} +function prepareWritableAncestors(paths) { + // parsePaths removes descendants of changed directories, so these are all + // unchanged live ancestors. Read every mode before mutating any permission. + const modes = ["", ...ancestorPaths(paths)].map((relative) => { + const target = relative ? targetPath(root, relative) : root; + const stats = fs.lstatSync(target); + if (stats.isSymbolicLink() || !stats.isDirectory()) { + throw new Error("unsafe accepted workspace parent"); + } + return { relative, mode: stats.mode & 0o7777 }; + }); + writeAncestorModes(modes); + makeAncestorsWritable(modes); + return modes; +} +function makeAncestorsWritable(modes) { + const widened = []; + try { + for (const entry of modes) { + const target = entry.relative ? targetPath(root, entry.relative) : root; + const stats = fs.lstatSync(target); + if (stats.isSymbolicLink() || !stats.isDirectory()) { + throw new Error("unsafe accepted workspace parent"); + } + const currentMode = stats.mode & 0o7777; + const writableMode = entry.mode | 0o700; + if (currentMode !== writableMode) { + fs.chmodSync(target, writableMode); + widened.push(entry); + } + } + } catch (error) { + try { + restoreAncestorModes(widened); + } catch (restoreError) { + const failure = new Error("accepted workspace ancestor mode rollback failed", { + cause: error, + }); + Object.defineProperty(failure, "restoreFailure", { value: restoreError }); + throw failure; + } + throw error; + } +} +function restoreAncestorModes(modes) { + for (const entry of [...modes].reverse()) { + const target = entry.relative ? targetPath(root, entry.relative) : root; + const stats = fs.lstatSync(target); + if (stats.isSymbolicLink() || !stats.isDirectory()) { + throw new Error("unsafe accepted workspace parent"); + } + if ((stats.mode & 0o7777) !== entry.mode) { + fs.chmodSync(target, entry.mode); + } + } +} +function removeTransaction(candidate = transaction) { + removeTree(candidate); +} +function restoreTransaction(candidate) { + if (!exists(candidate)) return; + const ancestorModes = readAncestorModes(candidate); + makeAncestorsWritable(ancestorModes); + const candidateState = path.join(candidate, "state.json"); + try { + if (exists(candidateState)) { + const candidateBackup = path.join(candidate, "backup"); + for (const entry of [...readState(candidate)].reverse()) { + const live = livePath(entry.relative); + const backup = targetPath(candidateBackup, entry.relative); + if (exists(backup)) { + removeTree(live); + fs.renameSync(backup, live); + if (entry.directoryMode !== undefined) { + fs.chmodSync(live, entry.directoryMode); + } + } else if (!entry.hadLive) { + removeTree(live); + } else if (entry.directoryMode !== undefined && exists(live)) { + fs.chmodSync(live, entry.directoryMode); + } + } + } + } finally { + restoreAncestorModes(ancestorModes); + } + removeTransaction(candidate); +} +function recoverTransaction(candidate) { + restoreTransaction(candidate); +} +function recoverTransactions() { + for (const name of fs.readdirSync(transactionRoot)) { + if ( + name.startsWith(cleanupPrefix) && + /^[a-f0-9]{32}$/.test(name.slice(cleanupPrefix.length)) + ) { + removeTransaction(path.join(transactionRoot, name)); + } + } + for (const name of fs.readdirSync(transactionRoot)) { + if ( + name.startsWith(transactionPrefix) && + /^[a-f0-9]{32}$/.test(name.slice(transactionPrefix.length)) + ) { + recoverTransaction(path.join(transactionRoot, name)); + } + } +} +if (action === "begin") { + const paths = parsePaths(fs.readFileSync(0, "utf8")); + recoverTransactions(); + fs.mkdirSync(transaction, { mode: 0o700 }); + fs.mkdirSync(nextRoot, { mode: 0o700 }); + fs.mkdirSync(backupRoot, { mode: 0o700 }); + fs.writeFileSync(pathsFile, JSON.stringify(paths), { mode: 0o600 }); + process.stdout.write(nextRoot + "\n"); +} else if (action === "apply") { + const paths = readPaths(); + try { + const ancestorModes = prepareWritableAncestors(paths); + const state = paths.map((relative) => { + const live = livePath(relative); + if (!exists(live)) return { relative, hadLive: false }; + const stats = fs.lstatSync(live); + return { + relative, + hadLive: true, + ...(stats.isDirectory() && !stats.isSymbolicLink() + ? { directoryMode: stats.mode & 0o7777 } + : {}), + }; + }); + const temporaryStateFile = stateFile + ".tmp"; + fs.writeFileSync(temporaryStateFile, JSON.stringify(state), { flag: "wx", mode: 0o600 }); + fs.renameSync(temporaryStateFile, stateFile); + for (const entry of state) { + if (!entry.hadLive) continue; + const source = livePath(entry.relative); + const sourceStats = fs.lstatSync(source); + const destination = targetPath(backupRoot, entry.relative); + fs.mkdirSync(path.dirname(destination), { recursive: true, mode: 0o700 }); + try { + if (sourceStats.isDirectory() && !sourceStats.isSymbolicLink()) { + fs.chmodSync(source, 0o700); + } + fs.renameSync(source, destination); + } catch (error) { + if (entry.directoryMode !== undefined && exists(source)) { + fs.chmodSync(source, entry.directoryMode); + } + throw error; + } + } + for (const entry of state) { + const source = targetPath(nextRoot, entry.relative); + if (!exists(source)) continue; + fs.renameSync(source, livePath(entry.relative)); + } + restoreAncestorModes(ancestorModes); + fs.writeFileSync(appliedFile, "", { flag: "wx", mode: 0o600 }); + } catch (error) { + restoreTransaction(transaction); + throw error; + } +} else if (action === "rollback") { + if (exists(cleanup)) { + if (exists(transaction)) throw new Error("ambiguous accepted workspace transaction state"); + fs.renameSync(cleanup, transaction); + } + restoreTransaction(transaction); +} else if (action === "recover") { + recoverTransactions(); +} else if (action === "commit") { + if (exists(transaction)) { + if (!exists(appliedFile)) throw new Error("accepted workspace transaction is not applied"); + // The namespace rename is the commit point. Later recovery removes the backup + // only after the gateway has had a chance to observe this command's success. + fs.renameSync(transaction, cleanup); + } +} else { + throw new Error("invalid accepted workspace transaction action"); +}`; diff --git a/src/gateway/worker-environments/workspace-manifest.ts b/src/gateway/worker-environments/workspace-manifest.ts index 66f0e21c9cdd..418d2fe7fe12 100644 --- a/src/gateway/worker-environments/workspace-manifest.ts +++ b/src/gateway/worker-environments/workspace-manifest.ts @@ -70,6 +70,10 @@ export function gitFileMode(mode: number): number { return (mode & 0o111) === 0 ? 0o644 : 0o755; } +function compareManifestPaths(left: { path: string }, right: { path: string }): number { + return left.path < right.path ? -1 : left.path > right.path ? 1 : 0; +} + type RawManifestEntry = | { path: string; type: "directory"; mode: number } | WorkerWorkspaceManifestEntry; @@ -173,7 +177,7 @@ export function serializeWorkerWorkspaceManifest(manifest: WorkerWorkspaceManife mode: 0o700, })), ...manifest.entries.filter((entry) => !isDerivedWorkspacePath(entry.path)), - ].toSorted((left, right) => left.path.localeCompare(right.path)), + ].toSorted(compareManifestPaths), }); } diff --git a/src/gateway/worker-environments/workspace-reconcile-apply.ts b/src/gateway/worker-environments/workspace-reconcile-apply.ts index 177b3b694b44..9d1770f975be 100644 --- a/src/gateway/worker-environments/workspace-reconcile-apply.ts +++ b/src/gateway/worker-environments/workspace-reconcile-apply.ts @@ -39,9 +39,19 @@ export async function applyStagedWorkerWorkspace(params: { base: WorkerWorkspaceManifest; current: WorkerWorkspaceManifest; journal: WorkerWorkspaceReconciliationJournalAdapter; + publishAcceptedManifest?: (accepted: { + manifestRef: string; + manifest: WorkerWorkspaceManifest; + conflictPaths: string[]; + }) => Promise; }): Promise { const root = await fs.realpath(params.root); const preserveDirectories = new Set(reconciliationDirectories(params.current.directories)); + // Git workspaces must keep the eligibility boundary established at dispatch. + // Local-only ignored files are outside both manifests and must never enter the accepted state. + const includePaths = params.current.baseCommit + ? new Set([...manifestNodes(params.base).keys(), ...manifestNodes(params.current).keys()]) + : undefined; const preflight = await preflightWorkspaceApply({ root, base: params.base, @@ -53,6 +63,7 @@ export async function applyStagedWorkerWorkspace(params: { root, baseCommit: params.current.baseCommit, preserveDirectories, + includePaths, }); const finalPreflight = await preflightWorkspaceApply({ root, @@ -64,17 +75,21 @@ export async function applyStagedWorkerWorkspace(params: { expectedRef: actual.manifestRef, baseCommit: actual.manifest.baseCommit, preserveDirectories, + includePaths, }); + const conflictPaths = retainedConflictPaths(finalPreflight, preflight.applyPaths); + await params.publishAcceptedManifest?.({ ...actual, conflictPaths }); params.journal.commit(actual.manifestRef); return { ...actual, - conflictPaths: retainedConflictPaths(finalPreflight, preflight.applyPaths), + conflictPaths, verifyLocalStable: async () => await assertActualWorkspaceManifest({ root, expectedRef: actual.manifestRef, baseCommit: actual.manifest.baseCommit, preserveDirectories, + includePaths, }), }; } @@ -169,6 +184,7 @@ export async function applyStagedWorkerWorkspace(params: { root, baseCommit: params.current.baseCommit, preserveDirectories, + includePaths, }); const finalPreflight = await preflightWorkspaceApply({ root, @@ -180,17 +196,21 @@ export async function applyStagedWorkerWorkspace(params: { expectedRef: actual.manifestRef, baseCommit: actual.manifest.baseCommit, preserveDirectories, + includePaths, }); + const conflictPaths = retainedConflictPaths(finalPreflight, preflight.applyPaths); + await params.publishAcceptedManifest?.({ ...actual, conflictPaths }); params.journal.commit(actual.manifestRef); return { ...actual, - conflictPaths: retainedConflictPaths(finalPreflight, preflight.applyPaths), + conflictPaths, verifyLocalStable: async () => await assertActualWorkspaceManifest({ root, expectedRef: actual.manifestRef, baseCommit: actual.manifest.baseCommit, preserveDirectories, + includePaths, }), }; } catch (error) { diff --git a/src/gateway/worker-environments/workspace-reconcile-core.ts b/src/gateway/worker-environments/workspace-reconcile-core.ts index 91c9da7f2ef1..7e2fc441b3de 100644 --- a/src/gateway/worker-environments/workspace-reconcile-core.ts +++ b/src/gateway/worker-environments/workspace-reconcile-core.ts @@ -189,6 +189,7 @@ export async function readActualWorkspaceManifest(params: { root: string; baseCommit: string | null; preserveDirectories?: ReadonlySet; + includePaths?: ReadonlySet; }): Promise<{ manifest: WorkerWorkspaceManifest; manifestRef: string }> { const rawEntries: Array< WorkerWorkspaceManifestEntry | { path: string; type: "directory"; mode: number } @@ -232,6 +233,9 @@ export async function readActualWorkspaceManifest(params: { hasDerivedEntry = true; continue; } + if (params.includePaths && !params.includePaths.has(relative)) { + continue; + } const absolute = localPath(params.root, relative); const stats = await fs.lstat(absolute); if (stats.isDirectory() && !stats.isSymbolicLink()) { @@ -355,6 +359,7 @@ export async function assertActualWorkspaceManifest(params: { expectedRef: string; baseCommit: string | null; preserveDirectories?: ReadonlySet; + includePaths?: ReadonlySet; }): Promise { const actual = await readActualWorkspaceManifest(params); if (actual.manifestRef !== params.expectedRef) { diff --git a/src/gateway/worker-environments/workspace-reconcile.test.ts b/src/gateway/worker-environments/workspace-reconcile.test.ts index 3bf71b589e10..1eb83b82f3d7 100644 --- a/src/gateway/worker-environments/workspace-reconcile.test.ts +++ b/src/gateway/worker-environments/workspace-reconcile.test.ts @@ -99,12 +99,18 @@ async function applyWorkspace(params: { begin?: (journal: WorkerWorkspaceReconciliationJournal) => void; commit?: (manifestRef: string) => void; abort?: () => void; + publishAcceptedManifest?: (accepted: { + manifestRef: string; + manifest: WorkerWorkspaceManifest; + conflictPaths: string[]; + }) => Promise; }) { let pending: WorkerWorkspaceReconciliationJournal | undefined; return await applyStagedWorkerWorkspace({ ...params, baseManifestRef: `sha256:${"a".repeat(64)}`, currentManifestRef: `sha256:${"b".repeat(64)}`, + publishAcceptedManifest: params.publishAcceptedManifest, journal: { load: () => pending, begin: (journal) => { @@ -124,6 +130,55 @@ async function applyWorkspace(params: { } describe("worker workspace reconciliation", () => { + it("keeps local-only paths outside a Git workspace's accepted manifest", async () => { + const local = await temporaryDirectory("workspace-accepted-git-eligibility"); + const staging = await temporaryDirectory("workspace-accepted-git-eligibility-staging"); + await fs.writeFile(path.join(local, "tracked.txt"), "tracked\n"); + const manifest = { + ...(await manifestFor(local)), + baseCommit: "a".repeat(40), + }; + await fs.writeFile(path.join(local, "local-ignored-secret.txt"), "private\n"); + let published: WorkerWorkspaceManifest | undefined; + + const applied = await applyWorkspace({ + root: local, + stagingRoot: staging, + base: manifest, + current: manifest, + publishAcceptedManifest: async (accepted) => { + published = accepted.manifest; + }, + }); + + expect(published?.entries.map((entry) => entry.path)).toEqual(["tracked.txt"]); + expect(applied.manifest.entries.map((entry) => entry.path)).toEqual(["tracked.txt"]); + await expect(fs.readFile(path.join(local, "local-ignored-secret.txt"), "utf8")).resolves.toBe( + "private\n", + ); + }); + + it("publishes the accepted manifest before advancing the journal", async () => { + const local = await temporaryDirectory("workspace-publish-before-commit"); + const staging = await temporaryDirectory("workspace-publish-before-commit-staging"); + await fs.writeFile(path.join(local, "result.txt"), "accepted\n"); + const manifest = await manifestFor(local); + const events: string[] = []; + + const applied = await applyWorkspace({ + root: local, + stagingRoot: staging, + base: manifest, + current: manifest, + publishAcceptedManifest: async (accepted) => { + events.push(`publish:${accepted.manifestRef}`); + }, + commit: (manifestRef) => events.push(`commit:${manifestRef}`), + }); + + expect(events).toEqual([`publish:${applied.manifestRef}`, `commit:${applied.manifestRef}`]); + }); + it("rejects unsafe claim ids before constructing a staged result ref", () => { expect(workerWorkspaceResultRef("6f77e833-83d2-4db4-bdd4-2ad1d37edc28")).toBe( "refs/openclaw/worker-results/6f77e833-83d2-4db4-bdd4-2ad1d37edc28", diff --git a/src/gateway/worker-environments/workspace-result-staging.ts b/src/gateway/worker-environments/workspace-result-staging.ts index 345ce12fb01e..9cc0cc55778e 100644 --- a/src/gateway/worker-environments/workspace-result-staging.ts +++ b/src/gateway/worker-environments/workspace-result-staging.ts @@ -456,6 +456,11 @@ export async function applyStagedWorkerWorkspaceResult(params: { expectedBaseManifestRef: string; alreadyAccepted?: boolean; journal: WorkerWorkspaceReconciliationJournalAdapter; + publishAcceptedManifest?: (accepted: { + manifestRef: string; + manifest: WorkerWorkspaceManifest; + conflictPaths: string[]; + }) => Promise; }): Promise { const root = await fs.realpath(params.root); const staged = await loadStagedWorkerWorkspace(root, params.stagedResultRef); @@ -509,6 +514,7 @@ export async function applyStagedWorkerWorkspaceResult(params: { base: staged.base, current: staged.current, journal: params.journal, + publishAcceptedManifest: params.publishAcceptedManifest, }); return { ...applied, changed: changed.size > 0 }; } finally { @@ -522,6 +528,11 @@ async function prepareRequestedWorkerWorkspaceResult(params: { currentManifestRef: string; baseManifestRaw: string; currentManifestRaw: string; + publishAcceptedManifest?: (accepted: { + manifestRef: string; + manifest: WorkerWorkspaceManifest; + conflictPaths: string[]; + }) => Promise; }): Promise<{ applyPreparedStagedResult(): Promise; getAppliedWorkspaceResult(): WorkerWorkspaceApplyResult | undefined; @@ -552,6 +563,7 @@ async function prepareRequestedWorkerWorkspaceResult(params: { stagedResultRef: candidateRef, expectedBaseManifestRef: params.request.baseManifestRef, journal: params.request.journal, + publishAcceptedManifest: params.publishAcceptedManifest, }); }, getAppliedWorkspaceResult: () => appliedWorkspaceResult, diff --git a/src/gateway/worker-environments/workspace-sync-helpers.ts b/src/gateway/worker-environments/workspace-sync-helpers.ts index a3305cb3a3e7..1b3c41d91215 100644 --- a/src/gateway/worker-environments/workspace-sync-helpers.ts +++ b/src/gateway/worker-environments/workspace-sync-helpers.ts @@ -4,10 +4,16 @@ import path from "node:path"; import { setTimeout as delay } from "node:timers/promises"; import { redactSensitiveText } from "../../logging/redact.js"; import type { CommandOptions, SpawnResult } from "../../process/exec.js"; -import { type PreparedWorkerSsh, workerSshCommandOptions } from "./ssh.js"; -import type { WorkerWorkspaceSyncRequest } from "./tunnel-contract.js"; +import { + type PreparedWorkerSsh, + workerSshCommandOptions, + workerSshOptions, + workerSshRemoteCommand, +} from "./ssh.js"; +import type { WorkerWorkspaceCommand, WorkerWorkspaceSyncRequest } from "./tunnel-contract.js"; +import { REMOTE_WORKSPACE_MANIFEST_JS } from "./workspace-sync-scripts.js"; -export const MANIFEST_REF_PATTERN = /^sha256:[a-f0-9]{64}$/u; +const MANIFEST_REF_PATTERN = /^sha256:[a-f0-9]{64}$/u; export type WorkerWorkspaceActionsOptions = { environmentId: string; @@ -51,6 +57,106 @@ export function workspaceSyncError(result: SpawnResult): Error { ); } +export function workerWorkspaceRsyncRemoteCommand(prepared: PreparedWorkerSsh): string { + return workerSshRemoteCommand([ + "ssh", + ...workerSshOptions(prepared, { forwarding: "disabled" }), + "-a", + "-x", + "-T", + "-p", + String(prepared.port), + ]); +} + +export function workerWorkspaceSshArgv( + prepared: PreparedWorkerSsh, + remoteArgv: readonly string[], +): string[] { + return [ + "ssh", + ...workerSshOptions(prepared, { forwarding: "disabled" }), + "-a", + "-x", + "-T", + "-p", + String(prepared.port), + "--", + prepared.sshTarget, + workerSshRemoteCommand(remoteArgv), + ]; +} + +async function resolveRemoteWorkspaceBaseManifest( + runWorkspaceCommand: (command: WorkerWorkspaceCommand) => Promise, + remoteWorkspaceDir: string, + expectedRef: string, +): Promise { + const baseDigest = MANIFEST_REF_PATTERN.test(expectedRef) ? expectedRef.slice(7) : ""; + if (!baseDigest) { + throw new Error("Worker workspace base manifest reference is invalid"); + } + const resolved = await runWorkspaceCommand({ + argv: [ + "node", + "-e", + REMOTE_WORKSPACE_MANIFEST_JS, + remoteWorkspaceDir, + "", + "resolve", + baseDigest, + Intl.DateTimeFormat().resolvedOptions().locale, + ], + }); + if (!workerWorkspaceCommandSucceeded(resolved)) { + throw workspaceSyncError(resolved); + } + if (parseManifestRef(resolved.stdout.trim()) !== expectedRef) { + throw new Error("Worker workspace base manifest resolution returned the wrong reference"); + } + return baseDigest; +} + +export async function resolveRemoteWorkspaceManifest( + runWorkspaceCommand: (command: WorkerWorkspaceCommand) => Promise, + remoteWorkspaceDir: string, + expectedRef: string, +) { + return await resolveRemoteWorkspaceBaseManifest( + runWorkspaceCommand, + remoteWorkspaceDir, + expectedRef, + ); +} + +export async function verifyRemoteWorkspaceManifest(params: { + runWorkspaceCommand: (command: WorkerWorkspaceCommand) => Promise; + remoteWorkspaceDir: string; + baseCommit: string | null; + baseDigest: string; + expectedRef: string; +}): Promise { + const expectedDigest = params.expectedRef.slice("sha256:".length); + const verified = await params.runWorkspaceCommand({ + argv: [ + "node", + "-e", + REMOTE_WORKSPACE_MANIFEST_JS, + params.remoteWorkspaceDir, + params.baseCommit ?? "", + // Seed both manifests so a deleted path recreated under a new ignore rule + // still invalidates the fence. + ...(params.baseCommit ? ["eligible", expectedDigest, params.baseDigest] : []), + ], + }); + if (!workerWorkspaceCommandSucceeded(verified)) { + throw workspaceSyncError(verified); + } + if (parseManifestRef(verified.stdout.trim()) !== params.expectedRef) { + throw new Error("Cloud workspace changed during final reconciliation"); + } +} + export async function probeWorkspaceGitMode(params: { localPath: string; commandOptions: CommandOptions; diff --git a/src/gateway/worker-environments/workspace-sync-scripts.test.ts b/src/gateway/worker-environments/workspace-sync-scripts.test.ts index 55468a221c48..0ecdc8c9092d 100644 --- a/src/gateway/worker-environments/workspace-sync-scripts.test.ts +++ b/src/gateway/worker-environments/workspace-sync-scripts.test.ts @@ -5,6 +5,11 @@ import path from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import { runCommandWithTimeout } from "../../process/exec.js"; import { + parseWorkerWorkspaceManifest, + serializeWorkerWorkspaceManifest, +} from "./workspace-manifest.js"; +import { + REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS, REMOTE_WORKSPACE_MANIFEST_JS, REMOTE_WORKSPACE_QUIESCE_JS, REMOTE_WORKSPACE_RENEW_QUIESCENCE_JS, @@ -185,6 +190,312 @@ describe("remote workspace quiescence scripts", () => { }); describe("remote workspace manifest script", () => { + it("atomically applies and rolls back accepted workspace paths", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-accepted-paths-test-")); + roots.push(root); + const home = path.join(root, "home"); + const workspace = path.join(root, "workspace"); + await Promise.all([fs.mkdir(home), fs.mkdir(workspace)]); + await fs.writeFile(path.join(workspace, "node"), "old file\n"); + const env = { ...process.env, HOME: home }; + const runTransaction = async (action: string, nonce: string, input?: string) => + await runCommandWithTimeout( + [ + process.execPath, + "-e", + REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS, + action, + workspace, + nonce, + ], + { timeoutMs: 10_000, baseEnv: env, input }, + ); + for (const unsafePath of [".", ".."]) { + const rejected = await runTransaction("begin", "f".repeat(32), JSON.stringify([unsafePath])); + expect(rejected.code).not.toBe(0); + await expect(fs.access(workspace)).resolves.toBeUndefined(); + } + + const nonce = "a".repeat(32); + const begun = await runTransaction( + "begin", + nonce, + JSON.stringify(["node/child.txt", "node", "added.txt"]), + ); + expect(begun.code).toBe(0); + const staging = begun.stdout.trim(); + await fs.mkdir(path.join(staging, "node")); + await Promise.all([ + fs.writeFile(path.join(staging, "node/child.txt"), "new child\n"), + fs.writeFile(path.join(staging, "added.txt"), "added\n"), + ]); + expect((await runTransaction("apply", nonce)).code).toBe(0); + await expect(fs.readFile(path.join(workspace, "node/child.txt"), "utf8")).resolves.toBe( + "new child\n", + ); + await expect(fs.readFile(path.join(workspace, "added.txt"), "utf8")).resolves.toBe("added\n"); + + await fs.rm(path.join(path.dirname(staging), "applied")); + const recoveryNonce = "b".repeat(32); + const recoveryBegin = await runTransaction("begin", recoveryNonce, JSON.stringify(["node"])); + expect(recoveryBegin.code).toBe(0); + await expect(fs.readFile(path.join(workspace, "node"), "utf8")).resolves.toBe("old file\n"); + await expect(fs.access(path.join(workspace, "added.txt"))).rejects.toThrow(); + expect((await runTransaction("rollback", recoveryNonce)).code).toBe(0); + await fs.rm(path.join(workspace, "node")); + await fs.mkdir(path.join(workspace, "node")); + await fs.writeFile(path.join(workspace, "node/old.txt"), "read only\n"); + await fs.chmod(path.join(workspace, "node"), 0o555); + + const committedNonce = "c".repeat(32); + const committedBegin = await runTransaction( + "begin", + committedNonce, + JSON.stringify(["node/child.txt", "node"]), + ); + const committedStaging = committedBegin.stdout.trim(); + await fs.mkdir(path.join(committedStaging, "node")); + await fs.writeFile(path.join(committedStaging, "node/child.txt"), "committed\n"); + expect(await runTransaction("apply", committedNonce)).toMatchObject({ code: 0, stderr: "" }); + const committedTransaction = path.dirname(committedStaging); + const interruptedCleanup = path.join( + path.dirname(committedTransaction), + path + .basename(committedTransaction) + .replace(".openclaw-accepted-", ".openclaw-accepted-cleanup-"), + ); + await fs.rename(committedTransaction, interruptedCleanup); + + const cleanupNonce = "d".repeat(32); + const cleanupBegin = await runTransaction("begin", cleanupNonce, JSON.stringify(["node"])); + expect(cleanupBegin.code).toBe(0); + expect((await runTransaction("rollback", cleanupNonce)).code).toBe(0); + + await expect(fs.readFile(path.join(workspace, "node/child.txt"), "utf8")).resolves.toBe( + "committed\n", + ); + await expect(fs.access(interruptedCleanup)).rejects.toThrow(); + + await fs.chmod(path.join(workspace, "node"), 0o555); + const modeRollbackNonce = "e".repeat(32); + const modeRollbackBegin = await runTransaction( + "begin", + modeRollbackNonce, + JSON.stringify(["node"]), + ); + const modeRollbackStaging = modeRollbackBegin.stdout.trim(); + await fs.mkdir(path.join(modeRollbackStaging, "node")); + await fs.writeFile(path.join(modeRollbackStaging, "node/replacement.txt"), "replacement\n"); + expect((await runTransaction("apply", modeRollbackNonce)).code).toBe(0); + expect((await runTransaction("rollback", modeRollbackNonce)).code).toBe(0); + expect((await fs.stat(path.join(workspace, "node"))).mode & 0o777).toBe(0o555); + await expect(fs.readFile(path.join(workspace, "node/child.txt"), "utf8")).resolves.toBe( + "committed\n", + ); + await fs.chmod(path.join(workspace, "node"), 0o700); + + const interruptedModeNonce = "1".repeat(32); + const interruptedModeBegin = await runTransaction( + "begin", + interruptedModeNonce, + JSON.stringify(["node"]), + ); + const interruptedModeTransaction = path.dirname(interruptedModeBegin.stdout.trim()); + await fs.writeFile( + path.join(interruptedModeTransaction, "state.json"), + JSON.stringify([{ relative: "node", hadLive: true, directoryMode: 0o555 }]), + { mode: 0o600 }, + ); + expect((await runTransaction("rollback", interruptedModeNonce)).code).toBe(0); + expect((await fs.stat(path.join(workspace, "node"))).mode & 0o777).toBe(0o555); + await fs.chmod(path.join(workspace, "node"), 0o700); + + await fs.mkdir(path.join(workspace, "parent")); + await fs.writeFile(path.join(workspace, "parent/child.txt"), "before\n"); + const ancestorModeNonce = "2".repeat(32); + const ancestorModeBegin = await runTransaction( + "begin", + ancestorModeNonce, + JSON.stringify(["parent/child.txt"]), + ); + const ancestorModeStaging = ancestorModeBegin.stdout.trim(); + await fs.mkdir(path.join(ancestorModeStaging, "parent")); + await fs.writeFile(path.join(ancestorModeStaging, "parent/child.txt"), "after\n"); + await fs.chmod(path.join(workspace, "parent"), 0o555); + await fs.chmod(workspace, 0o555); + expect(await runTransaction("apply", ancestorModeNonce)).toMatchObject({ code: 0, stderr: "" }); + await expect(fs.readFile(path.join(workspace, "parent/child.txt"), "utf8")).resolves.toBe( + "after\n", + ); + expect((await fs.stat(workspace)).mode & 0o777).toBe(0o555); + expect((await fs.stat(path.join(workspace, "parent"))).mode & 0o777).toBe(0o555); + expect((await runTransaction("rollback", ancestorModeNonce)).code).toBe(0); + await expect(fs.readFile(path.join(workspace, "parent/child.txt"), "utf8")).resolves.toBe( + "before\n", + ); + expect((await fs.stat(workspace)).mode & 0o777).toBe(0o555); + expect((await fs.stat(path.join(workspace, "parent"))).mode & 0o777).toBe(0o555); + await fs.chmod(workspace, 0o700); + await fs.chmod(path.join(workspace, "parent"), 0o700); + }); + + it("keeps the gateway's canonical manifest available across a second turn", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-manifest-lifecycle-test-")); + roots.push(root); + const home = path.join(root, "home"); + const workspace = path.join(root, "workspace"); + await Promise.all([fs.mkdir(home), fs.mkdir(workspace)]); + await fs.writeFile(path.join(workspace, ".gitignore"), ""); + for (const args of [ + ["init", "--quiet"], + ["add", ".gitignore"], + [ + "-c", + "user.name=OpenClaw Test", + "-c", + "user.email=test@openclaw.invalid", + "commit", + "--quiet", + "-m", + "base", + ], + ]) { + const result = await runCommandWithTimeout(["git", "-C", workspace, ...args], { + timeoutMs: 10_000, + }); + expect(result.code).toBe(0); + } + const baseCommit = ( + await runCommandWithTimeout(["git", "-C", workspace, "rev-parse", "HEAD"], { + timeoutMs: 10_000, + }) + ).stdout.trim(); + const env = { ...process.env, HOME: home }; + const initial = await runCommandWithTimeout( + [process.execPath, "-e", REMOTE_WORKSPACE_MANIFEST_JS, workspace, baseCommit, "eligible"], + { timeoutMs: 10_000, baseEnv: env }, + ); + expect(initial.code).toBe(0); + + await fs.writeFile(path.join(workspace, "notes.md"), "cloud edit\n", { mode: 0o664 }); + await Promise.all([ + fs.writeFile(path.join(workspace, "Zebra.md"), "upper\n"), + fs.writeFile(path.join(workspace, "éclair.md"), "unicode\n"), + fs.writeFile(path.join(workspace, "älg.md"), "collation\n"), + ]); + const firstTurn = await runCommandWithTimeout( + [ + process.execPath, + "-e", + REMOTE_WORKSPACE_MANIFEST_JS, + workspace, + baseCommit, + "eligible", + initial.stdout.trim().slice("sha256:".length), + ], + { timeoutMs: 10_000, baseEnv: env }, + ); + expect(firstTurn.code).toBe(0); + const firstTurnRef = firstTurn.stdout.trim(); + const firstTurnDigest = firstTurnRef.slice("sha256:".length); + const manifestRoot = path.join(home, ".openclaw-worker", "manifests"); + const firstTurnPath = path.join(manifestRoot, `${firstTurnDigest}.json`); + const firstTurnRaw = await fs.readFile(firstTurnPath, "utf8"); + const firstTurnManifest = parseWorkerWorkspaceManifest(firstTurnRaw, firstTurnRef); + expect(firstTurnRaw).toBe(serializeWorkerWorkspaceManifest(firstTurnManifest)); + const firstTurnPaths = ( + JSON.parse(firstTurnRaw) as { entries: Array<{ path: string }> } + ).entries.map((entry) => entry.path); + expect(firstTurnPaths).toEqual( + firstTurnPaths.toSorted((left, right) => (left < right ? -1 : left > right ? 1 : 0)), + ); + + await fs.rm(firstTurnPath); + const published = await runCommandWithTimeout( + [ + process.execPath, + "-e", + REMOTE_WORKSPACE_MANIFEST_JS, + workspace, + "", + "publish", + firstTurnDigest, + ], + { timeoutMs: 10_000, baseEnv: env, input: firstTurnRaw }, + ); + expect(published.code).toBe(0); + expect(published.stdout.trim()).toBe(firstTurnRef); + await expect(fs.readFile(firstTurnPath, "utf8")).resolves.toBe(firstTurnRaw); + + const legacy = JSON.parse(firstTurnRaw) as { + entries: Array<{ path: string; type: string; mode: number }>; + }; + for (const entry of legacy.entries) { + if (entry.path === "notes.md") { + entry.mode = 0o664; + } + } + legacy.entries.sort((left, right) => + left.path < right.path ? -1 : left.path > right.path ? 1 : 0, + ); + const legacyRaw = JSON.stringify(legacy); + const legacyDigest = createHash("sha256").update(legacyRaw).digest("hex"); + await fs.writeFile(path.join(manifestRoot, `${legacyDigest}.json`), legacyRaw); + await fs.rm(firstTurnPath); + + const legacyCanonical = structuredClone(legacy); + for (const entry of legacyCanonical.entries) { + if (entry.type === "directory") { + entry.mode = 0o700; + } else if (entry.type === "symlink") { + entry.mode = 0o777; + } else { + entry.mode = (entry.mode & 0o111) === 0 ? 0o644 : 0o755; + } + } + const legacyProducerLocale = "en-US"; + legacyCanonical.entries.sort((left, right) => + left.path.localeCompare(right.path, legacyProducerLocale), + ); + const acceptedRaw = JSON.stringify(legacyCanonical); + const acceptedDigest = createHash("sha256").update(acceptedRaw).digest("hex"); + const acceptedRef = `sha256:${acceptedDigest}`; + const acceptedPath = path.join(manifestRoot, `${acceptedDigest}.json`); + + const recovered = await runCommandWithTimeout( + [ + process.execPath, + "-e", + REMOTE_WORKSPACE_MANIFEST_JS, + workspace, + "", + "resolve", + acceptedDigest, + legacyProducerLocale, + ], + { timeoutMs: 10_000, baseEnv: env }, + ); + expect(recovered.code).toBe(0); + expect(recovered.stdout.trim()).toBe(acceptedRef); + await expect(fs.readFile(acceptedPath, "utf8")).resolves.toBe(acceptedRaw); + + await fs.writeFile(path.join(workspace, "notes.md"), "second cloud edit\n"); + const secondTurn = await runCommandWithTimeout( + [ + process.execPath, + "-e", + REMOTE_WORKSPACE_MANIFEST_JS, + workspace, + baseCommit, + "eligible", + acceptedDigest, + ], + { timeoutMs: 10_000, baseEnv: env }, + ); + expect(secondTurn.code).toBe(0); + expect(secondTurn.stdout.trim()).not.toBe(acceptedRef); + }); + it("drops derived artifacts from the worker manifest", async () => { const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-manifest-derived-test-")); roots.push(root); diff --git a/src/gateway/worker-environments/workspace-sync-scripts.ts b/src/gateway/worker-environments/workspace-sync-scripts.ts index ada5c6050885..dc0d005ca90a 100644 --- a/src/gateway/worker-environments/workspace-sync-scripts.ts +++ b/src/gateway/worker-environments/workspace-sync-scripts.ts @@ -1,3 +1,8 @@ +import { + REMOTE_WORKSPACE_MANIFEST_CANONICAL_JS, + REMOTE_WORKSPACE_MANIFEST_REGISTRY_JS, +} from "./workspace-manifest-remote-script.js"; +export { REMOTE_WORKSPACE_ACCEPTED_TRANSACTION_JS } from "./workspace-manifest-remote-script.js"; import { DERIVED_WORKSPACE_DIRECTORY_NAMES, DERIVED_WORKSPACE_FILE_NAMES, @@ -477,11 +482,15 @@ const isDerivedWorkspacePath = ${isDerivedWorkspacePath.toString()}; const root = fs.realpathSync(process.argv[1]); const requestedBaseCommit = process.argv[2] || null; const eligibleOnly = process.argv[3] === "eligible"; +const requestedManifestDigest = process.argv[3] === "resolve" ? process.argv[4] : null; +const publishedManifestDigest = process.argv[3] === "publish" ? process.argv[4] : null; +const legacyGatewayLocale = requestedManifestDigest ? process.argv[5] : null; const priorManifestDigests = [...new Set(process.argv.slice(4).filter(Boolean))]; const entriesByPath = new Map(); function fail(message) { throw new Error(message); } +${REMOTE_WORKSPACE_MANIFEST_CANONICAL_JS} function addEntry(relative) { if ( !relative || @@ -611,9 +620,7 @@ function eligiblePaths() { return [...selected].filter((relative) => !isDerivedWorkspacePath(relative)).sort(); } async function hashFiles() { - const entries = [...entriesByPath.values()].sort((a, b) => - a.path < b.path ? -1 : a.path > b.path ? 1 : 0, - ); + const entries = [...entriesByPath.values()]; for (const entry of entries) { if (entry.type !== "file") { continue; @@ -642,7 +649,27 @@ function ensurePrivateDirectory(directory) { } fs.chmodSync(directory, 0o700); } +${REMOTE_WORKSPACE_MANIFEST_REGISTRY_JS} async function main() { + const workerRoot = path.join(process.env.HOME, ".openclaw-worker"); + const manifestRoot = path.join(workerRoot, "manifests"); + ensurePrivateDirectory(workerRoot); + ensurePrivateDirectory(manifestRoot); + if (publishedManifestDigest) { + const manifest = fs.readFileSync(0, "utf8"); + if (crypto.createHash("sha256").update(manifest).digest("hex") !== publishedManifestDigest) { + fail("published workspace manifest digest mismatch"); + } + if (publishManifest(manifestRoot, manifest) !== publishedManifestDigest) { + fail("published workspace manifest reference mismatch"); + } + process.stdout.write("sha256:" + publishedManifestDigest + "\n"); + return; + } + if (requestedManifestDigest) { + process.stdout.write("sha256:" + resolveManifest(manifestRoot, requestedManifestDigest) + "\n"); + return; + } if (eligibleOnly) { for (const relative of eligiblePaths()) addWithParents(relative); } else { @@ -650,32 +677,8 @@ async function main() { } const entries = await hashFiles(); const baseCommit = requestedBaseCommit; - const manifest = JSON.stringify({ version: 1, baseCommit, entries }); - const digest = crypto.createHash("sha256").update(manifest).digest("hex"); - const workerRoot = path.join(process.env.HOME, ".openclaw-worker"); - const manifestRoot = path.join(workerRoot, "manifests"); - ensurePrivateDirectory(workerRoot); - ensurePrivateDirectory(manifestRoot); - const manifestPath = path.join(manifestRoot, digest + ".json"); - const temporaryPath = manifestPath + "." + process.pid + "." + crypto.randomBytes(4).toString("hex"); - fs.writeFileSync(temporaryPath, manifest, { encoding: "utf8", flag: "wx", mode: 0o600 }); - try { - try { - fs.linkSync(temporaryPath, manifestPath); - } catch (error) { - const existing = error && error.code === "EEXIST" ? fs.lstatSync(manifestPath) : null; - if ( - !existing || - existing.isSymbolicLink() || - !existing.isFile() || - fs.readFileSync(manifestPath, "utf8") !== manifest - ) { - throw error; - } - } - } finally { - fs.rmSync(temporaryPath, { force: true }); - } + const manifest = serializeManifest(baseCommit, entries); + const digest = publishManifest(manifestRoot, manifest); process.stdout.write("sha256:" + digest + "\n"); } main().catch((error) => { diff --git a/src/gateway/worker-environments/workspace-sync.ts b/src/gateway/worker-environments/workspace-sync.ts index 03b7b3233dcf..dd40ad655dca 100644 --- a/src/gateway/worker-environments/workspace-sync.ts +++ b/src/gateway/worker-environments/workspace-sync.ts @@ -2,12 +2,7 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import type { CommandOptions, SpawnResult } from "../../process/exec.js"; -import { - type PreparedWorkerSsh, - workerSshCommandOptions, - workerSshOptions, - workerSshRemoteCommand, -} from "./ssh.js"; +import { type PreparedWorkerSsh, workerSshCommandOptions } from "./ssh.js"; import type { WorkerTunnelHandle, WorkerWorkspaceCommand, @@ -16,6 +11,10 @@ import type { WorkerWorkspaceSyncRequest, WorkerWorkspaceSyncResult, } from "./tunnel-contract.js"; +import { + createAcceptedWorkspacePublisherFactory, + recoverAcceptedWorkspacePublication, +} from "./workspace-accepted-sync.js"; import { DERIVED_WORKSPACE_RSYNC_EXCLUDES } from "./workspace-path-exclusions.js"; import { applyStagedWorkerWorkspace, @@ -33,16 +32,19 @@ import { workerWorkspaceTransferPaths, } from "./workspace-result-staging.js"; import { - MANIFEST_REF_PATTERN, parseManifestRef, parseRemoteWorkspaceDirectory, probeWorkspaceGitMode, readTransferredManifest, + resolveRemoteWorkspaceManifest, runBoundedInboundRsync as runBoundedInboundRsyncTransfer, stableWorkerPathComponent, validateWorkspaceSyncRequest, + verifyRemoteWorkspaceManifest, waitForQuiescenceRenewal, workerWorkspaceCommandSucceeded as success, + workerWorkspaceRsyncRemoteCommand, + workerWorkspaceSshArgv, workspaceSyncError, type WorkerWorkspaceActionsOptions, } from "./workspace-sync-helpers.js"; @@ -112,18 +114,7 @@ export function createWorkerWorkspaceActions( const runWorkspaceCommand = async (command: WorkerWorkspaceCommand): Promise => { const prepared = requirePrepared(); return await runTask( - [ - "ssh", - ...workerSshOptions(prepared, { forwarding: "disabled" }), - "-a", - "-x", - "-T", - "-p", - String(prepared.port), - "--", - prepared.sshTarget, - workerSshRemoteCommand(command.argv), - ], + workerWorkspaceSshArgv(prepared, command.argv), workerSshCommandOptions({ input: command.input, timeoutMs: command.timeoutMs ?? WORKSPACE_TIMEOUT_MS, @@ -267,15 +258,7 @@ export function createWorkerWorkspaceActions( const temporaryDirectory = await fs.mkdtemp( path.join(os.tmpdir(), "openclaw-worker-workspace-sync-"), ); - const rsyncSsh = workerSshRemoteCommand([ - "ssh", - ...workerSshOptions(prepared, { forwarding: "disabled" }), - "-a", - "-x", - "-T", - "-p", - String(prepared.port), - ]); + const rsyncSsh = workerWorkspaceRsyncRemoteCommand(prepared); try { let fileListPath: string | undefined; if (mode === "git") { @@ -490,10 +473,11 @@ export function createWorkerWorkspaceActions( await recoverWorkerWorkspaceReconciliation({ root: request.localPath, journal: pending }); request.journal.abort(); } - const baseDigest = MANIFEST_REF_PATTERN.exec(request.baseManifestRef)?.[0]?.slice(7); - if (!baseDigest) { - throw new Error("Worker workspace base manifest reference is invalid"); - } + const baseDigest = await resolveRemoteWorkspaceManifest( + runWorkspaceCommand, + request.remoteWorkspaceDir, + request.baseManifestRef, + ); const prepared = requirePrepared(); const temporaryDirectory = await fs.mkdtemp( path.join(os.tmpdir(), "openclaw-worker-workspace-reconcile-"), @@ -502,15 +486,16 @@ export function createWorkerWorkspaceActions( const manifestRoot = path.join(temporaryDirectory, "manifests"); const baseManifestPath = path.join(manifestRoot, `${baseDigest}.json`); const transferListPath = path.join(temporaryDirectory, "transfer-list"); - const rsyncSsh = workerSshRemoteCommand([ - "ssh", - ...workerSshOptions(prepared, { forwarding: "disabled" }), - "-a", - "-x", - "-T", - "-p", - String(prepared.port), - ]); + const rsyncSsh = workerWorkspaceRsyncRemoteCommand(prepared); + const acceptedWorkspacePublisher = createAcceptedWorkspacePublisherFactory({ + runWorkspaceCommand, + runTask, + ownerSignal: options.ownerSignal, + rsyncSsh, + scpTarget: prepared.scpTarget, + localPath: request.localPath, + remoteWorkspaceDir: request.remoteWorkspaceDir, + }); try { await fs.mkdir(stagingRoot, { mode: 0o700 }); await fs.mkdir(manifestRoot, { mode: 0o700 }); @@ -538,28 +523,20 @@ export function createWorkerWorkspaceActions( const baseRaw = await readTransferredManifest(baseManifestPath); const base = parseWorkerWorkspaceManifest(baseRaw, request.baseManifestRef); await fs.rm(baseManifestPath); - await assertWorkspaceMatchesManifest({ root: request.localPath, manifest: base }); - const verifyStable = async (expectedRef: string): Promise => { - const expectedDigest = expectedRef.slice("sha256:".length); - const verified = await runWorkspaceCommand({ - argv: [ - "node", - "-e", - REMOTE_WORKSPACE_MANIFEST_JS, - request.remoteWorkspaceDir, - base.baseCommit ?? "", - // The accepted result omits deleted paths. Seed both manifests so a - // deleted path recreated under a new ignore rule still invalidates the fence. - ...(base.baseCommit ? ["eligible", expectedDigest, baseDigest] : []), - ], + // Finish or undo any interrupted accepted-state publication before measuring + // the current worker tree; otherwise reconciliation would plan from a partial swap. + await recoverAcceptedWorkspacePublication({ + runWorkspaceCommand, + remoteWorkspaceDir: request.remoteWorkspaceDir, + }); + const verifyStable = async (expectedRef: string): Promise => + await verifyRemoteWorkspaceManifest({ + runWorkspaceCommand, + remoteWorkspaceDir: request.remoteWorkspaceDir, + baseCommit: base.baseCommit, + baseDigest, + expectedRef, }); - if (!success(verified)) { - throw workspaceSyncError(verified); - } - if (parseManifestRef(verified.stdout.trim()) !== expectedRef) { - throw new Error("Cloud workspace changed during final reconciliation"); - } - }; const currentResult = await runWorkspaceCommand({ argv: [ "node", @@ -576,6 +553,10 @@ export function createWorkerWorkspaceActions( } const currentRef = parseManifestRef(currentResult.stdout.trim()); if (currentRef === request.baseManifestRef) { + const { expectedRemoteRef, publishAcceptedManifest } = acceptedWorkspacePublisher( + base, + currentRef, + ); await verifyStable(currentRef); const stagedResult = request.stagedResult ? await workerWorkspaceResultStaging.prepareRequestedWorkerWorkspaceResult({ @@ -584,21 +565,32 @@ export function createWorkerWorkspaceActions( currentManifestRef: currentRef, baseManifestRaw: baseRaw, currentManifestRaw: baseRaw, + publishAcceptedManifest, }) : undefined; + let appliedWorkspaceResult: WorkerWorkspaceApplyResult | undefined; if (!stagedResult) { - request.journal.commit(currentRef); + appliedWorkspaceResult = await applyStagedWorkerWorkspace({ + root: request.localPath, + stagingRoot, + baseManifestRef: request.baseManifestRef, + currentManifestRef: currentRef, + base, + current: base, + journal: request.journal, + publishAcceptedManifest, + }); } return { - manifestRef: currentRef, + get manifestRef() { + return expectedRemoteRef(); + }, changed: false, - verifyStable: async () => await verifyStable(currentRef), + verifyStable: async () => await verifyStable(expectedRemoteRef()), verifyLocalStable: async () => - await assertWorkspaceResultStable({ - root: request.localPath, - base, - current: base, - }), + await (appliedWorkspaceResult?.verifyLocalStable() ?? + assertWorkspaceResultStable({ root: request.localPath, base, current: base })), + getAppliedWorkspaceResult: () => appliedWorkspaceResult, ...stagedResult, }; } @@ -627,6 +619,10 @@ export function createWorkerWorkspaceActions( } const currentRaw = await readTransferredManifest(currentManifestPath); const current = parseWorkerWorkspaceManifest(currentRaw, currentRef); + const { expectedRemoteRef, publishAcceptedManifest } = acceptedWorkspacePublisher( + current, + currentRef, + ); const transferPaths = workerWorkspaceTransferPaths(current, base); const transferPathSet = new Set(transferPaths); if (transferPaths.length > 0) { @@ -672,6 +668,7 @@ export function createWorkerWorkspaceActions( currentManifestRef: currentRef, baseManifestRaw: baseRaw, currentManifestRaw: currentRaw, + publishAcceptedManifest, }) : undefined; let appliedWorkspaceResult: WorkerWorkspaceApplyResult | undefined; @@ -684,12 +681,15 @@ export function createWorkerWorkspaceActions( base, current, journal: request.journal, + publishAcceptedManifest, }); } return { - manifestRef: currentRef, + get manifestRef() { + return expectedRemoteRef(); + }, changed: true, - verifyStable: async () => await verifyStable(currentRef), + verifyStable: async () => await verifyStable(expectedRemoteRef()), verifyLocalStable: async () => appliedWorkspaceResult ? await appliedWorkspaceResult.verifyLocalStable()