mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-21 10:16:44 +00:00
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
This commit is contained in:
Peter Steinberger
@@ -59,14 +59,17 @@ function deferred<T>() {
|
||||
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(
|
||||
|
||||
@@ -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<SpawnResult>;
|
||||
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<SpawnResult>;
|
||||
runTask: (argv: string[], options: CommandOptions) => Promise<SpawnResult>;
|
||||
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<Parameters<typeof createAcceptedWorkspacePublisher>[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;
|
||||
},
|
||||
};
|
||||
};
|
||||
}
|
||||
@@ -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");
|
||||
}`;
|
||||
@@ -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),
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -39,9 +39,19 @@ export async function applyStagedWorkerWorkspace(params: {
|
||||
base: WorkerWorkspaceManifest;
|
||||
current: WorkerWorkspaceManifest;
|
||||
journal: WorkerWorkspaceReconciliationJournalAdapter;
|
||||
publishAcceptedManifest?: (accepted: {
|
||||
manifestRef: string;
|
||||
manifest: WorkerWorkspaceManifest;
|
||||
conflictPaths: string[];
|
||||
}) => Promise<void>;
|
||||
}): Promise<WorkerWorkspaceApplyResult> {
|
||||
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) {
|
||||
|
||||
@@ -189,6 +189,7 @@ export async function readActualWorkspaceManifest(params: {
|
||||
root: string;
|
||||
baseCommit: string | null;
|
||||
preserveDirectories?: ReadonlySet<string>;
|
||||
includePaths?: ReadonlySet<string>;
|
||||
}): 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<string>;
|
||||
includePaths?: ReadonlySet<string>;
|
||||
}): Promise<void> {
|
||||
const actual = await readActualWorkspaceManifest(params);
|
||||
if (actual.manifestRef !== params.expectedRef) {
|
||||
|
||||
@@ -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<void>;
|
||||
}) {
|
||||
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",
|
||||
|
||||
@@ -456,6 +456,11 @@ export async function applyStagedWorkerWorkspaceResult(params: {
|
||||
expectedBaseManifestRef: string;
|
||||
alreadyAccepted?: boolean;
|
||||
journal: WorkerWorkspaceReconciliationJournalAdapter;
|
||||
publishAcceptedManifest?: (accepted: {
|
||||
manifestRef: string;
|
||||
manifest: WorkerWorkspaceManifest;
|
||||
conflictPaths: string[];
|
||||
}) => Promise<void>;
|
||||
}): Promise<WorkerWorkspaceApplyResult & { changed: boolean }> {
|
||||
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<void>;
|
||||
}): Promise<{
|
||||
applyPreparedStagedResult(): Promise<void>;
|
||||
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,
|
||||
|
||||
@@ -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<SpawnResult>,
|
||||
remoteWorkspaceDir: string,
|
||||
expectedRef: string,
|
||||
): Promise<string> {
|
||||
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<SpawnResult>,
|
||||
remoteWorkspaceDir: string,
|
||||
expectedRef: string,
|
||||
) {
|
||||
return await resolveRemoteWorkspaceBaseManifest(
|
||||
runWorkspaceCommand,
|
||||
remoteWorkspaceDir,
|
||||
expectedRef,
|
||||
);
|
||||
}
|
||||
|
||||
export async function verifyRemoteWorkspaceManifest(params: {
|
||||
runWorkspaceCommand: (command: WorkerWorkspaceCommand) => Promise<SpawnResult>;
|
||||
remoteWorkspaceDir: string;
|
||||
baseCommit: string | null;
|
||||
baseDigest: string;
|
||||
expectedRef: string;
|
||||
}): Promise<void> {
|
||||
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;
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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) => {
|
||||
|
||||
@@ -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<SpawnResult> => {
|
||||
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<void> => {
|
||||
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<void> =>
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user