From 7a7d6bb51f42bd896de2b8a4df2ee66f3dce0a21 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 18 Jul 2026 23:24:50 +0100 Subject: [PATCH] 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 --- package.json | 2 +- .../placement-dispatch-failure.ts | 1 + .../placement-dispatch-recovery.ts | 170 +++++- .../placement-dispatch-staged-results.test.ts | 279 +++++++++ .../placement-dispatch-test-harness.ts | 284 +++++++++ .../placement-dispatch.test.ts | 2 + .../placement-force-abandon.test.ts | 35 ++ .../placement-force-abandon.ts | 48 +- .../placement-store.test.ts | 9 + .../placement-workspace-result.ts | 31 + .../worker-environments/tunnel-contract.ts | 9 + .../worker-environments/tunnel.test.ts | 97 +++ .../worker-turn-launcher.test.ts | 11 +- .../worker-turn-launcher.ts | 25 +- .../workspace-finalize.test.ts | 144 ++++- .../worker-environments/workspace-finalize.ts | 19 + .../workspace-reconcile.test.ts | 238 +++++++- .../workspace-reconcile.ts | 24 +- .../workspace-result-staging.ts | 571 ++++++++++++++++++ .../workspace-sync-helpers.ts | 40 ++ .../worker-environments/workspace-sync.ts | 76 ++- src/state/openclaw-state-db-contract.ts | 4 +- .../openclaw-state-db-schema-additive.ts | 1 + src/state/openclaw-state-db.generated.d.ts | 1 + src/state/openclaw-state-db.test.ts | 28 + src/state/openclaw-state-schema.generated.ts | 1 + src/state/openclaw-state-schema.sql | 1 + 27 files changed, 2077 insertions(+), 74 deletions(-) create mode 100644 src/gateway/worker-environments/placement-dispatch-staged-results.test.ts create mode 100644 src/gateway/worker-environments/placement-dispatch-test-harness.ts create mode 100644 src/gateway/worker-environments/workspace-result-staging.ts diff --git a/package.json b/package.json index 40edfaabf6ad..d4d57a30dfe4 100644 --- a/package.json +++ b/package.json @@ -3,7 +3,7 @@ "version": "2026.7.2", "openclaw": { "schemaVersions": { - "state": 4, + "state": 5, "agent": 13 } }, diff --git a/src/gateway/worker-environments/placement-dispatch-failure.ts b/src/gateway/worker-environments/placement-dispatch-failure.ts index 63b66ea7b450..4ec4e128a6e2 100644 --- a/src/gateway/worker-environments/placement-dispatch-failure.ts +++ b/src/gateway/worker-environments/placement-dispatch-failure.ts @@ -39,6 +39,7 @@ export type WorkerDispatchPlacementStore = Pick< | "listWorkspaceReconciliationOwners" | "listPendingWorkspaceResults" | "workspaceResultInstanceId" + | "recordStagedWorkspaceResult" | "acceptWorkspaceResult" | "completeWorkspaceResultAndReleaseTurn" | "abandonWorkspaceResult" diff --git a/src/gateway/worker-environments/placement-dispatch-recovery.ts b/src/gateway/worker-environments/placement-dispatch-recovery.ts index 90957a01eb7a..6ba1665fd502 100644 --- a/src/gateway/worker-environments/placement-dispatch-recovery.ts +++ b/src/gateway/worker-environments/placement-dispatch-recovery.ts @@ -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> => { + const stagedResultOwners = new Set(); 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[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[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 => { diff --git a/src/gateway/worker-environments/placement-dispatch-staged-results.test.ts b/src/gateway/worker-environments/placement-dispatch-staged-results.test.ts new file mode 100644 index 000000000000..8c7e1baacd4b --- /dev/null +++ b/src/gateway/worker-environments/placement-dispatch-staged-results.test.ts @@ -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; + 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 }); + }); +}); diff --git a/src/gateway/worker-environments/placement-dispatch-test-harness.ts b/src/gateway/worker-environments/placement-dispatch-test-harness.ts new file mode 100644 index 000000000000..3ee61f4897b1 --- /dev/null +++ b/src/gateway/worker-environments/placement-dispatch-test-harness.ts @@ -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 = 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, + }; +} diff --git a/src/gateway/worker-environments/placement-dispatch.test.ts b/src/gateway/worker-environments/placement-dispatch.test.ts index 9a28b41273b8..b18fdd82c15a 100644 --- a/src/gateway/worker-environments/placement-dispatch.test.ts +++ b/src/gateway/worker-environments/placement-dispatch.test.ts @@ -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), diff --git a/src/gateway/worker-environments/placement-force-abandon.test.ts b/src/gateway/worker-environments/placement-force-abandon.test.ts index 72f40768d48e..80e3219bfbe8 100644 --- a/src/gateway/worker-environments/placement-force-abandon.test.ts +++ b/src/gateway/worker-environments/placement-force-abandon.test.ts @@ -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([]); + }); }); diff --git a/src/gateway/worker-environments/placement-force-abandon.ts b/src/gateway/worker-environments/placement-force-abandon.ts index f53d8263c5a3..36347ce4b320 100644 --- a/src/gateway/worker-environments/placement-force-abandon.ts +++ b/src/gateway/worker-environments/placement-force-abandon.ts @@ -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, + placement: { sessionId: string; sessionKey: string; agentId: string }, +): Promise { + 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); } } diff --git a/src/gateway/worker-environments/placement-store.test.ts b/src/gateway/worker-environments/placement-store.test.ts index 3b54d3edd4f1..a9f48d2f3592 100644 --- a/src/gateway/worker-environments/placement-store.test.ts +++ b/src/gateway/worker-environments/placement-store.test.ts @@ -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 }, diff --git a/src/gateway/worker-environments/placement-workspace-result.ts b/src/gateway/worker-environments/placement-workspace-result.ts index 577600f87fc5..06830ff67c49 100644 --- a/src/gateway/worker-environments/placement-workspace-result.ts +++ b/src/gateway/worker-environments/placement-workspace-result.ts @@ -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); diff --git a/src/gateway/worker-environments/tunnel-contract.ts b/src/gateway/worker-environments/tunnel-contract.ts index 1d6f738c49bb..ff7aa392d6ff 100644 --- a/src/gateway/worker-environments/tunnel-contract.ts +++ b/src/gateway/worker-environments/tunnel-contract.ts @@ -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; /** Re-read the accepted local result after the remote stability fence. */ verifyLocalStable(): Promise; + /** Apply the prepared candidate locally without making it restart-authoritative. */ + applyPreparedStagedResult?(): Promise; + /** Publish the verified candidate for restart recovery. */ + publishStagedResult?(): Promise; + discardPreparedStagedResult?(): Promise; }; export type WorkerWorkspaceQuiescence = { diff --git a/src/gateway/worker-environments/tunnel.test.ts b/src/gateway/worker-environments/tunnel.test.ts index d90e01684f03..1be9a3d62ddd 100644 --- a/src/gateway/worker-environments/tunnel.test.ts +++ b/src/gateway/worker-environments/tunnel.test.ts @@ -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"), diff --git a/src/gateway/worker-environments/worker-turn-launcher.test.ts b/src/gateway/worker-environments/worker-turn-launcher.test.ts index cf0cb0361aa4..05d9519d0256 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.test.ts @@ -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, diff --git a/src/gateway/worker-environments/worker-turn-launcher.ts b/src/gateway/worker-environments/worker-turn-launcher.ts index 9b91b45ec5cb..eef074060e16 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.ts @@ -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); diff --git a/src/gateway/worker-environments/workspace-finalize.test.ts b/src/gateway/worker-environments/workspace-finalize.test.ts index 54f3f03e939d..e4210afc4f94 100644 --- a/src/gateway/worker-environments/workspace-finalize.test.ts +++ b/src/gateway/worker-environments/workspace-finalize.test.ts @@ -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(); + }); }); diff --git a/src/gateway/worker-environments/workspace-finalize.ts b/src/gateway/worker-environments/workspace-finalize.ts index 924c60266769..021ea49582c4 100644 --- a/src/gateway/worker-environments/workspace-finalize.ts +++ b/src/gateway/worker-environments/workspace-finalize.ts @@ -8,6 +8,25 @@ export async function verifyReconciledWorkspaceFinal( reconciliation: WorkerWorkspaceReconcileResult, quiescence: WorkerWorkspaceQuiescence, ): Promise { + 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(); diff --git a/src/gateway/worker-environments/workspace-reconcile.test.ts b/src/gateway/worker-environments/workspace-reconcile.test.ts index 385b69d7bae4..2c6c3745e1d5 100644 --- a/src/gateway/worker-environments/workspace-reconcile.test.ts +++ b/src/gateway/worker-environments/workspace-reconcile.test.ts @@ -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"); diff --git a/src/gateway/worker-environments/workspace-reconcile.ts b/src/gateway/worker-environments/workspace-reconcile.ts index d557b020afc8..737f8159ee9f 100644 --- a/src/gateway/worker-environments/workspace-reconcile.ts +++ b/src/gateway/worker-environments/workspace-reconcile.ts @@ -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; diff --git a/src/gateway/worker-environments/workspace-result-staging.ts b/src/gateway/worker-environments/workspace-result-staging.ts new file mode 100644 index 000000000000..15247dec3001 --- /dev/null +++ b/src/gateway/worker-environments/workspace-result-staging.ts @@ -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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 0 +0000\ncommitter OpenClaw 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; +}; + +async function loadStagedWorkerWorkspace( + root: string, + stagedResultRef: string, +): Promise { + 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(); + 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 { + 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 }> { + 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; + publishStagedResult(): Promise; + discardPreparedStagedResult(): Promise; +}> { + 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 { + 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, +}; diff --git a/src/gateway/worker-environments/workspace-sync-helpers.ts b/src/gateway/worker-environments/workspace-sync-helpers.ts index 71941f8dd67f..a3305cb3a3e7 100644 --- a/src/gateway/worker-environments/workspace-sync-helpers.ts +++ b/src/gateway/worker-environments/workspace-sync-helpers.ts @@ -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; +}): 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); } diff --git a/src/gateway/worker-environments/workspace-sync.ts b/src/gateway/worker-environments/workspace-sync.ts index 01af187d9e06..5bb192b1fe80 100644 --- a/src/gateway/worker-environments/workspace-sync.ts +++ b/src/gateway/worker-environments/workspace-sync.ts @@ -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); diff --git a/src/state/openclaw-state-db-contract.ts b/src/state/openclaw-state-db-contract.ts index 8373e6d6f391..42b5ace5bfb4 100644 --- a/src/state/openclaw-state-db-contract.ts +++ b/src/state/openclaw-state-db-contract.ts @@ -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; diff --git a/src/state/openclaw-state-db-schema-additive.ts b/src/state/openclaw-state-db-schema-additive.ts index b7b80ecfb935..b022d86f4b90 100644 --- a/src/state/openclaw-state-db-schema-additive.ts +++ b/src/state/openclaw-state-db-schema-additive.ts @@ -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", diff --git a/src/state/openclaw-state-db.generated.d.ts b/src/state/openclaw-state-db.generated.d.ts index 0f7559179449..55103a23cf94 100644 --- a/src/state/openclaw-state-db.generated.d.ts +++ b/src/state/openclaw-state-db.generated.d.ts @@ -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; } diff --git a/src/state/openclaw-state-db.test.ts b/src/state/openclaw-state-db.test.ts index 3a907cd4feb9..af4022c5b6a4 100644 --- a/src/state/openclaw-state-db.test.ts +++ b/src/state/openclaw-state-db.test.ts @@ -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({ diff --git a/src/state/openclaw-state-schema.generated.ts b/src/state/openclaw-state-schema.generated.ts index ddbd5be2e037..3a0473dc4c3e 100644 --- a/src/state/openclaw-state-schema.generated.ts +++ b/src/state/openclaw-state-schema.generated.ts @@ -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; diff --git a/src/state/openclaw-state-schema.sql b/src/state/openclaw-state-schema.sql index e7f1d957890d..024edb46a876 100644 --- a/src/state/openclaw-state-schema.sql +++ b/src/state/openclaw-state-schema.sql @@ -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;