mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-27 22:41:14 +00:00
fix: cloud-worker results are lost when the box dies before reconciliation (#110952)
* fix(cloud-workers): stage worker results durably before reconciliation * fix(cloud-workers): model result staging test seam * fix(cloud-workers): disable staging git hooks safely * chore: defer cloud worker release note * fix(cloud-workers): harden staged result recovery
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
a21385a372
commit
7a7d6bb51f
@@ -3,7 +3,7 @@
|
||||
"version": "2026.7.2",
|
||||
"openclaw": {
|
||||
"schemaVersions": {
|
||||
"state": 4,
|
||||
"state": 5,
|
||||
"agent": 13
|
||||
}
|
||||
},
|
||||
|
||||
@@ -39,6 +39,7 @@ export type WorkerDispatchPlacementStore = Pick<
|
||||
| "listWorkspaceReconciliationOwners"
|
||||
| "listPendingWorkspaceResults"
|
||||
| "workspaceResultInstanceId"
|
||||
| "recordStagedWorkspaceResult"
|
||||
| "acceptWorkspaceResult"
|
||||
| "completeWorkspaceResultAndReleaseTurn"
|
||||
| "abandonWorkspaceResult"
|
||||
|
||||
@@ -13,6 +13,14 @@ import {
|
||||
import type { WorkerEnvironmentService } from "./service.js";
|
||||
import { verifyReconciledWorkspaceFinal } from "./workspace-finalize.js";
|
||||
import type { WorkerWorkspaceOperationCoordinator } from "./workspace-operation-coordinator.js";
|
||||
import { recoverWorkerWorkspaceReconciliation } from "./workspace-reconcile.js";
|
||||
import {
|
||||
applyStagedWorkerWorkspaceResult,
|
||||
deleteStagedWorkerWorkspaceResult,
|
||||
hasWorkerWorkspaceResultRef,
|
||||
preparedWorkerWorkspaceResultRef,
|
||||
workerWorkspaceResultRef,
|
||||
} from "./workspace-result-staging.js";
|
||||
|
||||
function sameActiveEnvironment(
|
||||
placement: WorkerActiveDispatchPlacement | WorkerDrainingDispatchPlacement,
|
||||
@@ -59,7 +67,11 @@ export function createPlacementRecoveryActions(deps: {
|
||||
const { environments, failure, placements } = deps;
|
||||
|
||||
const recoverPendingWorkspaceResults = async (): Promise<Set<string>> => {
|
||||
const stagedResultOwners = new Set<string>();
|
||||
for (const pending of placements.listPendingWorkspaceResults()) {
|
||||
if (pending.stagedResultRef) {
|
||||
stagedResultOwners.add(pending.sessionId);
|
||||
}
|
||||
const sameGatewayInstance =
|
||||
pending.gatewayInstanceId === placements.workspaceResultInstanceId();
|
||||
if (sameGatewayInstance && pending.recoveryRequestedAtMs === null) {
|
||||
@@ -78,6 +90,23 @@ export function createPlacementRecoveryActions(deps: {
|
||||
claim.generation !== pending.placementGeneration ||
|
||||
claim.ownerEpoch !== pending.ownerEpoch
|
||||
) {
|
||||
if (pending.stagedResultRef && pending.workspaceAcceptedAtMs === null) {
|
||||
// A staged unaccepted result outlives stale placement ownership. Only
|
||||
// explicit operator abandonment may delete its durable Git ref.
|
||||
continue;
|
||||
}
|
||||
if (pending.stagedResultRef) {
|
||||
if (!placement) {
|
||||
throw new Error(
|
||||
`Staged cloud workspace result lost its placement: ${pending.sessionId}`,
|
||||
);
|
||||
}
|
||||
const root = await deps.resolveWorkspacePath(placement);
|
||||
await deleteStagedWorkerWorkspaceResult({
|
||||
root,
|
||||
stagedResultRef: pending.stagedResultRef,
|
||||
});
|
||||
}
|
||||
placements.abandonWorkspaceResult(pending);
|
||||
if (placement?.state === "active") {
|
||||
await failure.failActive(
|
||||
@@ -107,6 +136,30 @@ export function createPlacementRecoveryActions(deps: {
|
||||
ownerEpoch: placement.activeOwnerEpoch,
|
||||
},
|
||||
};
|
||||
const localPath = await deps.resolveWorkspacePath({
|
||||
sessionId: placement.sessionId,
|
||||
sessionKey: placement.sessionKey,
|
||||
agentId: placement.agentId,
|
||||
});
|
||||
const canonicalStagedResultRef = workerWorkspaceResultRef(turnClaim.claimId);
|
||||
let stagedResultRef = pending.stagedResultRef;
|
||||
if (
|
||||
!stagedResultRef &&
|
||||
(await hasWorkerWorkspaceResultRef({
|
||||
root: localPath,
|
||||
stagedResultRef: canonicalStagedResultRef,
|
||||
}))
|
||||
) {
|
||||
placements.recordStagedWorkspaceResult(turnClaim, canonicalStagedResultRef);
|
||||
stagedResultRef = canonicalStagedResultRef;
|
||||
stagedResultOwners.add(pending.sessionId);
|
||||
}
|
||||
const hasPreparedResult =
|
||||
!stagedResultRef &&
|
||||
(await hasWorkerWorkspaceResultRef({
|
||||
root: localPath,
|
||||
stagedResultRef: preparedWorkerWorkspaceResultRef(canonicalStagedResultRef),
|
||||
}));
|
||||
const environment = environments.get(placement.environmentId);
|
||||
if (
|
||||
environment?.state === "attached" &&
|
||||
@@ -117,7 +170,82 @@ export function createPlacementRecoveryActions(deps: {
|
||||
// Keep the durable claim fenced until environment ownership is unambiguous.
|
||||
continue;
|
||||
}
|
||||
if (stagedResultRef) {
|
||||
// A staged result must never be destroyed by environment lifecycle.
|
||||
// Keep its fence and placement until the local apply is durably accepted.
|
||||
const owner = {
|
||||
sessionId: placement.sessionId,
|
||||
environmentId: placement.environmentId,
|
||||
ownerEpoch: placement.activeOwnerEpoch,
|
||||
placementGeneration: placement.generation,
|
||||
};
|
||||
const journal = {
|
||||
load: () => placements.loadWorkspaceReconciliation(owner),
|
||||
begin: (next: Parameters<typeof placements.beginWorkspaceReconciliation>[1]) =>
|
||||
placements.beginWorkspaceReconciliation(owner, next),
|
||||
commit: (manifestRef: string) =>
|
||||
placements.updateWorkspaceBaseManifest({ claim: turnClaim, manifestRef }),
|
||||
abort: () => placements.abortWorkspaceReconciliation(owner),
|
||||
};
|
||||
await deps.workspaceOperations.run(placement.environmentId, async () => {
|
||||
const owned = placements.get(placement.sessionId);
|
||||
const ownedClaim = owned?.turnClaim;
|
||||
if (
|
||||
(owned?.state !== "active" && owned?.state !== "draining") ||
|
||||
owned.generation !== placement.generation ||
|
||||
owned.environmentId !== placement.environmentId ||
|
||||
owned.activeOwnerEpoch !== placement.activeOwnerEpoch ||
|
||||
ownedClaim?.owner !== "worker" ||
|
||||
ownedClaim.claimId !== claim.claimId ||
|
||||
ownedClaim.runId !== claim.runId
|
||||
) {
|
||||
throw new Error("Recovered workspace result lost its placement owner");
|
||||
}
|
||||
const interrupted = journal.load();
|
||||
if (interrupted) {
|
||||
await recoverWorkerWorkspaceReconciliation({ root: localPath, journal: interrupted });
|
||||
journal.abort();
|
||||
}
|
||||
if (pending.workspaceAcceptedAtMs === null) {
|
||||
const reconciliation = await applyStagedWorkerWorkspaceResult({
|
||||
root: localPath,
|
||||
stagedResultRef,
|
||||
expectedBaseManifestRef: placement.workspaceBaseManifestRef,
|
||||
journal,
|
||||
});
|
||||
await reconciliation.verifyLocalStable();
|
||||
placements.acceptWorkspaceResult(turnClaim);
|
||||
}
|
||||
await deleteStagedWorkerWorkspaceResult({
|
||||
root: localPath,
|
||||
stagedResultRef,
|
||||
});
|
||||
const currentEnvironment = environments.get(placement.environmentId);
|
||||
if (
|
||||
currentEnvironment &&
|
||||
currentEnvironment.state !== "destroyed" &&
|
||||
currentEnvironment.ownerEpoch === placement.activeOwnerEpoch
|
||||
) {
|
||||
await environments.destroy(placement.environmentId);
|
||||
}
|
||||
const reclaimed = placements.completeWorkspaceResultAndReleaseTurn(turnClaim, {
|
||||
reclaim: true,
|
||||
});
|
||||
if (reclaimed.state !== "reclaimed") {
|
||||
throw new Error("Recovered worker result did not reclaim its stale environment");
|
||||
}
|
||||
await environments
|
||||
.stopTunnel(placement.environmentId, placement.activeOwnerEpoch)
|
||||
.catch(() => undefined);
|
||||
});
|
||||
continue;
|
||||
}
|
||||
if (!sameActiveEnvironment(placement, environment)) {
|
||||
if (hasPreparedResult) {
|
||||
// Verification did not publish this prepared snapshot before the
|
||||
// crash. Preserve the fence for retry or operator inspection.
|
||||
continue;
|
||||
}
|
||||
if (pending.workspaceAcceptedAtMs !== null && environment?.state === "destroyed") {
|
||||
placements.completeWorkspaceResultAndReleaseTurn(turnClaim, { reclaim: true });
|
||||
continue;
|
||||
@@ -138,17 +266,20 @@ export function createPlacementRecoveryActions(deps: {
|
||||
}
|
||||
continue;
|
||||
}
|
||||
const localPath = await deps.resolveWorkspacePath({
|
||||
sessionId: placement.sessionId,
|
||||
sessionKey: placement.sessionKey,
|
||||
agentId: placement.agentId,
|
||||
});
|
||||
const owner = {
|
||||
sessionId: placement.sessionId,
|
||||
environmentId: placement.environmentId,
|
||||
ownerEpoch: placement.activeOwnerEpoch,
|
||||
placementGeneration: placement.generation,
|
||||
};
|
||||
const journal = {
|
||||
load: () => placements.loadWorkspaceReconciliation(owner),
|
||||
begin: (next: Parameters<typeof placements.beginWorkspaceReconciliation>[1]) =>
|
||||
placements.beginWorkspaceReconciliation(owner, next),
|
||||
commit: (manifestRef: string) =>
|
||||
placements.updateWorkspaceBaseManifest({ claim: turnClaim, manifestRef }),
|
||||
abort: () => placements.abortWorkspaceReconciliation(owner),
|
||||
};
|
||||
const tunnel = await environments.startTunnel({
|
||||
environmentId: placement.environmentId,
|
||||
ownerEpoch: placement.activeOwnerEpoch,
|
||||
@@ -175,15 +306,29 @@ export function createPlacementRecoveryActions(deps: {
|
||||
remoteWorkspaceDir: placement.remoteWorkspaceDir,
|
||||
baseManifestRef: placement.workspaceBaseManifestRef,
|
||||
journal: {
|
||||
load: () => placements.loadWorkspaceReconciliation(owner),
|
||||
begin: (journal) => placements.beginWorkspaceReconciliation(owner, journal),
|
||||
commit: (manifestRef) =>
|
||||
placements.updateWorkspaceBaseManifest({ claim: turnClaim, manifestRef }),
|
||||
abort: () => placements.abortWorkspaceReconciliation(owner),
|
||||
...journal,
|
||||
},
|
||||
stagedResult: {
|
||||
ref: canonicalStagedResultRef,
|
||||
record: (ref) => placements.recordStagedWorkspaceResult(turnClaim, ref),
|
||||
},
|
||||
});
|
||||
await verifyReconciledWorkspaceFinal(reconciliation, quiescence);
|
||||
placements.acceptWorkspaceResult(turnClaim);
|
||||
const recordedStagedResultRef = placements
|
||||
.listPendingWorkspaceResults()
|
||||
.find(
|
||||
(result) =>
|
||||
result.sessionId === turnClaim.sessionId &&
|
||||
result.claimId === turnClaim.claimId &&
|
||||
result.runId === turnClaim.runId,
|
||||
)?.stagedResultRef;
|
||||
if (recordedStagedResultRef) {
|
||||
await deleteStagedWorkerWorkspaceResult({
|
||||
root: localPath,
|
||||
stagedResultRef: recordedStagedResultRef,
|
||||
});
|
||||
}
|
||||
if (sameGatewayInstance) {
|
||||
await quiescence.resume();
|
||||
quiescenceHandled = true;
|
||||
@@ -211,7 +356,10 @@ export function createPlacementRecoveryActions(deps: {
|
||||
// Keep the result, claim, and environment fenced. The next sweep retries.
|
||||
}
|
||||
}
|
||||
return new Set(placements.listPendingWorkspaceResults().map((pending) => pending.sessionId));
|
||||
return new Set([
|
||||
...stagedResultOwners,
|
||||
...placements.listPendingWorkspaceResults().map((pending) => pending.sessionId),
|
||||
]);
|
||||
};
|
||||
|
||||
const adoptActive = async (placement: WorkerActiveDispatchPlacement): Promise<void> => {
|
||||
|
||||
@@ -0,0 +1,279 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import { runCommandWithTimeout } from "../../process/exec.js";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
openOpenClawStateDatabase,
|
||||
type OpenClawStateDatabase,
|
||||
} from "../../state/openclaw-state-db.js";
|
||||
import { type PlacementStore, REQUEST } from "./placement-dispatch-test-fixtures.js";
|
||||
import { createHarness } from "./placement-dispatch-test-harness.js";
|
||||
import { createWorkerSessionPlacementStore } from "./placement-store.js";
|
||||
import {
|
||||
workerWorkspaceResultRef,
|
||||
workerWorkspaceResultStaging,
|
||||
} from "./workspace-result-staging.js";
|
||||
|
||||
const { stageWorkerWorkspaceResult } = workerWorkspaceResultStaging;
|
||||
|
||||
describe("staged worker placement result recovery", () => {
|
||||
let root: string;
|
||||
let database: OpenClawStateDatabase;
|
||||
let placementStore: PlacementStore;
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await fs.mkdtemp(path.join(await fs.realpath(os.tmpdir()), "openclaw-staged-dispatch-"));
|
||||
database = openOpenClawStateDatabase({ env: { OPENCLAW_STATE_DIR: root } });
|
||||
placementStore = createWorkerSessionPlacementStore({ database, now: () => 1_000 });
|
||||
});
|
||||
|
||||
async function stagePendingResult(params: {
|
||||
store: PlacementStore;
|
||||
claim: ReturnType<PlacementStore["claimTurn"]>;
|
||||
workspacePath: string;
|
||||
base: string;
|
||||
current: string;
|
||||
record?: boolean;
|
||||
}): Promise<{ currentManifestRef: string; stagedResultRef: string }> {
|
||||
await fs.mkdir(params.workspacePath, { recursive: true });
|
||||
const initialized = await runCommandWithTimeout(
|
||||
["git", "-C", params.workspacePath, "init", "--quiet"],
|
||||
{ timeoutMs: 10_000 },
|
||||
);
|
||||
expect(initialized.code).toBe(0);
|
||||
const payload = path.join(params.workspacePath, ".staged-payload");
|
||||
await fs.mkdir(payload);
|
||||
await Promise.all([
|
||||
fs.writeFile(path.join(params.workspacePath, "result.txt"), params.base),
|
||||
fs.writeFile(path.join(payload, "result.txt"), params.current),
|
||||
]);
|
||||
const encode = (content: string) => {
|
||||
const raw = JSON.stringify({
|
||||
version: 1,
|
||||
baseCommit: null,
|
||||
entries: [
|
||||
{
|
||||
path: "result.txt",
|
||||
type: "file",
|
||||
mode: 0o644,
|
||||
size: Buffer.byteLength(content),
|
||||
sha256: createHash("sha256").update(content).digest("hex"),
|
||||
},
|
||||
],
|
||||
});
|
||||
return { raw, ref: `sha256:${createHash("sha256").update(raw).digest("hex")}` };
|
||||
};
|
||||
const base = encode(params.base);
|
||||
const current = encode(params.current);
|
||||
params.store.updateWorkspaceBaseManifest({ claim: params.claim, manifestRef: base.ref });
|
||||
params.store.markWorkspaceResultPending(params.claim);
|
||||
const stagedResultRef = workerWorkspaceResultRef(params.claim.claimId);
|
||||
await stageWorkerWorkspaceResult({
|
||||
root: params.workspacePath,
|
||||
stagingRoot: payload,
|
||||
stagedResultRef,
|
||||
baseManifestRef: base.ref,
|
||||
currentManifestRef: current.ref,
|
||||
baseManifestRaw: base.raw,
|
||||
currentManifestRaw: current.raw,
|
||||
});
|
||||
if (params.record !== false) {
|
||||
params.store.recordStagedWorkspaceResult(params.claim, stagedResultRef);
|
||||
}
|
||||
await fs.rm(payload, { recursive: true, force: true });
|
||||
return { currentManifestRef: current.ref, stagedResultRef };
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
await fs.rm(root, { recursive: true, force: true });
|
||||
});
|
||||
it("applies a staged pending result without a tunnel and reclaims the worker", async () => {
|
||||
const workspacePath = path.join(root, "same-worker-staged-result");
|
||||
const harness = createHarness(placementStore, { workspacePath });
|
||||
const active = harness.placements.seedActive(2);
|
||||
harness.markEnvironmentOwnerEpoch(2);
|
||||
if (active.state !== "active") {
|
||||
throw new Error("active placement fixture was not active");
|
||||
}
|
||||
const claim = placementStore.claimTurn({
|
||||
...REQUEST,
|
||||
claimId: "same-worker-staged-claim",
|
||||
runId: "same-worker-staged-run",
|
||||
owner: {
|
||||
kind: "worker",
|
||||
environmentId: active.environmentId,
|
||||
ownerEpoch: active.activeOwnerEpoch,
|
||||
},
|
||||
});
|
||||
const staged = await stagePendingResult({
|
||||
store: placementStore,
|
||||
claim,
|
||||
workspacePath,
|
||||
base: "base\n",
|
||||
current: "worker\n",
|
||||
});
|
||||
placementStore.handoffWorkspaceResultRecovery(claim);
|
||||
|
||||
await harness.service.reconcile();
|
||||
|
||||
await expect(fs.readFile(path.join(workspacePath, "result.txt"), "utf8")).resolves.toBe(
|
||||
"worker\n",
|
||||
);
|
||||
expect(harness.placements.current()).toMatchObject({
|
||||
state: "reclaimed",
|
||||
turnClaim: null,
|
||||
workspaceBaseManifestRef: staged.currentManifestRef,
|
||||
});
|
||||
expect(placementStore.listPendingWorkspaceResults()).toEqual([]);
|
||||
expect(harness.environments.startTunnel).not.toHaveBeenCalled();
|
||||
expect(harness.environments.destroy).toHaveBeenCalledWith(active.environmentId);
|
||||
expect(
|
||||
(
|
||||
await runCommandWithTimeout(
|
||||
["git", "-C", workspacePath, "show-ref", "--verify", staged.stagedResultRef],
|
||||
{ timeoutMs: 10_000 },
|
||||
)
|
||||
).code,
|
||||
).not.toBe(0);
|
||||
});
|
||||
|
||||
it("applies a staged result after restart even when the worker is dead", async () => {
|
||||
const workspacePath = path.join(root, "dead-worker-staged-result");
|
||||
const originalHarness = createHarness(placementStore, { workspacePath });
|
||||
const active = originalHarness.placements.seedActive(2);
|
||||
if (active.state !== "active") {
|
||||
throw new Error("active placement fixture was not active");
|
||||
}
|
||||
const claim = placementStore.claimTurn({
|
||||
...REQUEST,
|
||||
claimId: "dead-worker-staged-claim",
|
||||
runId: "dead-worker-staged-run",
|
||||
owner: {
|
||||
kind: "worker",
|
||||
environmentId: active.environmentId,
|
||||
ownerEpoch: active.activeOwnerEpoch,
|
||||
},
|
||||
});
|
||||
const staged = await stagePendingResult({
|
||||
store: placementStore,
|
||||
claim,
|
||||
workspacePath,
|
||||
base: "base\n",
|
||||
current: "worker\n",
|
||||
});
|
||||
const restartedStore = createWorkerSessionPlacementStore({ database, now: () => 2_000 });
|
||||
const restartedHarness = createHarness(restartedStore, { workspacePath });
|
||||
restartedHarness.markEnvironmentDestroyed();
|
||||
|
||||
await restartedHarness.service.reconcile();
|
||||
|
||||
await expect(fs.readFile(path.join(workspacePath, "result.txt"), "utf8")).resolves.toBe(
|
||||
"worker\n",
|
||||
);
|
||||
expect(restartedHarness.placements.current()).toMatchObject({
|
||||
state: "reclaimed",
|
||||
turnClaim: null,
|
||||
workspaceBaseManifestRef: staged.currentManifestRef,
|
||||
});
|
||||
expect(restartedStore.listPendingWorkspaceResults()).toEqual([]);
|
||||
expect(restartedHarness.environments.startTunnel).not.toHaveBeenCalled();
|
||||
expect(restartedHarness.log).not.toContain("placement:failed");
|
||||
});
|
||||
|
||||
it("adopts a published result after a crash before its fence-row update", async () => {
|
||||
const workspacePath = path.join(root, "published-unrecorded-result");
|
||||
const originalHarness = createHarness(placementStore, { workspacePath });
|
||||
const active = originalHarness.placements.seedActive(2);
|
||||
if (active.state !== "active") {
|
||||
throw new Error("active placement fixture was not active");
|
||||
}
|
||||
const claim = placementStore.claimTurn({
|
||||
...REQUEST,
|
||||
claimId: "published-unrecorded-claim",
|
||||
runId: "published-unrecorded-run",
|
||||
owner: {
|
||||
kind: "worker",
|
||||
environmentId: active.environmentId,
|
||||
ownerEpoch: active.activeOwnerEpoch,
|
||||
},
|
||||
});
|
||||
const staged = await stagePendingResult({
|
||||
store: placementStore,
|
||||
claim,
|
||||
workspacePath,
|
||||
base: "base\n",
|
||||
current: "worker\n",
|
||||
record: false,
|
||||
});
|
||||
const restartedStore = createWorkerSessionPlacementStore({ database, now: () => 2_000 });
|
||||
const restartedHarness = createHarness(restartedStore, { workspacePath });
|
||||
restartedHarness.markEnvironmentDestroyed();
|
||||
|
||||
await restartedHarness.service.reconcile();
|
||||
|
||||
await expect(fs.readFile(path.join(workspacePath, "result.txt"), "utf8")).resolves.toBe(
|
||||
"worker\n",
|
||||
);
|
||||
expect(restartedHarness.placements.current()).toMatchObject({
|
||||
state: "reclaimed",
|
||||
turnClaim: null,
|
||||
workspaceBaseManifestRef: staged.currentManifestRef,
|
||||
});
|
||||
expect(restartedStore.listPendingWorkspaceResults()).toEqual([]);
|
||||
});
|
||||
|
||||
it("keeps a staged fence and placement alive when local recovery diverges", async () => {
|
||||
const workspacePath = path.join(root, "diverged-staged-result");
|
||||
const originalHarness = createHarness(placementStore, { workspacePath });
|
||||
const active = originalHarness.placements.seedActive(2);
|
||||
if (active.state !== "active") {
|
||||
throw new Error("active placement fixture was not active");
|
||||
}
|
||||
const claim = placementStore.claimTurn({
|
||||
...REQUEST,
|
||||
claimId: "diverged-staged-claim",
|
||||
runId: "diverged-staged-run",
|
||||
owner: {
|
||||
kind: "worker",
|
||||
environmentId: active.environmentId,
|
||||
ownerEpoch: active.activeOwnerEpoch,
|
||||
},
|
||||
});
|
||||
const staged = await stagePendingResult({
|
||||
store: placementStore,
|
||||
claim,
|
||||
workspacePath,
|
||||
base: "base\n",
|
||||
current: "worker\n",
|
||||
});
|
||||
await fs.writeFile(path.join(workspacePath, "result.txt"), "local divergence\n");
|
||||
const restartedStore = createWorkerSessionPlacementStore({ database, now: () => 2_000 });
|
||||
const restartedHarness = createHarness(restartedStore, { workspacePath });
|
||||
restartedHarness.markEnvironmentDestroyed();
|
||||
|
||||
await restartedHarness.service.reconcile();
|
||||
|
||||
expect(restartedHarness.placements.current()).toMatchObject({
|
||||
state: "active",
|
||||
turnClaim: { claimId: claim.claimId },
|
||||
});
|
||||
expect(restartedStore.listPendingWorkspaceResults()).toMatchObject([
|
||||
{ stagedResultRef: staged.stagedResultRef, workspaceAcceptedAtMs: null },
|
||||
]);
|
||||
expect(restartedHarness.environments.startTunnel).not.toHaveBeenCalled();
|
||||
expect(restartedHarness.log).not.toContain("placement:failed");
|
||||
await expect(fs.readFile(path.join(workspacePath, "result.txt"), "utf8")).resolves.toBe(
|
||||
"local divergence\n",
|
||||
);
|
||||
expect(
|
||||
await runCommandWithTimeout(
|
||||
["git", "-C", workspacePath, "show-ref", "--verify", staged.stagedResultRef],
|
||||
{ timeoutMs: 10_000 },
|
||||
),
|
||||
).toMatchObject({ code: 0 });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,284 @@
|
||||
import { vi } from "vitest";
|
||||
import type { MintedWorkerCredential } from "./credential.js";
|
||||
import type {
|
||||
WorkerDispatchEnvironmentService,
|
||||
WorkerDispatchPlacementStore,
|
||||
} from "./placement-dispatch-failure.js";
|
||||
import {
|
||||
BUNDLE_HASH,
|
||||
createDispatchEnvironmentFixtures,
|
||||
type DispatchStage,
|
||||
MANIFEST_REF,
|
||||
type PlacementStore,
|
||||
REQUEST,
|
||||
seedActivePlacement,
|
||||
seedStartingPlacement,
|
||||
} from "./placement-dispatch-test-fixtures.js";
|
||||
import { createWorkerPlacementDispatchService } from "./placement-dispatch.js";
|
||||
import type { WorkerTunnelHandle } from "./tunnel.js";
|
||||
import { createWorkerWorkspaceOperationCoordinator } from "./workspace-operation-coordinator.js";
|
||||
|
||||
export function createHarness(
|
||||
placementStore: PlacementStore,
|
||||
options: {
|
||||
failAt?: DispatchStage;
|
||||
destroyFails?: boolean;
|
||||
claimOnDrain?: boolean;
|
||||
reconcileFails?: boolean;
|
||||
verifyFails?: boolean;
|
||||
leaseFails?: boolean;
|
||||
localVerifyFails?: boolean;
|
||||
resumeFails?: boolean;
|
||||
workspacePath?: string;
|
||||
} = {},
|
||||
) {
|
||||
const reconciledManifestRef = MANIFEST_REF.replaceAll("b", "c");
|
||||
const log: string[] = [];
|
||||
const fail = (stage: DispatchStage) => {
|
||||
log.push(stage);
|
||||
if (options.failAt === stage) {
|
||||
throw new Error(`${stage} failed`);
|
||||
}
|
||||
};
|
||||
const placements: WorkerDispatchPlacementStore = {
|
||||
get: (sessionId) => placementStore.get(sessionId),
|
||||
loadWorkspaceReconciliation: (owner) => placementStore.loadWorkspaceReconciliation(owner),
|
||||
beginWorkspaceReconciliation: (owner, journal) =>
|
||||
placementStore.beginWorkspaceReconciliation(owner, journal),
|
||||
abortWorkspaceReconciliation: (owner) => placementStore.abortWorkspaceReconciliation(owner),
|
||||
listWorkspaceReconciliationOwners: () => placementStore.listWorkspaceReconciliationOwners(),
|
||||
listPendingWorkspaceResults: () => placementStore.listPendingWorkspaceResults(),
|
||||
workspaceResultInstanceId: () => placementStore.workspaceResultInstanceId(),
|
||||
recordStagedWorkspaceResult: (claim, ref) =>
|
||||
placementStore.recordStagedWorkspaceResult(claim, ref),
|
||||
acceptWorkspaceResult: (claim) => placementStore.acceptWorkspaceResult(claim),
|
||||
completeWorkspaceResultAndReleaseTurn: (claim, completionOptions) =>
|
||||
placementStore.completeWorkspaceResultAndReleaseTurn(claim, completionOptions),
|
||||
abandonWorkspaceResult: (pending) => placementStore.abandonWorkspaceResult(pending),
|
||||
releaseTurn: (claim) => placementStore.releaseTurn(claim),
|
||||
updateWorkspaceBaseManifest: (params) => placementStore.updateWorkspaceBaseManifest(params),
|
||||
acceptIdleWorkspaceReconciliation: (params) =>
|
||||
placementStore.acceptIdleWorkspaceReconciliation(params),
|
||||
startDispatch: (params) => {
|
||||
log.push("placement:requested");
|
||||
return placementStore.startDispatch(params);
|
||||
},
|
||||
transition: (params) => {
|
||||
log.push(`placement:${params.to}`);
|
||||
return placementStore.transition(params);
|
||||
},
|
||||
fail: (params) => {
|
||||
log.push("placement:failed");
|
||||
return placementStore.fail(params);
|
||||
},
|
||||
finishReclaim: (params) => {
|
||||
log.push("placement:reclaimed");
|
||||
if (options.claimOnDrain && !placementStore.get(params.sessionId)?.turnClaim) {
|
||||
placementStore.claimTurn({
|
||||
sessionId: params.sessionId,
|
||||
sessionKey: REQUEST.sessionKey,
|
||||
agentId: REQUEST.agentId,
|
||||
claimId: "claim-on-drain",
|
||||
runId: "run-on-drain",
|
||||
owner: {
|
||||
kind: "worker",
|
||||
environmentId: params.environmentId,
|
||||
ownerEpoch: params.ownerEpoch,
|
||||
},
|
||||
});
|
||||
}
|
||||
return placementStore.finishReclaim(params);
|
||||
},
|
||||
listForReconcile: () => placementStore.listForReconcile(),
|
||||
startDrain: (params) => {
|
||||
log.push("placement:draining");
|
||||
if (options.claimOnDrain && !placementStore.get(params.sessionId)?.turnClaim) {
|
||||
placementStore.claimTurn({
|
||||
sessionId: params.sessionId,
|
||||
sessionKey: REQUEST.sessionKey,
|
||||
agentId: REQUEST.agentId,
|
||||
claimId: "claim-on-drain",
|
||||
runId: "run-on-drain",
|
||||
owner: {
|
||||
kind: "worker",
|
||||
environmentId: params.environmentId,
|
||||
ownerEpoch: params.ownerEpoch,
|
||||
},
|
||||
});
|
||||
}
|
||||
return placementStore.startDrain(params);
|
||||
},
|
||||
startReconcile: (params) => {
|
||||
log.push("placement:reconciling");
|
||||
return placementStore.startReconcile(params);
|
||||
},
|
||||
adoptActive: (params) => {
|
||||
log.push("placement:adopted");
|
||||
return placementStore.adoptActive(params);
|
||||
},
|
||||
};
|
||||
const { attached, destroyedEnvironment, environmentId, ready } =
|
||||
createDispatchEnvironmentFixtures();
|
||||
let currentEnvironment: ReturnType<WorkerDispatchEnvironmentService["get"]> = ready;
|
||||
const tunnelHandle = (ownerEpoch: number): WorkerTunnelHandle => ({
|
||||
environmentId: ready.environmentId,
|
||||
ownerEpoch,
|
||||
remoteSocketPath: "/worker/gateway.sock",
|
||||
quiesceWorkspace: vi.fn(async () => {
|
||||
log.push("workspace:quiesce");
|
||||
return {
|
||||
assertActive: vi.fn(async () => {
|
||||
log.push("workspace:lease");
|
||||
if (options.leaseFails) {
|
||||
throw new Error("workspace quiescence expired");
|
||||
}
|
||||
}),
|
||||
resume: vi.fn(async () => {
|
||||
log.push("workspace:resume");
|
||||
if (options.resumeFails) {
|
||||
throw new Error("workspace resume failed");
|
||||
}
|
||||
}),
|
||||
};
|
||||
}),
|
||||
reconcileWorkspace: vi.fn(async (request) => {
|
||||
log.push("workspace:reconcile");
|
||||
if (options.reconcileFails) {
|
||||
throw new Error("workspace conflict");
|
||||
}
|
||||
request.journal.commit(reconciledManifestRef);
|
||||
return {
|
||||
manifestRef: reconciledManifestRef,
|
||||
changed: true,
|
||||
verifyStable: async () => {
|
||||
log.push("workspace:verify");
|
||||
if (options.verifyFails) {
|
||||
throw new Error("workspace changed after reconciliation");
|
||||
}
|
||||
},
|
||||
verifyLocalStable: async () => {
|
||||
log.push("workspace:verify-local");
|
||||
if (options.localVerifyFails) {
|
||||
throw new Error("local workspace changed after reconciliation");
|
||||
}
|
||||
},
|
||||
};
|
||||
}),
|
||||
runWorkspaceCommand: vi.fn(async () => ({
|
||||
stdout: "",
|
||||
stderr: "",
|
||||
code: 0,
|
||||
signal: null,
|
||||
killed: false,
|
||||
termination: "exit" as const,
|
||||
})),
|
||||
syncWorkspace: vi.fn(async () => {
|
||||
fail("sync");
|
||||
return {
|
||||
mode: "git" as const,
|
||||
remoteWorkspaceDir: "/worker/workspace",
|
||||
manifestRef: MANIFEST_REF,
|
||||
};
|
||||
}),
|
||||
stop: vi.fn(async () => {}),
|
||||
});
|
||||
const minted: MintedWorkerCredential = {
|
||||
credential: "fixture-credential",
|
||||
deliveryId: "fixture-delivery-id",
|
||||
environmentId: ready.environmentId,
|
||||
bundleHash: BUNDLE_HASH,
|
||||
sessionId: REQUEST.sessionId,
|
||||
rpcSetVersion: 1,
|
||||
ownerEpoch: 2,
|
||||
expiresAtMs: 10_000,
|
||||
};
|
||||
const environments: WorkerDispatchEnvironmentService = {
|
||||
create: vi.fn(async () => {
|
||||
fail("create");
|
||||
return ready;
|
||||
}),
|
||||
get: vi.fn(() => currentEnvironment),
|
||||
attachSession: vi.fn(async () => {
|
||||
fail("attach");
|
||||
currentEnvironment = attached;
|
||||
return minted;
|
||||
}),
|
||||
startTunnel: vi.fn(async ({ ownerEpoch }) => {
|
||||
fail(ownerEpoch === 1 ? "tunnel:ready" : "tunnel:attached");
|
||||
return tunnelHandle(ownerEpoch);
|
||||
}),
|
||||
stopTunnel: vi.fn(async () => {
|
||||
log.push("teardown:stop");
|
||||
}),
|
||||
destroy: vi.fn(async () => {
|
||||
log.push("teardown:destroy");
|
||||
if (options.destroyFails) {
|
||||
throw new Error("destroy pending");
|
||||
}
|
||||
const destroyed = destroyedEnvironment((currentEnvironment?.ownerEpoch ?? 1) + 1);
|
||||
currentEnvironment = destroyed;
|
||||
return destroyed;
|
||||
}),
|
||||
reconcileOnce: vi.fn(async () => {
|
||||
log.push("environment:reconcile");
|
||||
}),
|
||||
};
|
||||
const service = createWorkerPlacementDispatchService({
|
||||
placements,
|
||||
environments,
|
||||
workspaceOperations: createWorkerWorkspaceOperationCoordinator(),
|
||||
runLocalBarrier: async ({ startDispatch }) => {
|
||||
log.push("barrier");
|
||||
const placement = startDispatch();
|
||||
if (options.failAt === "barrier") {
|
||||
throw new Error("barrier failed");
|
||||
}
|
||||
return placement;
|
||||
},
|
||||
runActivationBarrier: async ({ activate }) => {
|
||||
fail("activation");
|
||||
return activate();
|
||||
},
|
||||
runReclaimBarrier: async ({ reclaim }) =>
|
||||
await reclaim(options.workspacePath ?? "/gateway/workspace"),
|
||||
resolveWorkspacePath: async () => {
|
||||
fail("workspace");
|
||||
return options.workspacePath ?? "/gateway/workspace";
|
||||
},
|
||||
});
|
||||
return {
|
||||
log,
|
||||
reconciledManifestRef,
|
||||
placements: {
|
||||
current: () => placementStore.get(REQUEST.sessionId),
|
||||
seedStarting: () => seedStartingPlacement(placementStore, environmentId),
|
||||
seedActive: (ownerEpoch: number) =>
|
||||
seedActivePlacement(placementStore, { environmentId, ownerEpoch }),
|
||||
seedDraining: (ownerEpoch: number) => {
|
||||
const active = seedActivePlacement(placementStore, { environmentId, ownerEpoch });
|
||||
if (active.state !== "active") {
|
||||
throw new Error("active placement fixture was not active");
|
||||
}
|
||||
return placementStore.startDrain({
|
||||
sessionId: active.sessionId,
|
||||
environmentId: active.environmentId,
|
||||
ownerEpoch: active.activeOwnerEpoch,
|
||||
expectedGeneration: active.generation,
|
||||
});
|
||||
},
|
||||
},
|
||||
environments,
|
||||
markEnvironmentDestroyed: () => {
|
||||
currentEnvironment = destroyedEnvironment((currentEnvironment?.ownerEpoch ?? 1) + 1);
|
||||
},
|
||||
markEnvironmentOwnerEpoch: (ownerEpoch: number) => {
|
||||
currentEnvironment = { ...attached, ownerEpoch };
|
||||
},
|
||||
markEnvironmentAttachments: (attachedSessionIds: string[]) => {
|
||||
currentEnvironment = { ...attached, attachedSessionIds };
|
||||
},
|
||||
service,
|
||||
ready,
|
||||
attached,
|
||||
};
|
||||
}
|
||||
@@ -58,6 +58,8 @@ function createHarness(
|
||||
listWorkspaceReconciliationOwners: () => placementStore.listWorkspaceReconciliationOwners(),
|
||||
listPendingWorkspaceResults: () => placementStore.listPendingWorkspaceResults(),
|
||||
workspaceResultInstanceId: () => placementStore.workspaceResultInstanceId(),
|
||||
recordStagedWorkspaceResult: (claim, ref) =>
|
||||
placementStore.recordStagedWorkspaceResult(claim, ref),
|
||||
acceptWorkspaceResult: (claim) => placementStore.acceptWorkspaceResult(claim),
|
||||
completeWorkspaceResultAndReleaseTurn: (claim, completionOptions) =>
|
||||
placementStore.completeWorkspaceResultAndReleaseTurn(claim, completionOptions),
|
||||
|
||||
@@ -57,4 +57,39 @@ describe("forced worker environment abandonment", () => {
|
||||
});
|
||||
expect(store.listPendingWorkspaceResults()).toEqual([]);
|
||||
});
|
||||
|
||||
it("releases a pending worker claim when its workspace is already gone", async () => {
|
||||
const store = createWorkerSessionPlacementStore({ database, now: () => 1_000 });
|
||||
const { environmentId } = createDispatchEnvironmentFixtures();
|
||||
const active = seedActivePlacement(store, { environmentId, ownerEpoch: 2 });
|
||||
if (active.state !== "active") {
|
||||
throw new Error("active placement fixture was not active");
|
||||
}
|
||||
const claim = store.claimTurn({
|
||||
...REQUEST,
|
||||
claimId: "forced-missing-workspace-claim",
|
||||
runId: "forced-missing-workspace-run",
|
||||
owner: { kind: "worker", environmentId, ownerEpoch: 2 },
|
||||
});
|
||||
store.markWorkspaceResultPending(claim);
|
||||
store.recordStagedWorkspaceResult(
|
||||
claim,
|
||||
"refs/openclaw/worker-results/forced-missing-workspace-claim",
|
||||
);
|
||||
|
||||
await forceAbandonWorkerEnvironment({
|
||||
placements: store,
|
||||
environmentId,
|
||||
resolveWorkspacePath: async () => {
|
||||
throw new Error("session-owned managed worktree is missing");
|
||||
},
|
||||
});
|
||||
|
||||
expect(store.get(REQUEST.sessionId)).toMatchObject({
|
||||
state: "failed",
|
||||
turnClaim: null,
|
||||
recoveryError: "Cloud worker result abandoned by forced operator teardown",
|
||||
});
|
||||
expect(store.listPendingWorkspaceResults()).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,5 +1,28 @@
|
||||
import type { WorkerDispatchPlacementStore } from "./placement-dispatch-failure.js";
|
||||
import { recoverWorkerWorkspaceReconciliation } from "./workspace-reconcile.js";
|
||||
import {
|
||||
deleteStagedWorkerWorkspaceResult,
|
||||
hasWorkerWorkspaceResultRef,
|
||||
preparedWorkerWorkspaceResultRef,
|
||||
workerWorkspaceResultRef,
|
||||
} from "./workspace-result-staging.js";
|
||||
|
||||
async function tryResolveWorkspacePath(
|
||||
resolveWorkspacePath: (placement: {
|
||||
sessionId: string;
|
||||
sessionKey: string;
|
||||
agentId: string;
|
||||
}) => Promise<string>,
|
||||
placement: { sessionId: string; sessionKey: string; agentId: string },
|
||||
): Promise<string | undefined> {
|
||||
try {
|
||||
return await resolveWorkspacePath(placement);
|
||||
} catch {
|
||||
// Forced teardown is the last-resort state owner. If the session/worktree is
|
||||
// already gone, skip local repair/ref cleanup and still release the claim.
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
export async function forceAbandonWorkerEnvironment(params: {
|
||||
placements: WorkerDispatchPlacementStore;
|
||||
@@ -27,13 +50,34 @@ export async function forceAbandonWorkerEnvironment(params: {
|
||||
}
|
||||
const journal = placements.loadWorkspaceReconciliation(owner);
|
||||
if (journal) {
|
||||
const root = await params.resolveWorkspacePath(placement);
|
||||
await recoverWorkerWorkspaceReconciliation({ root, journal });
|
||||
const root = await tryResolveWorkspacePath(params.resolveWorkspacePath, placement);
|
||||
if (root) {
|
||||
await recoverWorkerWorkspaceReconciliation({ root, journal });
|
||||
}
|
||||
placements.abortWorkspaceReconciliation(owner);
|
||||
}
|
||||
}
|
||||
for (const pending of placements.listPendingWorkspaceResults()) {
|
||||
if (pending.environmentId === environmentId) {
|
||||
const placement = placements.get(pending.sessionId);
|
||||
if (!placement) {
|
||||
if (pending.stagedResultRef) {
|
||||
throw new Error(
|
||||
`Forced teardown found a staged result without a placement: ${pending.sessionId}`,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
const root = await tryResolveWorkspacePath(params.resolveWorkspacePath, placement);
|
||||
if (root) {
|
||||
const finalRef = pending.stagedResultRef ?? workerWorkspaceResultRef(pending.claimId);
|
||||
const refs = [finalRef, preparedWorkerWorkspaceResultRef(finalRef)];
|
||||
for (const stagedResultRef of refs) {
|
||||
if (await hasWorkerWorkspaceResultRef({ root, stagedResultRef })) {
|
||||
await deleteStagedWorkerWorkspaceResult({ root, stagedResultRef });
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
placements.abandonWorkspaceResult(pending);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -743,11 +743,20 @@ describe("worker session placement store", () => {
|
||||
gatewayInstanceId: store.workspaceResultInstanceId(),
|
||||
recoveryRequestedAtMs: null,
|
||||
workspaceAcceptedAtMs: null,
|
||||
stagedResultRef: null,
|
||||
},
|
||||
]);
|
||||
expect(() => store.releaseTurn(claim)).toThrow("pending cloud workspace result");
|
||||
|
||||
const manifestRef = `sha256:${"f".repeat(64)}`;
|
||||
const stagedResultRef = `refs/openclaw/worker-results/${claim.claimId}`;
|
||||
expect(() =>
|
||||
store.recordStagedWorkspaceResult(claim, "refs/openclaw/worker-results/unsafe.claim"),
|
||||
).toThrow("Worker workspace staged result reference is invalid");
|
||||
store.recordStagedWorkspaceResult(claim, stagedResultRef);
|
||||
expect(store.listPendingWorkspaceResults()).toMatchObject([
|
||||
{ sessionId: active.sessionId, stagedResultRef },
|
||||
]);
|
||||
store.updateWorkspaceBaseManifest({ claim, manifestRef });
|
||||
expect(store.listPendingWorkspaceResults()).toMatchObject([
|
||||
{ sessionId: active.sessionId, workspaceAcceptedAtMs: null },
|
||||
|
||||
@@ -22,6 +22,7 @@ type WorkerWorkspacePendingResult = {
|
||||
gatewayInstanceId: string;
|
||||
recoveryRequestedAtMs: number | null;
|
||||
workspaceAcceptedAtMs: number | null;
|
||||
stagedResultRef: string | null;
|
||||
};
|
||||
|
||||
export function clearWorkerWorkspacePendingResult(db: DatabaseSync, sessionId: string): void {
|
||||
@@ -96,6 +97,7 @@ export function insertWorkerWorkspacePendingResult(
|
||||
gateway_instance_id: gatewayInstanceId,
|
||||
recovery_requested_at_ms: null,
|
||||
workspace_accepted_at_ms: null,
|
||||
staged_result_ref: null,
|
||||
created_at_ms: nowMs,
|
||||
})
|
||||
.onConflict((conflict) => conflict.column("session_id").doNothing()),
|
||||
@@ -191,6 +193,7 @@ export function createPlacementWorkspaceResultOps(runtime: PlacementStoreRuntime
|
||||
"gateway_instance_id",
|
||||
"recovery_requested_at_ms",
|
||||
"workspace_accepted_at_ms",
|
||||
"staged_result_ref",
|
||||
])
|
||||
.orderBy("session_id"),
|
||||
).rows.map((row) => ({
|
||||
@@ -203,6 +206,7 @@ export function createPlacementWorkspaceResultOps(runtime: PlacementStoreRuntime
|
||||
gatewayInstanceId: row.gateway_instance_id,
|
||||
recoveryRequestedAtMs: row.recovery_requested_at_ms,
|
||||
workspaceAcceptedAtMs: row.workspace_accepted_at_ms,
|
||||
stagedResultRef: row.staged_result_ref,
|
||||
}));
|
||||
},
|
||||
|
||||
@@ -212,6 +216,33 @@ export function createPlacementWorkspaceResultOps(runtime: PlacementStoreRuntime
|
||||
});
|
||||
},
|
||||
|
||||
recordStagedWorkspaceResult(claim: WorkerSessionTurnClaim, stagedResultRef: string): void {
|
||||
if (!/^refs\/openclaw\/worker-results\/[A-Za-z0-9-]+$/u.test(stagedResultRef)) {
|
||||
throw new Error("Worker workspace staged result reference is invalid");
|
||||
}
|
||||
write((db) => {
|
||||
const pending = assertPendingClaim(db, claim);
|
||||
if (pending.workspace_accepted_at_ms !== null) {
|
||||
throw new Error(`Cannot restage accepted worker workspace result for ${claim.sessionId}`);
|
||||
}
|
||||
if (pending.staged_result_ref && pending.staged_result_ref !== stagedResultRef) {
|
||||
throw new Error(`Worker workspace result ref changed for ${claim.sessionId}`);
|
||||
}
|
||||
const result = executeSqliteQuerySync(
|
||||
db,
|
||||
query(db)
|
||||
.updateTable("worker_workspace_pending_results")
|
||||
.set({ staged_result_ref: stagedResultRef })
|
||||
.where("session_id", "=", claim.sessionId)
|
||||
.where("claim_id", "=", claim.claimId)
|
||||
.where("run_id", "=", claim.runId),
|
||||
);
|
||||
if (result.numAffectedRows !== 1n) {
|
||||
throw new Error(`Cannot stage stale worker workspace result for ${claim.sessionId}`);
|
||||
}
|
||||
});
|
||||
},
|
||||
|
||||
acceptWorkspaceResult(claim: WorkerSessionTurnClaim): void {
|
||||
write((db) => {
|
||||
assertPendingClaim(db, claim);
|
||||
|
||||
@@ -32,6 +32,10 @@ export type WorkerWorkspaceReconcileRequest = {
|
||||
remoteWorkspaceDir: string;
|
||||
baseManifestRef: string;
|
||||
journal: WorkerWorkspaceReconciliationJournalAdapter;
|
||||
stagedResult?: {
|
||||
ref: string;
|
||||
record(ref: string): void;
|
||||
};
|
||||
};
|
||||
|
||||
export type WorkerWorkspaceReconcileResult = {
|
||||
@@ -41,6 +45,11 @@ export type WorkerWorkspaceReconcileResult = {
|
||||
verifyStable(): Promise<void>;
|
||||
/** Re-read the accepted local result after the remote stability fence. */
|
||||
verifyLocalStable(): Promise<void>;
|
||||
/** Apply the prepared candidate locally without making it restart-authoritative. */
|
||||
applyPreparedStagedResult?(): Promise<void>;
|
||||
/** Publish the verified candidate for restart recovery. */
|
||||
publishStagedResult?(): Promise<void>;
|
||||
discardPreparedStagedResult?(): Promise<void>;
|
||||
};
|
||||
|
||||
export type WorkerWorkspaceQuiescence = {
|
||||
|
||||
@@ -395,6 +395,100 @@ describe("worker tunnel manager", () => {
|
||||
await handle.stop();
|
||||
});
|
||||
|
||||
it("does not downgrade an operational HEAD probe failure to plain sync", async () => {
|
||||
const remoteWorkspaceDir = "/home/worker/.openclaw-worker/workspaces/env/session/3";
|
||||
const localPath = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-worker-head-probe-"));
|
||||
await fs.mkdir(path.join(localPath, ".git"));
|
||||
const fake = fakeRunner((argv, options) => {
|
||||
if (argv.includes("--show-toplevel")) {
|
||||
return success(`${localPath}\n`);
|
||||
}
|
||||
if (argv.includes("--verify")) {
|
||||
return {
|
||||
...success("", "HEAD probe timed out"),
|
||||
code: null,
|
||||
killed: true,
|
||||
termination: "timeout",
|
||||
};
|
||||
}
|
||||
if (
|
||||
typeof options.input === "string" &&
|
||||
options.input.includes("unsafe worker workspace directory")
|
||||
) {
|
||||
return success(`${remoteWorkspaceDir}\n`);
|
||||
}
|
||||
return undefined;
|
||||
});
|
||||
const manager = createWorkerTunnelManager({ runner: fake.runner });
|
||||
const starting = manager.start({
|
||||
environmentId: "worker:head-probe-failure",
|
||||
ownerEpoch: 3,
|
||||
ssh: SSH,
|
||||
gateway: { host: "127.0.0.1", port: 18789 },
|
||||
resolveIdentity,
|
||||
});
|
||||
await waitForStarts(fake.starts, 1);
|
||||
fake.starts[0]?.process.becomeReady();
|
||||
const handle = await starting;
|
||||
|
||||
try {
|
||||
await expect(
|
||||
handle.syncWorkspace({ localPath, sessionId: "session:three", generation: 3 }),
|
||||
).rejects.toThrow("Worker workspace sync failed: HEAD probe timed out");
|
||||
expect(fake.runs.some((entry) => entry.argv[0] === "rsync")).toBe(false);
|
||||
} finally {
|
||||
await handle.stop();
|
||||
await fs.rm(localPath, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("does not downgrade an operational repository-root probe failure to plain sync", async () => {
|
||||
const remoteWorkspaceDir = "/home/worker/.openclaw-worker/workspaces/env/session/4";
|
||||
const localPath = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-worker-root-probe-"));
|
||||
await fs.mkdir(path.join(localPath, ".git"));
|
||||
const fake = fakeRunner((argv, options) => {
|
||||
if (argv.includes("--show-toplevel")) {
|
||||
return {
|
||||
...success("", "root probe timed out"),
|
||||
code: null,
|
||||
killed: true,
|
||||
termination: "timeout",
|
||||
};
|
||||
}
|
||||
if (argv.includes("--verify")) {
|
||||
return success("0123456789abcdef0123456789abcdef01234567\n");
|
||||
}
|
||||
if (
|
||||
typeof options.input === "string" &&
|
||||
options.input.includes("unsafe worker workspace directory")
|
||||
) {
|
||||
return success(`${remoteWorkspaceDir}\n`);
|
||||
}
|
||||
return undefined;
|
||||
});
|
||||
const manager = createWorkerTunnelManager({ runner: fake.runner });
|
||||
const starting = manager.start({
|
||||
environmentId: "worker:root-probe-failure",
|
||||
ownerEpoch: 4,
|
||||
ssh: SSH,
|
||||
gateway: { host: "127.0.0.1", port: 18789 },
|
||||
resolveIdentity,
|
||||
});
|
||||
await waitForStarts(fake.starts, 1);
|
||||
fake.starts[0]?.process.becomeReady();
|
||||
const handle = await starting;
|
||||
|
||||
try {
|
||||
await expect(
|
||||
handle.syncWorkspace({ localPath, sessionId: "session:four", generation: 4 }),
|
||||
).rejects.toThrow("Worker workspace sync failed: root probe timed out");
|
||||
expect(fake.runs.some((entry) => entry.argv[0] === "rsync")).toBe(false);
|
||||
} finally {
|
||||
await handle.stop();
|
||||
await fs.rm(localPath, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("materializes a large dirty git workspace as a credential-free commit-capable clone", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-worker-git-sync-"));
|
||||
const localPath = path.join(root, "local");
|
||||
@@ -607,6 +701,9 @@ describe("worker tunnel manager", () => {
|
||||
fs.writeFile(path.join(plainPath, "hello.txt"), "plain\n"),
|
||||
fs.writeFile(path.join(plainPath, "nested/.git/config"), "private metadata\n"),
|
||||
]);
|
||||
// Result staging stores refs in an unborn repository for a plain workspace.
|
||||
// A later dispatch must keep using plain-mode sync until the user creates HEAD.
|
||||
await git(plainPath, "init");
|
||||
await fs.mkdir(path.join(plainPath, "__pycache__"));
|
||||
await Promise.all([
|
||||
fs.writeFile(path.join(plainPath, "__pycache__/fizzbuzz.pyc"), "derived\n"),
|
||||
|
||||
@@ -8,7 +8,7 @@ import {
|
||||
makeAgentAssistantMessage,
|
||||
makeAgentUserMessage,
|
||||
} from "../../agents/test-helpers/agent-message-fixtures.js";
|
||||
import type { SpawnResult } from "../../process/exec.js";
|
||||
import { runCommandWithTimeout, type SpawnResult } from "../../process/exec.js";
|
||||
import { createDeferred } from "../../shared/deferred.js";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
@@ -471,6 +471,10 @@ describe("worker turn launcher", () => {
|
||||
);
|
||||
|
||||
it("launches the active worker with projected history and releases its claim", async () => {
|
||||
const initialized = await runCommandWithTimeout(["git", "-C", root, "init", "--quiet"], {
|
||||
timeoutMs: 10_000,
|
||||
});
|
||||
expect(initialized.code).toBe(0);
|
||||
seedActivePlacement();
|
||||
const manager = SessionManager.open(sessionFile);
|
||||
const earlierRequestId = manager.appendMessage(
|
||||
@@ -557,6 +561,11 @@ describe("worker turn launcher", () => {
|
||||
throw new Error("unexpected workspace sync");
|
||||
}),
|
||||
reconcileWorkspace: vi.fn(async (request) => {
|
||||
expect(request.stagedResult).toBeDefined();
|
||||
request.stagedResult!.record(request.stagedResult!.ref);
|
||||
expect(placements.listPendingWorkspaceResults()).toMatchObject([
|
||||
{ stagedResultRef: request.stagedResult!.ref, workspaceAcceptedAtMs: null },
|
||||
]);
|
||||
request.journal.commit(MANIFEST_REF);
|
||||
return {
|
||||
manifestRef: MANIFEST_REF,
|
||||
|
||||
@@ -31,6 +31,10 @@ import {
|
||||
type WorkerWorkspaceOperationCoordinator,
|
||||
} from "./workspace-operation-coordinator.js";
|
||||
import { recoverWorkerWorkspaceReconciliation } from "./workspace-reconcile.js";
|
||||
import {
|
||||
deleteStagedWorkerWorkspaceResult,
|
||||
workerWorkspaceResultRef,
|
||||
} from "./workspace-result-staging.js";
|
||||
|
||||
const WORKER_LAUNCH_SCRIPT = 'exec node "$HOME/.openclaw-worker/$1/openclaw.mjs" worker';
|
||||
|
||||
@@ -461,17 +465,36 @@ async function executeWorkerTurn(params: {
|
||||
const quiescence = await tunnel.quiesceWorkspace(currentPlacement.remoteWorkspaceDir);
|
||||
let resumed = false;
|
||||
try {
|
||||
const stagedResultRef = workerWorkspaceResultRef(params.turnClaim.claimId);
|
||||
const reconciliation = await tunnel.reconcileWorkspace({
|
||||
localPath: turn.workspaceDir,
|
||||
remoteWorkspaceDir: currentPlacement.remoteWorkspaceDir,
|
||||
baseManifestRef: currentPlacement.workspaceBaseManifestRef,
|
||||
journal,
|
||||
stagedResult: {
|
||||
ref: stagedResultRef,
|
||||
record: (ref) => params.placements.recordStagedWorkspaceResult(params.turnClaim, ref),
|
||||
},
|
||||
});
|
||||
await verifyReconciledWorkspaceFinal(reconciliation, quiescence);
|
||||
if (!manifestAccepted) {
|
||||
throw new Error("Cloud worker workspace reconciliation was not durably accepted");
|
||||
}
|
||||
await verifyReconciledWorkspaceFinal(reconciliation, quiescence);
|
||||
params.placements.acceptWorkspaceResult(params.turnClaim);
|
||||
const recordedStagedResultRef = params.placements
|
||||
.listPendingWorkspaceResults()
|
||||
.find(
|
||||
(pending) =>
|
||||
pending.sessionId === params.turnClaim.sessionId &&
|
||||
pending.claimId === params.turnClaim.claimId &&
|
||||
pending.runId === params.turnClaim.runId,
|
||||
)?.stagedResultRef;
|
||||
if (recordedStagedResultRef) {
|
||||
await deleteStagedWorkerWorkspaceResult({
|
||||
root: turn.workspaceDir,
|
||||
stagedResultRef: recordedStagedResultRef,
|
||||
});
|
||||
}
|
||||
await quiescence.resume();
|
||||
resumed = true;
|
||||
params.placements.completeWorkspaceResultAndReleaseTurn(params.turnClaim);
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { verifyReconciledWorkspaceFinal } from "./workspace-finalize.js";
|
||||
|
||||
describe("final worker workspace fences", () => {
|
||||
@@ -46,4 +46,146 @@ describe("final worker workspace fences", () => {
|
||||
).rejects.toThrow("late remote write");
|
||||
expect(remoteVerifications).toBe(2);
|
||||
});
|
||||
|
||||
it("publishes between remote stability fences under quiescence", async () => {
|
||||
const log: string[] = [];
|
||||
await verifyReconciledWorkspaceFinal(
|
||||
{
|
||||
manifestRef: "sha256:" + "b".repeat(64),
|
||||
changed: true,
|
||||
verifyStable: async () => {
|
||||
log.push("remote");
|
||||
},
|
||||
verifyLocalStable: async () => {
|
||||
log.push("local");
|
||||
},
|
||||
applyPreparedStagedResult: async () => {
|
||||
log.push("apply-prepared");
|
||||
},
|
||||
publishStagedResult: async () => {
|
||||
log.push("publish");
|
||||
},
|
||||
},
|
||||
{
|
||||
assertActive: async () => {
|
||||
log.push("quiescence");
|
||||
},
|
||||
resume: async () => {},
|
||||
},
|
||||
);
|
||||
expect(log).toEqual([
|
||||
"remote",
|
||||
"quiescence",
|
||||
"remote",
|
||||
"apply-prepared",
|
||||
"local",
|
||||
"quiescence",
|
||||
"remote",
|
||||
"local",
|
||||
"publish",
|
||||
]);
|
||||
});
|
||||
|
||||
it("rejects quiescence lost while the staged result is finalized", async () => {
|
||||
const log: string[] = [];
|
||||
let quiescenceChecks = 0;
|
||||
await expect(
|
||||
verifyReconciledWorkspaceFinal(
|
||||
{
|
||||
manifestRef: "sha256:" + "c".repeat(64),
|
||||
changed: true,
|
||||
verifyStable: async () => {
|
||||
log.push("remote");
|
||||
},
|
||||
verifyLocalStable: async () => {
|
||||
log.push("local");
|
||||
},
|
||||
applyPreparedStagedResult: async () => {
|
||||
log.push("apply-prepared");
|
||||
},
|
||||
publishStagedResult: async () => {
|
||||
log.push("publish");
|
||||
},
|
||||
discardPreparedStagedResult: async () => {
|
||||
log.push("discard-prepared");
|
||||
},
|
||||
},
|
||||
{
|
||||
assertActive: async () => {
|
||||
quiescenceChecks += 1;
|
||||
log.push("quiescence");
|
||||
if (quiescenceChecks === 2) {
|
||||
throw new Error("quiescence expired during finalization");
|
||||
}
|
||||
},
|
||||
resume: async () => {},
|
||||
},
|
||||
),
|
||||
).rejects.toThrow("quiescence expired during finalization");
|
||||
expect(log).toEqual([
|
||||
"remote",
|
||||
"quiescence",
|
||||
"remote",
|
||||
"apply-prepared",
|
||||
"local",
|
||||
"quiescence",
|
||||
"discard-prepared",
|
||||
]);
|
||||
});
|
||||
|
||||
it("discards a prepared result when the final remote fence fails", async () => {
|
||||
const log: string[] = [];
|
||||
let remoteVerifications = 0;
|
||||
await expect(
|
||||
verifyReconciledWorkspaceFinal(
|
||||
{
|
||||
manifestRef: "sha256:" + "c".repeat(64),
|
||||
changed: true,
|
||||
verifyStable: async () => {
|
||||
remoteVerifications += 1;
|
||||
if (remoteVerifications === 3) {
|
||||
throw new Error("late remote write");
|
||||
}
|
||||
},
|
||||
verifyLocalStable: async () => {
|
||||
log.push("local");
|
||||
},
|
||||
applyPreparedStagedResult: async () => {
|
||||
log.push("apply-prepared");
|
||||
},
|
||||
publishStagedResult: async () => {
|
||||
log.push("publish");
|
||||
},
|
||||
discardPreparedStagedResult: async () => {
|
||||
log.push("discard-prepared");
|
||||
},
|
||||
},
|
||||
{ assertActive: async () => {}, resume: async () => {} },
|
||||
),
|
||||
).rejects.toThrow("late remote write");
|
||||
expect(log).toEqual(["apply-prepared", "local", "discard-prepared"]);
|
||||
});
|
||||
|
||||
it("best-effort discards a candidate when staged finalization fails", async () => {
|
||||
const discard = vi.fn(async () => {
|
||||
throw new Error("candidate cleanup failed");
|
||||
});
|
||||
await expect(
|
||||
verifyReconciledWorkspaceFinal(
|
||||
{
|
||||
manifestRef: "sha256:" + "d".repeat(64),
|
||||
changed: true,
|
||||
verifyStable: async () => {},
|
||||
verifyLocalStable: async () => {},
|
||||
applyPreparedStagedResult: async () => {},
|
||||
publishStagedResult: async () => {
|
||||
throw new Error("publish failed");
|
||||
},
|
||||
discardPreparedStagedResult: discard,
|
||||
},
|
||||
{ assertActive: async () => {}, resume: async () => {} },
|
||||
),
|
||||
).rejects.toThrow("publish failed");
|
||||
expect(discard).toHaveBeenCalledOnce();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -8,6 +8,25 @@ export async function verifyReconciledWorkspaceFinal(
|
||||
reconciliation: WorkerWorkspaceReconcileResult,
|
||||
quiescence: WorkerWorkspaceQuiescence,
|
||||
): Promise<void> {
|
||||
if (reconciliation.applyPreparedStagedResult && reconciliation.publishStagedResult) {
|
||||
try {
|
||||
await reconciliation.verifyStable();
|
||||
await quiescence.assertActive();
|
||||
await reconciliation.verifyStable();
|
||||
await reconciliation.applyPreparedStagedResult();
|
||||
await reconciliation.verifyLocalStable();
|
||||
// Applying can outlive the lease renewed above. Only publish the candidate
|
||||
// after both owners pass a fresh fence, so restart recovery cannot adopt it early.
|
||||
await quiescence.assertActive();
|
||||
await reconciliation.verifyStable();
|
||||
await reconciliation.verifyLocalStable();
|
||||
await reconciliation.publishStagedResult();
|
||||
} catch (error) {
|
||||
await reconciliation.discardPreparedStagedResult?.().catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
return;
|
||||
}
|
||||
await reconciliation.verifyStable();
|
||||
await reconciliation.verifyLocalStable();
|
||||
await quiescence.assertActive();
|
||||
|
||||
@@ -15,8 +15,19 @@ import {
|
||||
parseWorkerWorkspaceManifest,
|
||||
recoverWorkerWorkspaceReconciliation,
|
||||
type WorkerWorkspaceReconciliationJournal,
|
||||
workerWorkspaceTransferPaths,
|
||||
} from "./workspace-reconcile.js";
|
||||
import {
|
||||
applyStagedWorkerWorkspaceResult,
|
||||
deleteStagedWorkerWorkspaceResult,
|
||||
hasWorkerWorkspaceResultRef,
|
||||
preparedWorkerWorkspaceResultRef,
|
||||
workerWorkspaceResultStaging,
|
||||
workerWorkspaceResultRef,
|
||||
workerWorkspaceTransferPaths,
|
||||
} from "./workspace-result-staging.js";
|
||||
|
||||
const { prepareRequestedWorkerWorkspaceResult, stageWorkerWorkspaceResult } =
|
||||
workerWorkspaceResultStaging;
|
||||
|
||||
const roots: string[] = [];
|
||||
|
||||
@@ -80,6 +91,21 @@ function encodeManifest(value: unknown) {
|
||||
return { raw, ref: `sha256:${createHash("sha256").update(raw).digest("hex")}` };
|
||||
}
|
||||
|
||||
function encodeWorkspaceManifest(manifest: WorkerWorkspaceManifest) {
|
||||
return encodeManifest({
|
||||
version: manifest.version,
|
||||
baseCommit: manifest.baseCommit,
|
||||
entries: [
|
||||
...(manifest.directories ?? []).map((entryPath) => ({
|
||||
path: entryPath,
|
||||
type: "directory",
|
||||
mode: 0o700,
|
||||
})),
|
||||
...manifest.entries,
|
||||
].toSorted((left, right) => left.path.localeCompare(right.path)),
|
||||
});
|
||||
}
|
||||
|
||||
async function applyWorkspace(params: {
|
||||
root: string;
|
||||
stagingRoot: string;
|
||||
@@ -113,6 +139,216 @@ async function applyWorkspace(params: {
|
||||
}
|
||||
|
||||
describe("worker workspace reconciliation", () => {
|
||||
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",
|
||||
);
|
||||
for (const claimId of ["", "claim.with-dot", "claim_with_underscore", "claim/with-slash"]) {
|
||||
expect(() => workerWorkspaceResultRef(claimId)).toThrow(
|
||||
"Cloud workspace result claim id is invalid",
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
it("applies the complete candidate before publishing its recovery ref", async () => {
|
||||
const local = await temporaryDirectory("workspace-staged-ref-local");
|
||||
const payload = await temporaryDirectory("workspace-staged-ref-payload");
|
||||
const complete = await temporaryDirectory("workspace-staged-ref-complete");
|
||||
await Promise.all([
|
||||
fs.writeFile(path.join(local, "changed.txt"), "base\n"),
|
||||
fs.writeFile(path.join(local, "unchanged.txt"), "keep\n"),
|
||||
]);
|
||||
const base = encodeWorkspaceManifest(await manifestFor(local));
|
||||
await Promise.all([
|
||||
fs.writeFile(path.join(payload, "changed.txt"), "worker\n"),
|
||||
fs.writeFile(path.join(complete, "changed.txt"), "worker\n"),
|
||||
fs.writeFile(path.join(complete, "unchanged.txt"), "keep\n"),
|
||||
fs.writeFile(path.join(complete, "added.txt"), "added\n"),
|
||||
fs.writeFile(path.join(payload, "added.txt"), "added\n"),
|
||||
fs.writeFile(path.join(complete, "odd\nname.txt"), "quoted path\n"),
|
||||
fs.writeFile(path.join(payload, "odd\nname.txt"), "quoted path\n"),
|
||||
]);
|
||||
const current = encodeWorkspaceManifest(await manifestFor(complete));
|
||||
const ref = workerWorkspaceResultRef("claim-stage-order");
|
||||
let recordedRef: string | undefined;
|
||||
let acceptedManifestRef: string | undefined;
|
||||
const prepared = await prepareRequestedWorkerWorkspaceResult({
|
||||
request: {
|
||||
localPath: local,
|
||||
remoteWorkspaceDir: "/worker/workspace",
|
||||
baseManifestRef: base.ref,
|
||||
journal: {
|
||||
load: () => undefined,
|
||||
begin: () => {},
|
||||
commit: (manifestRef) => {
|
||||
expect(recordedRef).toBeUndefined();
|
||||
acceptedManifestRef = manifestRef;
|
||||
},
|
||||
abort: () => {},
|
||||
},
|
||||
stagedResult: {
|
||||
ref,
|
||||
record: (stagedResultRef) => {
|
||||
recordedRef = stagedResultRef;
|
||||
},
|
||||
},
|
||||
},
|
||||
stagingRoot: payload,
|
||||
currentManifestRef: current.ref,
|
||||
baseManifestRaw: base.raw,
|
||||
currentManifestRaw: current.raw,
|
||||
});
|
||||
|
||||
expect(
|
||||
await runCommandWithTimeout(
|
||||
["git", "-C", local, "show-ref", "--verify", preparedWorkerWorkspaceResultRef(ref)],
|
||||
{ timeoutMs: 10_000 },
|
||||
),
|
||||
).toMatchObject({ code: 0 });
|
||||
expect(
|
||||
(
|
||||
await runCommandWithTimeout(["git", "-C", local, "show-ref", "--verify", ref], {
|
||||
timeoutMs: 10_000,
|
||||
})
|
||||
).code,
|
||||
).not.toBe(0);
|
||||
expect(recordedRef).toBeUndefined();
|
||||
await expect(fs.readFile(path.join(local, "changed.txt"), "utf8")).resolves.toBe("base\n");
|
||||
await prepared.applyPreparedStagedResult();
|
||||
|
||||
expect(recordedRef).toBeUndefined();
|
||||
expect(acceptedManifestRef).toBe(current.ref);
|
||||
expect(
|
||||
await runCommandWithTimeout(
|
||||
["git", "-C", local, "show-ref", "--verify", preparedWorkerWorkspaceResultRef(ref)],
|
||||
{ timeoutMs: 10_000 },
|
||||
),
|
||||
).toMatchObject({ code: 0 });
|
||||
expect(
|
||||
(
|
||||
await runCommandWithTimeout(["git", "-C", local, "show-ref", "--verify", ref], {
|
||||
timeoutMs: 10_000,
|
||||
})
|
||||
).code,
|
||||
).not.toBe(0);
|
||||
await prepared.publishStagedResult();
|
||||
|
||||
expect(recordedRef).toBe(ref);
|
||||
expect(acceptedManifestRef).toBe(current.ref);
|
||||
await expect(fs.readFile(path.join(local, "changed.txt"), "utf8")).resolves.toBe("worker\n");
|
||||
await expect(fs.readFile(path.join(local, "unchanged.txt"), "utf8")).resolves.toBe("keep\n");
|
||||
await expect(fs.readFile(path.join(local, "added.txt"), "utf8")).resolves.toBe("added\n");
|
||||
await expect(fs.readFile(path.join(local, "odd\nname.txt"), "utf8")).resolves.toBe(
|
||||
"quoted path\n",
|
||||
);
|
||||
const replayCommit = vi.fn();
|
||||
await applyStagedWorkerWorkspaceResult({
|
||||
root: local,
|
||||
stagedResultRef: ref,
|
||||
expectedBaseManifestRef: current.ref,
|
||||
journal: {
|
||||
load: () => undefined,
|
||||
begin: () => {},
|
||||
commit: replayCommit,
|
||||
abort: () => {},
|
||||
},
|
||||
});
|
||||
expect(replayCommit).not.toHaveBeenCalled();
|
||||
await deleteStagedWorkerWorkspaceResult({ root: local, stagedResultRef: ref });
|
||||
expect(
|
||||
(
|
||||
await runCommandWithTimeout(["git", "-C", local, "show-ref", "--verify", ref], {
|
||||
timeoutMs: 10_000,
|
||||
})
|
||||
).code,
|
||||
).not.toBe(0);
|
||||
});
|
||||
|
||||
it("preserves the published result when its fence-row update fails", async () => {
|
||||
const local = await temporaryDirectory("workspace-staged-record-failure-local");
|
||||
const payload = await temporaryDirectory("workspace-staged-record-failure-payload");
|
||||
await fs.writeFile(path.join(local, "result.txt"), "base\n");
|
||||
await fs.writeFile(path.join(payload, "result.txt"), "worker\n");
|
||||
const base = encodeWorkspaceManifest(await manifestFor(local));
|
||||
const current = encodeWorkspaceManifest(await manifestFor(payload));
|
||||
const ref = workerWorkspaceResultRef("claim-record-failure");
|
||||
const prepared = await prepareRequestedWorkerWorkspaceResult({
|
||||
request: {
|
||||
localPath: local,
|
||||
remoteWorkspaceDir: "/worker/workspace",
|
||||
baseManifestRef: base.ref,
|
||||
journal: {
|
||||
load: () => undefined,
|
||||
begin: () => {},
|
||||
commit: () => {},
|
||||
abort: () => {},
|
||||
},
|
||||
stagedResult: {
|
||||
ref,
|
||||
record: () => {
|
||||
throw new Error("state database unavailable");
|
||||
},
|
||||
},
|
||||
},
|
||||
stagingRoot: payload,
|
||||
currentManifestRef: current.ref,
|
||||
baseManifestRaw: base.raw,
|
||||
currentManifestRaw: current.raw,
|
||||
});
|
||||
|
||||
await prepared.applyPreparedStagedResult();
|
||||
await expect(prepared.publishStagedResult()).rejects.toThrow("state database unavailable");
|
||||
await expect(hasWorkerWorkspaceResultRef({ root: local, stagedResultRef: ref })).resolves.toBe(
|
||||
true,
|
||||
);
|
||||
await expect(fs.readFile(path.join(local, "result.txt"), "utf8")).resolves.toBe("worker\n");
|
||||
await expect(
|
||||
hasWorkerWorkspaceResultRef({
|
||||
root: local,
|
||||
stagedResultRef: preparedWorkerWorkspaceResultRef(ref),
|
||||
}),
|
||||
).resolves.toBe(false);
|
||||
});
|
||||
|
||||
it("does not treat a Git probe failure as an absent staged result", async () => {
|
||||
const local = await temporaryDirectory("workspace-staged-probe-failure");
|
||||
await fs.writeFile(path.join(local, ".git"), "gitdir: /missing/openclaw-repository\n");
|
||||
|
||||
await expect(
|
||||
hasWorkerWorkspaceResultRef({
|
||||
root: local,
|
||||
stagedResultRef: workerWorkspaceResultRef("claim-probe-failure"),
|
||||
}),
|
||||
).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("disables workspace repository hooks while publishing result refs", async () => {
|
||||
const local = await temporaryDirectory("workspace-staged-hooks-local");
|
||||
const payload = await temporaryDirectory("workspace-staged-hooks-payload");
|
||||
await gitInit(local);
|
||||
const hook = path.join(local, ".git", "hooks", "reference-transaction");
|
||||
await fs.writeFile(hook, "#!/bin/sh\nexit 1\n", { mode: 0o755 });
|
||||
await fs.writeFile(path.join(local, "result.txt"), "base\n");
|
||||
await fs.writeFile(path.join(payload, "result.txt"), "worker\n");
|
||||
const base = encodeWorkspaceManifest(await manifestFor(local));
|
||||
const current = encodeWorkspaceManifest(await manifestFor(payload));
|
||||
const ref = workerWorkspaceResultRef("claim-disabled-hooks");
|
||||
|
||||
await stageWorkerWorkspaceResult({
|
||||
root: local,
|
||||
stagingRoot: payload,
|
||||
stagedResultRef: ref,
|
||||
baseManifestRef: base.ref,
|
||||
currentManifestRef: current.ref,
|
||||
baseManifestRaw: base.raw,
|
||||
currentManifestRaw: current.raw,
|
||||
});
|
||||
|
||||
await expect(hasWorkerWorkspaceResultRef({ root: local, stagedResultRef: ref })).resolves.toBe(
|
||||
true,
|
||||
);
|
||||
});
|
||||
|
||||
it("applies changed, added, deleted, executable, binary, and symlink results", async () => {
|
||||
const local = await temporaryDirectory("workspace-local");
|
||||
const staged = await temporaryDirectory("workspace-staged");
|
||||
|
||||
@@ -2,8 +2,7 @@ import { createHash, randomBytes } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { runCommandBuffered } from "../../process/exec.js";
|
||||
import { runCommandWithTimeout } from "../../process/exec.js";
|
||||
import { runCommandBuffered, runCommandWithTimeout } from "../../process/exec.js";
|
||||
import {
|
||||
MAX_RECONCILIATION_ENTRIES,
|
||||
MAX_RECONCILIATION_FILE_BYTES,
|
||||
@@ -98,27 +97,6 @@ function hasReplacedBaseEntryAncestor(
|
||||
return false;
|
||||
}
|
||||
|
||||
export function workerWorkspaceTransferPaths(
|
||||
current: WorkerWorkspaceManifest,
|
||||
base: WorkerWorkspaceManifest,
|
||||
): string[] {
|
||||
const changed = changedPaths(base, current);
|
||||
const paths = reconciliationEntries(current.entries)
|
||||
.filter((entry) => changed.has(entry.path))
|
||||
.map((entry) => {
|
||||
if (entry.type === "file" && entry.size > MAX_RECONCILIATION_FILE_BYTES) {
|
||||
throw new Error(`Cloud workspace result is too large: ${entry.path}`);
|
||||
}
|
||||
return entry.path;
|
||||
});
|
||||
if (paths.length > MAX_RECONCILIATION_ENTRIES) {
|
||||
throw new Error(
|
||||
`Cloud workspace reconciliation exceeds the ${MAX_RECONCILIATION_ENTRIES} entry limit`,
|
||||
);
|
||||
}
|
||||
return paths;
|
||||
}
|
||||
|
||||
async function preflightWorkspaceApply(params: {
|
||||
root: string;
|
||||
base: WorkerWorkspaceManifest;
|
||||
|
||||
571
src/gateway/worker-environments/workspace-result-staging.ts
Normal file
571
src/gateway/worker-environments/workspace-result-staging.ts
Normal file
@@ -0,0 +1,571 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { runCommandBuffered, runCommandWithTimeout } from "../../process/exec.js";
|
||||
import type { WorkerWorkspaceReconcileRequest } from "./tunnel-contract.js";
|
||||
import {
|
||||
MAX_RECONCILIATION_ENTRIES,
|
||||
MAX_RECONCILIATION_FILE_BYTES,
|
||||
MAX_RECONCILIATION_TOTAL_BYTES,
|
||||
parseWorkerWorkspaceManifest,
|
||||
type WorkerWorkspaceManifest,
|
||||
type WorkerWorkspaceManifestEntry,
|
||||
type WorkerWorkspaceReconciliationJournalAdapter,
|
||||
} from "./workspace-manifest.js";
|
||||
import { reconciliationEntries } from "./workspace-reconcile-derived-paths.js";
|
||||
import { absoluteEntryMatches, localPath } from "./workspace-reconcile-fs.js";
|
||||
import { applyStagedWorkerWorkspace, assertWorkspaceResultStable } from "./workspace-reconcile.js";
|
||||
|
||||
const PATCH_TIMEOUT_MS = 10 * 60_000;
|
||||
// Match managed-worktree refs/openclaw/snapshots: deleting the owning ref is
|
||||
// sufficient; unreachable objects may remain until normal Git GC.
|
||||
const WORKER_RESULT_REF_PREFIX = "refs/openclaw/worker-results";
|
||||
const WORKER_RESULT_CANDIDATE_REF_PREFIX = "refs/openclaw/worker-result-candidates";
|
||||
const WORKER_RESULT_CLAIM_ID_PATTERN = /^[A-Za-z0-9-]+$/u;
|
||||
const STAGED_RESULT_MESSAGE = "OpenClaw worker workspace result";
|
||||
const STAGED_RESULT_METADATA_LIMIT = 128 * 1024 * 1024 + 4_096;
|
||||
// Git documents the platform null device as the per-command way to disable
|
||||
// hooks. An unowned path under a shared temp dir could be populated by another user.
|
||||
const DISABLED_GIT_HOOKS_PATH = os.devNull;
|
||||
|
||||
function gitCommand(cwd: string, args: string[]): string[] {
|
||||
return ["git", "-c", `core.hooksPath=${DISABLED_GIT_HOOKS_PATH}`, "-C", cwd, ...args];
|
||||
}
|
||||
|
||||
function sameEntry(
|
||||
left: WorkerWorkspaceManifestEntry | undefined,
|
||||
right: WorkerWorkspaceManifestEntry | undefined,
|
||||
): boolean {
|
||||
return JSON.stringify(left) === JSON.stringify(right);
|
||||
}
|
||||
|
||||
function changedPaths(
|
||||
base: WorkerWorkspaceManifest,
|
||||
current: WorkerWorkspaceManifest,
|
||||
): Set<string> {
|
||||
const baseByPath = new Map(
|
||||
reconciliationEntries(base.entries).map((entry) => [entry.path, entry]),
|
||||
);
|
||||
const currentByPath = new Map(
|
||||
reconciliationEntries(current.entries).map((entry) => [entry.path, entry]),
|
||||
);
|
||||
return new Set(
|
||||
[...new Set([...baseByPath.keys(), ...currentByPath.keys()])].filter(
|
||||
(entryPath) => !sameEntry(baseByPath.get(entryPath), currentByPath.get(entryPath)),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
export function workerWorkspaceTransferPaths(
|
||||
current: WorkerWorkspaceManifest,
|
||||
base: WorkerWorkspaceManifest,
|
||||
): string[] {
|
||||
const changed = changedPaths(base, current);
|
||||
const paths = reconciliationEntries(current.entries)
|
||||
.filter((entry) => changed.has(entry.path))
|
||||
.map((entry) => {
|
||||
if (entry.type === "file" && entry.size > MAX_RECONCILIATION_FILE_BYTES) {
|
||||
throw new Error(`Cloud workspace result is too large: ${entry.path}`);
|
||||
}
|
||||
return entry.path;
|
||||
});
|
||||
if (paths.length > MAX_RECONCILIATION_ENTRIES) {
|
||||
throw new Error(
|
||||
`Cloud workspace reconciliation exceeds the ${MAX_RECONCILIATION_ENTRIES} entry limit`,
|
||||
);
|
||||
}
|
||||
return paths;
|
||||
}
|
||||
|
||||
async function requireGit(cwd: string, args: string[]): Promise<string> {
|
||||
const result = await runCommandWithTimeout(gitCommand(cwd, args), {
|
||||
timeoutMs: PATCH_TIMEOUT_MS,
|
||||
maxOutputBytes: 1024 * 1024,
|
||||
});
|
||||
if (result.termination !== "exit" || result.code !== 0) {
|
||||
throw new Error((result.stderr || result.stdout || `git ${args[0]} failed`).trim());
|
||||
}
|
||||
return result.stdout.trim();
|
||||
}
|
||||
|
||||
function requireWorkerResultStorageRef(ref: string): string {
|
||||
if (
|
||||
!new RegExp(
|
||||
`^(?:${WORKER_RESULT_REF_PREFIX}|${WORKER_RESULT_CANDIDATE_REF_PREFIX})/[A-Za-z0-9-]+$`,
|
||||
"u",
|
||||
).test(ref)
|
||||
) {
|
||||
throw new Error("Cloud workspace staged result reference is invalid");
|
||||
}
|
||||
return ref;
|
||||
}
|
||||
|
||||
function requireWorkerResultRef(ref: string): string {
|
||||
if (!ref.startsWith(`${WORKER_RESULT_REF_PREFIX}/`)) {
|
||||
throw new Error("Cloud workspace staged result reference is invalid");
|
||||
}
|
||||
return requireWorkerResultStorageRef(ref);
|
||||
}
|
||||
|
||||
export function workerWorkspaceResultRef(claimId: string): string {
|
||||
if (!WORKER_RESULT_CLAIM_ID_PATTERN.test(claimId)) {
|
||||
throw new Error("Cloud workspace result claim id is invalid");
|
||||
}
|
||||
return `${WORKER_RESULT_REF_PREFIX}/${claimId}`;
|
||||
}
|
||||
|
||||
export function preparedWorkerWorkspaceResultRef(stagedResultRef: string): string {
|
||||
const ref = requireWorkerResultRef(stagedResultRef);
|
||||
return `${WORKER_RESULT_CANDIDATE_REF_PREFIX}/${ref.slice(WORKER_RESULT_REF_PREFIX.length + 1)}`;
|
||||
}
|
||||
|
||||
async function hasGitAdminPath(root: string): Promise<boolean> {
|
||||
let current = root;
|
||||
while (true) {
|
||||
try {
|
||||
await fs.lstat(path.join(current, ".git"));
|
||||
return true;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
const parent = path.dirname(current);
|
||||
if (parent === current) {
|
||||
return false;
|
||||
}
|
||||
current = parent;
|
||||
}
|
||||
}
|
||||
|
||||
async function ensureWorkerWorkspaceResultRepository(root: string): Promise<string> {
|
||||
const resolved = await fs.realpath(root);
|
||||
const probe = await runCommandWithTimeout(gitCommand(resolved, ["rev-parse", "--git-dir"]), {
|
||||
timeoutMs: PATCH_TIMEOUT_MS,
|
||||
maxOutputBytes: 1024 * 1024,
|
||||
});
|
||||
if (probe.termination === "exit" && probe.code === 0) {
|
||||
return resolved;
|
||||
}
|
||||
await requireGit(resolved, ["init", "--quiet", "--object-format=sha1"]);
|
||||
return resolved;
|
||||
}
|
||||
|
||||
export async function hasWorkerWorkspaceResultRef(params: {
|
||||
root: string;
|
||||
stagedResultRef: string;
|
||||
}): Promise<boolean> {
|
||||
let root: string;
|
||||
try {
|
||||
root = await fs.realpath(params.root);
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
|
||||
return false;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
if (!(await hasGitAdminPath(root))) {
|
||||
return false;
|
||||
}
|
||||
const result = await runCommandWithTimeout(
|
||||
gitCommand(root, [
|
||||
"show-ref",
|
||||
"--verify",
|
||||
"--quiet",
|
||||
requireWorkerResultStorageRef(params.stagedResultRef),
|
||||
]),
|
||||
{ timeoutMs: PATCH_TIMEOUT_MS, maxOutputBytes: 1024 * 1024 },
|
||||
);
|
||||
if (result.termination === "exit" && result.code === 0) {
|
||||
return true;
|
||||
}
|
||||
if (result.termination === "exit" && result.code === 1) {
|
||||
return false;
|
||||
}
|
||||
throw new Error((result.stderr || result.stdout || "git show-ref failed").trim());
|
||||
}
|
||||
|
||||
function stagedResultMessage(params: {
|
||||
baseManifestRef: string;
|
||||
currentManifestRef: string;
|
||||
baseManifestRaw: string;
|
||||
currentManifestRaw: string;
|
||||
}): Buffer {
|
||||
const base = Buffer.from(params.baseManifestRaw);
|
||||
const current = Buffer.from(params.currentManifestRaw);
|
||||
const header = Buffer.from(
|
||||
`${STAGED_RESULT_MESSAGE}\nversion 1\nbase-ref ${params.baseManifestRef}\ncurrent-ref ${params.currentManifestRef}\nbase-bytes ${base.byteLength}\ncurrent-bytes ${current.byteLength}\n\n`,
|
||||
);
|
||||
return Buffer.concat([header, base, current]);
|
||||
}
|
||||
|
||||
function quoteFastImportPath(entryPath: string): string {
|
||||
const bytes = Buffer.from(entryPath);
|
||||
let quoted = '"';
|
||||
for (const byte of bytes) {
|
||||
if (byte === 0) {
|
||||
throw new Error("Cloud workspace staged result path contains a null byte");
|
||||
}
|
||||
if (byte === 0x22 || byte === 0x5c) {
|
||||
quoted += `\\${String.fromCharCode(byte)}`;
|
||||
} else if (byte >= 0x20 && byte < 0x7f) {
|
||||
quoted += String.fromCharCode(byte);
|
||||
} else {
|
||||
quoted += `\\${byte.toString(8).padStart(3, "0")}`;
|
||||
}
|
||||
}
|
||||
return `${quoted}"`;
|
||||
}
|
||||
|
||||
async function readGitBlob(params: {
|
||||
root: string;
|
||||
objectId: string;
|
||||
maxBytes: number;
|
||||
}): Promise<Buffer> {
|
||||
const result = await runCommandBuffered(
|
||||
gitCommand(params.root, ["cat-file", "blob", params.objectId]),
|
||||
{ timeoutMs: PATCH_TIMEOUT_MS, maxOutputBytes: params.maxBytes + 1 },
|
||||
);
|
||||
if (result.termination !== "exit" || result.code !== 0) {
|
||||
throw new Error(result.stderr.toString("utf8").trim() || "git cat-file failed");
|
||||
}
|
||||
if (result.stdout.byteLength > params.maxBytes) {
|
||||
throw new Error("Cloud workspace staged result exceeds its byte limit");
|
||||
}
|
||||
return result.stdout;
|
||||
}
|
||||
|
||||
async function stageWorkerWorkspaceResult(params: {
|
||||
root: string;
|
||||
stagingRoot: string;
|
||||
stagedResultRef: string;
|
||||
baseManifestRef: string;
|
||||
currentManifestRef: string;
|
||||
baseManifestRaw: string;
|
||||
currentManifestRaw: string;
|
||||
}): Promise<void> {
|
||||
const root = await ensureWorkerWorkspaceResultRepository(params.root);
|
||||
const stagedResultRef = requireWorkerResultStorageRef(params.stagedResultRef);
|
||||
const base = parseWorkerWorkspaceManifest(params.baseManifestRaw, params.baseManifestRef);
|
||||
const current = parseWorkerWorkspaceManifest(
|
||||
params.currentManifestRaw,
|
||||
params.currentManifestRef,
|
||||
);
|
||||
// This ref is the complete worker-result artifact, not a patch cache. Keep
|
||||
// unchanged blobs too so recovery is self-contained after the worker dies.
|
||||
const entries = reconciliationEntries(current.entries).toSorted((left, right) =>
|
||||
left.path.localeCompare(right.path),
|
||||
);
|
||||
if (entries.length > MAX_RECONCILIATION_ENTRIES) {
|
||||
throw new Error(
|
||||
`Cloud workspace reconciliation exceeds the ${MAX_RECONCILIATION_ENTRIES} entry limit`,
|
||||
);
|
||||
}
|
||||
const changed = changedPaths(base, current);
|
||||
const blobs: Array<{ entry: WorkerWorkspaceManifestEntry; mark: number; content: Buffer }> = [];
|
||||
let totalBytes = 0;
|
||||
for (const [index, entry] of entries.entries()) {
|
||||
if (entry.type === "file" && entry.size > MAX_RECONCILIATION_FILE_BYTES) {
|
||||
throw new Error(`Cloud workspace result is too large: ${entry.path}`);
|
||||
}
|
||||
const sourceRoot = changed.has(entry.path) ? params.stagingRoot : root;
|
||||
const source = localPath(sourceRoot, entry.path);
|
||||
if (!(await absoluteEntryMatches(source, entry))) {
|
||||
throw new Error(`Cloud workspace staged payload is invalid: ${entry.path}`);
|
||||
}
|
||||
const content =
|
||||
entry.type === "symlink" ? Buffer.from(entry.target) : await fs.readFile(source);
|
||||
if (
|
||||
entry.type === "file" &&
|
||||
(content.byteLength !== entry.size ||
|
||||
createHash("sha256").update(content).digest("hex") !== entry.sha256)
|
||||
) {
|
||||
throw new Error(`Cloud workspace staged payload changed while reading: ${entry.path}`);
|
||||
}
|
||||
totalBytes += content.byteLength;
|
||||
if (totalBytes > MAX_RECONCILIATION_TOTAL_BYTES) {
|
||||
throw new Error("Cloud workspace staged result exceeds its byte limit");
|
||||
}
|
||||
blobs.push({ entry, mark: index + 1, content });
|
||||
}
|
||||
const message = stagedResultMessage(params);
|
||||
const chunks: Uint8Array[] = [];
|
||||
for (const blob of blobs) {
|
||||
chunks.push(Buffer.from(`blob\nmark :${blob.mark}\ndata ${blob.content.byteLength}\n`));
|
||||
chunks.push(blob.content, Buffer.from("\n"));
|
||||
}
|
||||
chunks.push(
|
||||
Buffer.from(
|
||||
`commit ${stagedResultRef}\nauthor OpenClaw <openclaw@localhost> 0 +0000\ncommitter OpenClaw <openclaw@localhost> 0 +0000\ndata ${message.byteLength}\n`,
|
||||
),
|
||||
message,
|
||||
Buffer.from("\ndeleteall\n"),
|
||||
);
|
||||
for (const blob of blobs) {
|
||||
const mode =
|
||||
blob.entry.type === "symlink"
|
||||
? "120000"
|
||||
: (blob.entry.mode & 0o111) !== 0
|
||||
? "100755"
|
||||
: "100644";
|
||||
chunks.push(Buffer.from(`M ${mode} :${blob.mark} ${quoteFastImportPath(blob.entry.path)}\n`));
|
||||
}
|
||||
chunks.push(Buffer.from("done\n"));
|
||||
const imported = await runCommandBuffered(gitCommand(root, ["fast-import", "--quiet"]), {
|
||||
input: Buffer.concat(chunks),
|
||||
timeoutMs: PATCH_TIMEOUT_MS,
|
||||
maxOutputBytes: { stdout: 1024 * 1024, stderr: 1024 * 1024 },
|
||||
});
|
||||
if (imported.termination !== "exit" || imported.code !== 0) {
|
||||
throw new Error(imported.stderr.toString("utf8").trim() || "git fast-import failed");
|
||||
}
|
||||
await requireGit(root, ["rev-parse", `${stagedResultRef}^{commit}`]);
|
||||
}
|
||||
|
||||
type LoadedStagedWorkerWorkspace = {
|
||||
baseManifestRef: string;
|
||||
currentManifestRef: string;
|
||||
base: WorkerWorkspaceManifest;
|
||||
current: WorkerWorkspaceManifest;
|
||||
objectsByPath: Map<string, { mode: string; objectId: string }>;
|
||||
};
|
||||
|
||||
async function loadStagedWorkerWorkspace(
|
||||
root: string,
|
||||
stagedResultRef: string,
|
||||
): Promise<LoadedStagedWorkerWorkspace> {
|
||||
const ref = requireWorkerResultStorageRef(stagedResultRef);
|
||||
const rawCommit = await runCommandBuffered(gitCommand(root, ["cat-file", "commit", ref]), {
|
||||
timeoutMs: PATCH_TIMEOUT_MS,
|
||||
maxOutputBytes: STAGED_RESULT_METADATA_LIMIT,
|
||||
});
|
||||
if (rawCommit.termination !== "exit" || rawCommit.code !== 0) {
|
||||
throw new Error(rawCommit.stderr.toString("utf8").trim() || "git cat-file failed");
|
||||
}
|
||||
const commitHeaderEnd = rawCommit.stdout.indexOf("\n\n");
|
||||
if (commitHeaderEnd < 0) {
|
||||
throw new Error("Cloud workspace staged result metadata is invalid");
|
||||
}
|
||||
const message = rawCommit.stdout.subarray(commitHeaderEnd + 2);
|
||||
const metadataEnd = message.indexOf("\n\n");
|
||||
if (metadataEnd < 0) {
|
||||
throw new Error("Cloud workspace staged result metadata is invalid");
|
||||
}
|
||||
const lines = message.subarray(0, metadataEnd).toString("utf8").split("\n");
|
||||
const match = /^sha256:[a-f0-9]{64}$/u;
|
||||
const baseManifestRef = lines[2]?.slice("base-ref ".length) ?? "";
|
||||
const currentManifestRef = lines[3]?.slice("current-ref ".length) ?? "";
|
||||
const baseBytes = Number(lines[4]?.slice("base-bytes ".length));
|
||||
const currentBytes = Number(lines[5]?.slice("current-bytes ".length));
|
||||
if (
|
||||
lines[0] !== STAGED_RESULT_MESSAGE ||
|
||||
lines[1] !== "version 1" ||
|
||||
!lines[2]?.startsWith("base-ref ") ||
|
||||
!lines[3]?.startsWith("current-ref ") ||
|
||||
!lines[4]?.startsWith("base-bytes ") ||
|
||||
!lines[5]?.startsWith("current-bytes ") ||
|
||||
lines.length !== 6 ||
|
||||
!match.test(baseManifestRef) ||
|
||||
!match.test(currentManifestRef) ||
|
||||
!Number.isSafeInteger(baseBytes) ||
|
||||
baseBytes < 0 ||
|
||||
!Number.isSafeInteger(currentBytes) ||
|
||||
currentBytes < 0
|
||||
) {
|
||||
throw new Error("Cloud workspace staged result metadata is invalid");
|
||||
}
|
||||
const manifests = message.subarray(metadataEnd + 2);
|
||||
if (manifests.byteLength !== baseBytes + currentBytes) {
|
||||
throw new Error("Cloud workspace staged result metadata is truncated");
|
||||
}
|
||||
const baseRaw = manifests.subarray(0, baseBytes).toString("utf8");
|
||||
const currentRaw = manifests.subarray(baseBytes).toString("utf8");
|
||||
const base = parseWorkerWorkspaceManifest(baseRaw, baseManifestRef);
|
||||
const current = parseWorkerWorkspaceManifest(currentRaw, currentManifestRef);
|
||||
const tree = await runCommandBuffered(
|
||||
gitCommand(root, ["ls-tree", "-r", "-z", "--full-tree", ref]),
|
||||
{ timeoutMs: PATCH_TIMEOUT_MS, maxOutputBytes: 2 * MAX_RECONCILIATION_FILE_BYTES },
|
||||
);
|
||||
if (tree.termination !== "exit" || tree.code !== 0) {
|
||||
throw new Error(tree.stderr.toString("utf8").trim() || "git ls-tree failed");
|
||||
}
|
||||
const objectsByPath = new Map<string, { mode: string; objectId: string }>();
|
||||
for (const record of tree.stdout.toString("utf8").split("\0").filter(Boolean)) {
|
||||
const parsed = /^(100644|100755|120000) blob ([a-f0-9]{40}|[a-f0-9]{64})\t([\s\S]+)$/u.exec(
|
||||
record,
|
||||
);
|
||||
if (!parsed) {
|
||||
throw new Error("Cloud workspace staged result tree is invalid");
|
||||
}
|
||||
objectsByPath.set(parsed[3]!, { mode: parsed[1]!, objectId: parsed[2]! });
|
||||
}
|
||||
const entries = reconciliationEntries(current.entries);
|
||||
if (objectsByPath.size !== entries.length) {
|
||||
throw new Error("Cloud workspace staged result tree does not match its manifest");
|
||||
}
|
||||
for (const entry of entries) {
|
||||
const object = objectsByPath.get(entry.path);
|
||||
const expectedMode =
|
||||
entry.type === "symlink" ? "120000" : (entry.mode & 0o111) !== 0 ? "100755" : "100644";
|
||||
if (!object || object.mode !== expectedMode) {
|
||||
throw new Error(`Cloud workspace staged result tree is invalid: ${entry.path}`);
|
||||
}
|
||||
}
|
||||
return { baseManifestRef, currentManifestRef, base, current, objectsByPath };
|
||||
}
|
||||
|
||||
async function materializeStagedEntry(params: {
|
||||
root: string;
|
||||
entry: WorkerWorkspaceManifestEntry;
|
||||
content?: Uint8Array;
|
||||
}): Promise<void> {
|
||||
const target = localPath(params.root, params.entry.path);
|
||||
await fs.mkdir(path.dirname(target), { recursive: true, mode: 0o700 });
|
||||
if (params.entry.type === "symlink") {
|
||||
await fs.symlink(params.entry.target, target);
|
||||
return;
|
||||
}
|
||||
if (!params.content) {
|
||||
throw new Error(`Cloud workspace staged content is missing: ${params.entry.path}`);
|
||||
}
|
||||
await fs.writeFile(target, params.content, { mode: params.entry.mode, flag: "wx" });
|
||||
await fs.chmod(target, params.entry.mode);
|
||||
if (!(await absoluteEntryMatches(target, params.entry))) {
|
||||
throw new Error(`Cloud workspace staged payload is invalid: ${params.entry.path}`);
|
||||
}
|
||||
}
|
||||
|
||||
export async function applyStagedWorkerWorkspaceResult(params: {
|
||||
root: string;
|
||||
stagedResultRef: string;
|
||||
expectedBaseManifestRef: string;
|
||||
journal: WorkerWorkspaceReconciliationJournalAdapter;
|
||||
}): Promise<{ manifestRef: string; changed: boolean; verifyLocalStable(): Promise<void> }> {
|
||||
const root = await fs.realpath(params.root);
|
||||
const staged = await loadStagedWorkerWorkspace(root, params.stagedResultRef);
|
||||
if (staged.baseManifestRef !== params.expectedBaseManifestRef) {
|
||||
if (staged.currentManifestRef !== params.expectedBaseManifestRef) {
|
||||
throw new Error("Cloud workspace staged result does not match the placement base");
|
||||
}
|
||||
// The workspace-base commit and local tree precede fence acceptance. A crash
|
||||
// between them must make replay a stable no-op, not strand the durable result.
|
||||
await assertWorkspaceResultStable({ root, base: staged.base, current: staged.current });
|
||||
return {
|
||||
manifestRef: staged.currentManifestRef,
|
||||
changed: changedPaths(staged.base, staged.current).size > 0,
|
||||
verifyLocalStable: async () =>
|
||||
await assertWorkspaceResultStable({ root, base: staged.base, current: staged.current }),
|
||||
};
|
||||
}
|
||||
const changed = changedPaths(staged.base, staged.current);
|
||||
const stagingRoot = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-staged-result-"));
|
||||
try {
|
||||
for (const entry of reconciliationEntries(staged.current.entries)) {
|
||||
if (!changed.has(entry.path)) {
|
||||
continue;
|
||||
}
|
||||
const object = staged.objectsByPath.get(entry.path)!;
|
||||
const content = await readGitBlob({
|
||||
root,
|
||||
objectId: object.objectId,
|
||||
maxBytes: MAX_RECONCILIATION_FILE_BYTES,
|
||||
});
|
||||
if (entry.type === "symlink") {
|
||||
if (content.toString("utf8") !== entry.target) {
|
||||
throw new Error(`Cloud workspace staged result tree is invalid: ${entry.path}`);
|
||||
}
|
||||
await materializeStagedEntry({ root: stagingRoot, entry });
|
||||
} else {
|
||||
await materializeStagedEntry({ root: stagingRoot, entry, content });
|
||||
}
|
||||
}
|
||||
await applyStagedWorkerWorkspace({
|
||||
root,
|
||||
stagingRoot,
|
||||
baseManifestRef: staged.baseManifestRef,
|
||||
currentManifestRef: staged.currentManifestRef,
|
||||
base: staged.base,
|
||||
current: staged.current,
|
||||
journal: params.journal,
|
||||
});
|
||||
} finally {
|
||||
await fs.rm(stagingRoot, { recursive: true, force: true });
|
||||
}
|
||||
return {
|
||||
manifestRef: staged.currentManifestRef,
|
||||
changed: changed.size > 0,
|
||||
verifyLocalStable: async () =>
|
||||
await assertWorkspaceResultStable({ root, base: staged.base, current: staged.current }),
|
||||
};
|
||||
}
|
||||
|
||||
async function prepareRequestedWorkerWorkspaceResult(params: {
|
||||
request: WorkerWorkspaceReconcileRequest;
|
||||
stagingRoot: string;
|
||||
currentManifestRef: string;
|
||||
baseManifestRaw: string;
|
||||
currentManifestRaw: string;
|
||||
}): Promise<{
|
||||
applyPreparedStagedResult(): Promise<void>;
|
||||
publishStagedResult(): Promise<void>;
|
||||
discardPreparedStagedResult(): Promise<void>;
|
||||
}> {
|
||||
const stagedResult = params.request.stagedResult;
|
||||
if (!stagedResult) {
|
||||
throw new Error("Cloud workspace durable result staging was not requested");
|
||||
}
|
||||
const candidateRef = preparedWorkerWorkspaceResultRef(stagedResult.ref);
|
||||
await stageWorkerWorkspaceResult({
|
||||
root: params.request.localPath,
|
||||
stagingRoot: params.stagingRoot,
|
||||
stagedResultRef: candidateRef,
|
||||
baseManifestRef: params.request.baseManifestRef,
|
||||
currentManifestRef: params.currentManifestRef,
|
||||
baseManifestRaw: params.baseManifestRaw,
|
||||
currentManifestRaw: params.currentManifestRaw,
|
||||
});
|
||||
return {
|
||||
applyPreparedStagedResult: async () => {
|
||||
const root = await ensureWorkerWorkspaceResultRepository(params.request.localPath);
|
||||
await applyStagedWorkerWorkspaceResult({
|
||||
root,
|
||||
stagedResultRef: candidateRef,
|
||||
expectedBaseManifestRef: params.request.baseManifestRef,
|
||||
journal: params.request.journal,
|
||||
});
|
||||
},
|
||||
publishStagedResult: async () => {
|
||||
const root = await ensureWorkerWorkspaceResultRepository(params.request.localPath);
|
||||
const commit = await requireGit(root, ["rev-parse", `${candidateRef}^{commit}`]);
|
||||
await requireGit(root, ["update-ref", stagedResult.ref, commit]);
|
||||
await requireGit(root, ["update-ref", "-d", candidateRef]);
|
||||
// Final fences precede publishing. Preserve the canonical ref on any
|
||||
// SQLite failure so restart recovery can discover the verified result.
|
||||
stagedResult.record(stagedResult.ref);
|
||||
},
|
||||
discardPreparedStagedResult: async () => {
|
||||
await deleteStagedWorkerWorkspaceResult({
|
||||
root: params.request.localPath,
|
||||
stagedResultRef: candidateRef,
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export async function deleteStagedWorkerWorkspaceResult(params: {
|
||||
root: string;
|
||||
stagedResultRef: string;
|
||||
}): Promise<void> {
|
||||
const root = await fs.realpath(params.root);
|
||||
const stagedResultRef = requireWorkerResultStorageRef(params.stagedResultRef);
|
||||
await requireGit(root, ["update-ref", "-d", stagedResultRef]);
|
||||
if (stagedResultRef.startsWith(`${WORKER_RESULT_REF_PREFIX}/`)) {
|
||||
await requireGit(root, ["update-ref", "-d", preparedWorkerWorkspaceResultRef(stagedResultRef)]);
|
||||
}
|
||||
}
|
||||
|
||||
export const workerWorkspaceResultStaging = {
|
||||
prepareRequestedWorkerWorkspaceResult,
|
||||
stageWorkerWorkspaceResult,
|
||||
};
|
||||
@@ -51,6 +51,46 @@ export function workspaceSyncError(result: SpawnResult): Error {
|
||||
);
|
||||
}
|
||||
|
||||
export async function probeWorkspaceGitMode(params: {
|
||||
localPath: string;
|
||||
commandOptions: CommandOptions;
|
||||
runTask: (argv: string[], options: CommandOptions) => Promise<SpawnResult>;
|
||||
}): Promise<{ mode: "git" | "plain"; gitRoot: string; baseCommit: string }> {
|
||||
const gitAdmin = await fs.lstat(path.join(params.localPath, ".git")).catch((error: unknown) => {
|
||||
if (typeof error === "object" && error !== null && "code" in error && error.code === "ENOENT") {
|
||||
return undefined;
|
||||
}
|
||||
throw error;
|
||||
});
|
||||
if (!gitAdmin) {
|
||||
return { mode: "plain", gitRoot: params.localPath, baseCommit: "" };
|
||||
}
|
||||
const [gitRootResult, gitBaseResult] = await Promise.all([
|
||||
params.runTask(
|
||||
["git", "-C", params.localPath, "rev-parse", "--show-toplevel"],
|
||||
params.commandOptions,
|
||||
),
|
||||
params.runTask(
|
||||
["git", "-C", params.localPath, "rev-parse", "--verify", "--quiet", "HEAD"],
|
||||
params.commandOptions,
|
||||
),
|
||||
]);
|
||||
if (!workerWorkspaceCommandSucceeded(gitRootResult)) {
|
||||
throw workspaceSyncError(gitRootResult);
|
||||
}
|
||||
if (workerWorkspaceCommandSucceeded(gitBaseResult)) {
|
||||
return {
|
||||
mode: "git",
|
||||
gitRoot: gitRootResult.stdout.trim(),
|
||||
baseCommit: gitBaseResult.stdout.trim(),
|
||||
};
|
||||
}
|
||||
if (gitBaseResult.termination === "exit" && gitBaseResult.code === 1) {
|
||||
return { mode: "plain", gitRoot: params.localPath, baseCommit: "" };
|
||||
}
|
||||
throw workspaceSyncError(gitBaseResult);
|
||||
}
|
||||
|
||||
export function stableWorkerPathComponent(value: string, length: number): string {
|
||||
return createHash("sha256").update(value).digest("hex").slice(0, length);
|
||||
}
|
||||
|
||||
@@ -26,12 +26,16 @@ import {
|
||||
MAX_RECONCILIATION_TOTAL_BYTES,
|
||||
parseWorkerWorkspaceManifest,
|
||||
recoverWorkerWorkspaceReconciliation,
|
||||
workerWorkspaceTransferPaths,
|
||||
} from "./workspace-reconcile.js";
|
||||
import {
|
||||
workerWorkspaceResultStaging,
|
||||
workerWorkspaceTransferPaths,
|
||||
} from "./workspace-result-staging.js";
|
||||
import {
|
||||
MANIFEST_REF_PATTERN,
|
||||
parseManifestRef,
|
||||
parseRemoteWorkspaceDirectory,
|
||||
probeWorkspaceGitMode,
|
||||
readTransferredManifest,
|
||||
runBoundedInboundRsync as runBoundedInboundRsyncTransfer,
|
||||
stableWorkerPathComponent,
|
||||
@@ -250,17 +254,15 @@ export function createWorkerWorkspaceActions(
|
||||
throw workspaceSyncError(setup);
|
||||
}
|
||||
const remoteWorkspaceDir = parseRemoteWorkspaceDirectory(setup.stdout.trim());
|
||||
|
||||
const gitRootResult = await runTask(
|
||||
["git", "-C", request.localPath, "rev-parse", "--show-toplevel"],
|
||||
workerSshCommandOptions({
|
||||
// Result refs can make plain workspaces unborn repos; only committed repos use Git sync.
|
||||
const { mode, gitRoot, baseCommit } = await probeWorkspaceGitMode({
|
||||
localPath: request.localPath,
|
||||
commandOptions: workerSshCommandOptions({
|
||||
timeoutMs: REMOTE_SETUP_TIMEOUT_MS,
|
||||
signal: options.ownerSignal,
|
||||
}),
|
||||
);
|
||||
const mode = success(gitRootResult) ? "git" : "plain";
|
||||
let baseCommit = "";
|
||||
let gitRoot = request.localPath;
|
||||
runTask,
|
||||
});
|
||||
const temporaryDirectory = await fs.mkdtemp(
|
||||
path.join(os.tmpdir(), "openclaw-worker-workspace-sync-"),
|
||||
);
|
||||
@@ -276,7 +278,6 @@ export function createWorkerWorkspaceActions(
|
||||
try {
|
||||
let fileListPath: string | undefined;
|
||||
if (mode === "git") {
|
||||
gitRoot = gitRootResult.stdout.trim();
|
||||
const [canonicalRequestPath, canonicalGitRoot] = await Promise.all([
|
||||
fs.realpath(request.localPath),
|
||||
fs.realpath(gitRoot),
|
||||
@@ -284,17 +285,6 @@ export function createWorkerWorkspaceActions(
|
||||
if (canonicalRequestPath !== canonicalGitRoot) {
|
||||
throw new Error("Worker git workspace sync requires the managed worktree root");
|
||||
}
|
||||
const gitBase = await runTask(
|
||||
["git", "-C", gitRoot, "rev-parse", "--verify", "HEAD"],
|
||||
workerSshCommandOptions({
|
||||
timeoutMs: REMOTE_SETUP_TIMEOUT_MS,
|
||||
signal: options.ownerSignal,
|
||||
}),
|
||||
);
|
||||
if (!success(gitBase)) {
|
||||
throw new Error("Worker git workspace has no base commit");
|
||||
}
|
||||
baseCommit = gitBase.stdout.trim();
|
||||
if (!GIT_COMMIT_PATTERN.test(baseCommit)) {
|
||||
throw new Error("Worker workspace git base is not a commit id");
|
||||
}
|
||||
@@ -586,7 +576,18 @@ export function createWorkerWorkspaceActions(
|
||||
const currentRef = parseManifestRef(currentResult.stdout.trim());
|
||||
if (currentRef === request.baseManifestRef) {
|
||||
await verifyStable(currentRef);
|
||||
request.journal.commit(currentRef);
|
||||
const stagedResult = request.stagedResult
|
||||
? await workerWorkspaceResultStaging.prepareRequestedWorkerWorkspaceResult({
|
||||
request,
|
||||
stagingRoot,
|
||||
currentManifestRef: currentRef,
|
||||
baseManifestRaw: baseRaw,
|
||||
currentManifestRaw: baseRaw,
|
||||
})
|
||||
: undefined;
|
||||
if (!stagedResult) {
|
||||
request.journal.commit(currentRef);
|
||||
}
|
||||
return {
|
||||
manifestRef: currentRef,
|
||||
changed: false,
|
||||
@@ -597,6 +598,7 @@ export function createWorkerWorkspaceActions(
|
||||
base,
|
||||
current: base,
|
||||
}),
|
||||
...stagedResult,
|
||||
};
|
||||
}
|
||||
const currentDigest = currentRef.slice("sha256:".length);
|
||||
@@ -662,21 +664,33 @@ export function createWorkerWorkspaceActions(
|
||||
// Stop performs this check once more after local acceptance, directly
|
||||
// before destroying the remote owner.
|
||||
await verifyStable(currentRef);
|
||||
await applyStagedWorkerWorkspace({
|
||||
root: request.localPath,
|
||||
stagingRoot,
|
||||
baseManifestRef: request.baseManifestRef,
|
||||
currentManifestRef: currentRef,
|
||||
base,
|
||||
current,
|
||||
journal: request.journal,
|
||||
});
|
||||
const stagedResult = request.stagedResult
|
||||
? await workerWorkspaceResultStaging.prepareRequestedWorkerWorkspaceResult({
|
||||
request,
|
||||
stagingRoot,
|
||||
currentManifestRef: currentRef,
|
||||
baseManifestRaw: baseRaw,
|
||||
currentManifestRaw: currentRaw,
|
||||
})
|
||||
: undefined;
|
||||
if (!stagedResult) {
|
||||
await applyStagedWorkerWorkspace({
|
||||
root: request.localPath,
|
||||
stagingRoot,
|
||||
baseManifestRef: request.baseManifestRef,
|
||||
currentManifestRef: currentRef,
|
||||
base,
|
||||
current,
|
||||
journal: request.journal,
|
||||
});
|
||||
}
|
||||
return {
|
||||
manifestRef: currentRef,
|
||||
changed: true,
|
||||
verifyStable: async () => await verifyStable(currentRef),
|
||||
verifyLocalStable: async () =>
|
||||
await assertWorkspaceResultStable({ root: request.localPath, base, current }),
|
||||
...stagedResult,
|
||||
};
|
||||
} finally {
|
||||
await fs.rm(temporaryDirectory, { recursive: true, force: true }).catch(() => undefined);
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import type { SqliteWalMaintenance } from "../infra/sqlite-wal.js";
|
||||
|
||||
// v4 replaces ambient session-watch sentinel rows with cursor provenance.
|
||||
export const OPENCLAW_STATE_SCHEMA_VERSION = 4;
|
||||
// v5 records durable cloud-worker result refs on pending workspace fences.
|
||||
export const OPENCLAW_STATE_SCHEMA_VERSION = 5;
|
||||
export const OPENCLAW_STATE_STRICT_SCHEMA_VERSION = 3;
|
||||
/** Maximum time one synchronous SQLite call may wait for a lock. */
|
||||
export const OPENCLAW_SQLITE_BUSY_TIMEOUT_MS = 5_000;
|
||||
|
||||
@@ -247,6 +247,7 @@ export function ensureAdditiveStateColumns(db: DatabaseSync): void {
|
||||
"owner_epoch INTEGER NOT NULL DEFAULT 0 CHECK (owner_epoch >= 0)",
|
||||
);
|
||||
ensureColumn(db, "worker_environments", "ssh_host_key TEXT");
|
||||
ensureColumn(db, "worker_workspace_pending_results", "staged_result_ref TEXT");
|
||||
ensureColumn(
|
||||
db,
|
||||
"worker_environments",
|
||||
|
||||
1
src/state/openclaw-state-db.generated.d.ts
vendored
1
src/state/openclaw-state-db.generated.d.ts
vendored
@@ -1237,6 +1237,7 @@ export interface WorkerWorkspacePendingResults {
|
||||
recovery_requested_at_ms: number | null;
|
||||
run_id: string;
|
||||
session_id: string;
|
||||
staged_result_ref: string | null;
|
||||
workspace_accepted_at_ms: number | null;
|
||||
}
|
||||
|
||||
|
||||
@@ -2211,6 +2211,34 @@ INSERT INTO macos_port_guardian_records VALUES (4242, 18789, '/usr/bin/ssh', 're
|
||||
expect(credentialTable?.name).toBe("worker_environment_credentials");
|
||||
});
|
||||
|
||||
it("adds staged worker-result refs during the v5 state migration", () => {
|
||||
const stateDir = createTempStateDir();
|
||||
const options = { env: { OPENCLAW_STATE_DIR: stateDir } };
|
||||
const databasePath = openOpenClawStateDatabase(options).path;
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
|
||||
const { DatabaseSync } = requireNodeSqlite();
|
||||
const legacyDb = new DatabaseSync(databasePath);
|
||||
legacyDb.exec(`
|
||||
ALTER TABLE worker_workspace_pending_results DROP COLUMN staged_result_ref;
|
||||
PRAGMA user_version = 4;
|
||||
UPDATE schema_meta SET schema_version = 4 WHERE meta_key = 'primary';
|
||||
`);
|
||||
legacyDb.close();
|
||||
|
||||
const reopened = openOpenClawStateDatabase(options);
|
||||
const columns = reopened.db
|
||||
.prepare("PRAGMA table_info(worker_workspace_pending_results)")
|
||||
.all() as Array<{ name?: string }>;
|
||||
expect(columns.map((column) => column.name)).toContain("staged_result_ref");
|
||||
expect(readSqliteNumberPragma(reopened.db, "user_version")).toBe(OPENCLAW_STATE_SCHEMA_VERSION);
|
||||
expect(
|
||||
reopened.db
|
||||
.prepare("SELECT schema_version FROM schema_meta WHERE meta_key = 'primary'")
|
||||
.get(),
|
||||
).toEqual({ schema_version: OPENCLAW_STATE_SCHEMA_VERSION });
|
||||
});
|
||||
|
||||
it("adds worker transcript commit tables to existing state databases", () => {
|
||||
const stateDir = createTempStateDir();
|
||||
const database = openOpenClawStateDatabase({
|
||||
|
||||
@@ -1817,6 +1817,7 @@ CREATE TABLE IF NOT EXISTS worker_workspace_pending_results (
|
||||
gateway_instance_id TEXT NOT NULL,
|
||||
recovery_requested_at_ms INTEGER,
|
||||
workspace_accepted_at_ms INTEGER,
|
||||
staged_result_ref TEXT,
|
||||
created_at_ms INTEGER NOT NULL,
|
||||
FOREIGN KEY (session_id) REFERENCES worker_session_placements(session_id) ON DELETE CASCADE
|
||||
) STRICT;
|
||||
|
||||
@@ -1812,6 +1812,7 @@ CREATE TABLE IF NOT EXISTS worker_workspace_pending_results (
|
||||
gateway_instance_id TEXT NOT NULL,
|
||||
recovery_requested_at_ms INTEGER,
|
||||
workspace_accepted_at_ms INTEGER,
|
||||
staged_result_ref TEXT,
|
||||
created_at_ms INTEGER NOT NULL,
|
||||
FOREIGN KEY (session_id) REFERENCES worker_session_placements(session_id) ON DELETE CASCADE
|
||||
) STRICT;
|
||||
|
||||
Reference in New Issue
Block a user