diff --git a/src/gateway/worker-environments/placement-dispatch-pending-results.ts b/src/gateway/worker-environments/placement-dispatch-pending-results.ts index 3a031c5eda6d..9ba22d309823 100644 --- a/src/gateway/worker-environments/placement-dispatch-pending-results.ts +++ b/src/gateway/worker-environments/placement-dispatch-pending-results.ts @@ -11,8 +11,13 @@ import { completeMovedWorkspaceTeardown, completeReclaimedWorkspaceTeardown, } from "./placement-teardown.js"; +import { isCurrentWorkerWorkspacePendingResultOwner } from "./placement-workspace-result.js"; import type { WorkerEnvironmentService } from "./service.js"; -import type { WorkerWorkspaceResultConflict } from "./workspace-conflicts.js"; +import { boundedWorkerError } from "./worker-error.js"; +import type { + WorkerWorkspaceRecoveryFailureReport, + WorkerWorkspaceResultConflict, +} from "./workspace-conflicts.js"; import { verifyReconciledWorkspaceFinal } from "./workspace-finalize.js"; import type { WorkerWorkspaceOperationCoordinator } from "./workspace-operation-coordinator.js"; import { recoverWorkerWorkspaceReconciliation } from "./workspace-reconcile.js"; @@ -48,6 +53,9 @@ export type PlacementRecoveryDeps = { | { cleared: true } ), ) => Promise; + reportWorkspaceResultRecoveryFailure?: ( + recovery: WorkerWorkspaceRecoveryFailureReport, + ) => Promise; resolveWorkspaceResultConflict: (params: { sessionId: string; sessionKey: string; @@ -480,8 +488,32 @@ export async function recoverPendingWorkspaceResults( } } }); - } catch { - // Keep the result, claim, and environment fenced. The next sweep retries. + } catch (error) { + try { + const current = placements.get(pending.sessionId); + const currentPending = placements + .listPendingWorkspaceResults() + .find( + (candidate) => + candidate.sessionId === pending.sessionId && + candidate.environmentId === pending.environmentId && + candidate.ownerEpoch === pending.ownerEpoch && + candidate.placementGeneration === pending.placementGeneration && + candidate.claimId === pending.claimId && + candidate.runId === pending.runId && + candidate.gatewayInstanceId === pending.gatewayInstanceId, + ); + if (currentPending && isCurrentWorkerWorkspacePendingResultOwner(current, currentPending)) { + await deps.reportWorkspaceResultRecoveryFailure?.({ + sessionId: current.sessionId, + sessionKey: current.sessionKey, + agentId: current.agentId, + error: boundedWorkerError(error), + }); + } + } catch { + // Transcript reporting must not weaken the durable recovery fence. + } } } if (cleanupOrphans) { diff --git a/src/gateway/worker-environments/placement-dispatch-staged-results.test.ts b/src/gateway/worker-environments/placement-dispatch-staged-results.test.ts index 726bd141729a..d87a1aab45ea 100644 --- a/src/gateway/worker-environments/placement-dispatch-staged-results.test.ts +++ b/src/gateway/worker-environments/placement-dispatch-staged-results.test.ts @@ -505,9 +505,21 @@ describe("staged worker placement result recovery", () => { 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(); - restartedHarness.reportWorkspaceResultConflict.mockRejectedValueOnce( - new Error("transcript report interrupted"), + restartedHarness.markEnvironmentOwnerEpoch(active.activeOwnerEpoch); + const secret = [ + String.fromCharCode(115, 107), + "proj", + "recovery", + "abcdefghijklmnopqrstuvwxyz", + ].join("-"); + const failure = new Error( + `transcript report interrupted token=${secret} ${"detail ".repeat(200)}`, + ); + restartedHarness.reportWorkspaceResultConflict + .mockRejectedValueOnce(failure) + .mockRejectedValueOnce(failure); + restartedHarness.reportWorkspaceResultRecoveryFailure.mockRejectedValueOnce( + new Error("recovery transcript temporarily unavailable"), ); await restartedHarness.service.reconcile(); @@ -515,6 +527,27 @@ describe("staged worker placement result recovery", () => { expect(restartedStore.listPendingWorkspaceResults()).toMatchObject([ { stagedResultRef: staged.stagedResultRef, workspaceAcceptedAtMs: 2_000 }, ]); + expect(restartedStore.get(active.sessionId)).toMatchObject({ + state: "draining", + turnClaim: { claimId: claim.claimId, runId: claim.runId }, + }); + expect(restartedHarness.environments.destroy).not.toHaveBeenCalled(); + expect(restartedHarness.reportWorkspaceResultRecoveryFailure).toHaveBeenCalledOnce(); + const recovery = restartedHarness.reportWorkspaceResultRecoveryFailure.mock.calls[0]?.[0]; + expect(recovery).toMatchObject({ + sessionId: active.sessionId, + sessionKey: active.sessionKey, + agentId: active.agentId, + error: expect.stringContaining("transcript report interrupted"), + }); + expect(JSON.stringify(recovery)).not.toContain(secret); + expect(recovery?.error.length).toBeLessThanOrEqual(1_024); + + await restartedHarness.service.reconcile(); + + expect(restartedHarness.reportWorkspaceResultRecoveryFailure).toHaveBeenCalledTimes(2); + expect(restartedStore.listPendingWorkspaceResults()).toHaveLength(1); + expect(restartedHarness.environments.destroy).not.toHaveBeenCalled(); expect( await runCommandWithTimeout( ["git", "-C", workspacePath, "show-ref", "--verify", staged.stagedResultRef], @@ -524,7 +557,7 @@ describe("staged worker placement result recovery", () => { await fs.writeFile(path.join(workspacePath, "result.txt"), "later local edit\n"); const finalStore = createWorkerSessionPlacementStore({ database, now: () => 3_000 }); const finalHarness = createHarness(finalStore, { workspacePath }); - finalHarness.markEnvironmentDestroyed(); + finalHarness.markEnvironmentOwnerEpoch(active.activeOwnerEpoch); await finalHarness.service.reconcile(); @@ -540,6 +573,8 @@ describe("staged worker placement result recovery", () => { expect(recovered?.workspaceBaseManifestRef).not.toBe(staged.currentManifestRef); expect(finalStore.listPendingWorkspaceResults()).toEqual([]); expect(finalHarness.environments.startTunnel).not.toHaveBeenCalled(); + expect(finalHarness.environments.destroy).toHaveBeenCalledWith(active.environmentId); + expect(finalHarness.reportWorkspaceResultRecoveryFailure).not.toHaveBeenCalled(); expect(finalHarness.log).not.toContain("placement:failed"); expect(finalHarness.reportWorkspaceResultConflict).toHaveBeenCalledWith({ sessionId: REQUEST.sessionId, diff --git a/src/gateway/worker-environments/placement-dispatch-test-harness.ts b/src/gateway/worker-environments/placement-dispatch-test-harness.ts index 93b3086a1204..fdb6eff9d0c1 100644 --- a/src/gateway/worker-environments/placement-dispatch-test-harness.ts +++ b/src/gateway/worker-environments/placement-dispatch-test-harness.ts @@ -20,6 +20,7 @@ import { createWorkerPlacementRunnerAvailabilityReader } from "./placement-proje import { completeReclaimedWorkspaceTeardown } from "./placement-teardown.js"; import { WorkerTunnelOwnerDisconnectedError } from "./tunnel-contract.js"; import type { WorkerTunnelHandle } from "./tunnel.js"; +import type { WorkerWorkspaceRecoveryFailureReport } from "./workspace-conflicts.js"; import { createWorkerWorkspaceOperationCoordinator, type WorkerWorkspaceOperationCoordinator, @@ -74,6 +75,9 @@ export function createHarness( let verifyCalls = 0; const log: string[] = []; const reportWorkspaceResultConflict = vi.fn(async () => {}); + const reportWorkspaceResultRecoveryFailure = vi.fn( + async (_recovery: WorkerWorkspaceRecoveryFailureReport) => {}, + ); const fail = (stage: DispatchStage) => { log.push(stage); if (options.failAt === stage) { @@ -456,6 +460,7 @@ export function createHarness( return options.workspacePath ?? "/gateway/workspace"; }, reportWorkspaceResultConflict, + reportWorkspaceResultRecoveryFailure, resolveWorkspaceResultConflict: vi.fn(async () => options.priorWorkspaceResultConflict), ...(options.prepareAcceptedWorkspacePublication ? { prepareAcceptedWorkspacePublication: options.prepareAcceptedWorkspacePublication } @@ -489,6 +494,7 @@ export function createHarness( }, environments, reportWorkspaceResultConflict, + reportWorkspaceResultRecoveryFailure, markEnvironmentDestroyed: () => { currentEnvironment = destroyedEnvironment((currentEnvironment?.ownerEpoch ?? 1) + 1); }, diff --git a/src/gateway/worker-environments/placement-dispatch.ts b/src/gateway/worker-environments/placement-dispatch.ts index dffee970d14e..10e43e9793c2 100644 --- a/src/gateway/worker-environments/placement-dispatch.ts +++ b/src/gateway/worker-environments/placement-dispatch.ts @@ -37,7 +37,10 @@ import type { import { deriveEnvironmentIntent } from "./service-contract.js"; import { isFailedWorkerPlacementEnvironmentGone } from "./session-placement-lifecycle.js"; import { WorkerTunnelOwnerDisconnectedError } from "./tunnel-contract.js"; -import type { WorkerWorkspaceResultConflict } from "./workspace-conflicts.js"; +import type { + WorkerWorkspaceRecoveryFailureReport, + WorkerWorkspaceResultConflict, +} from "./workspace-conflicts.js"; import { verifyReconciledWorkspaceFinal, WorkerWorkspaceFinalFenceError, @@ -115,6 +118,9 @@ type WorkerPlacementDispatchOptions = WorkerPlacementReclaimBarriers & { | { cleared: true } ), ) => Promise; + reportWorkspaceResultRecoveryFailure?: ( + recovery: WorkerWorkspaceRecoveryFailureReport, + ) => Promise; resolveWorkspaceResultConflict: (params: { sessionId: string; sessionKey: string; @@ -178,6 +184,9 @@ export function createWorkerPlacementDispatchService(options: WorkerPlacementDis placements, resolveWorkspacePath: options.resolveWorkspacePath, reportWorkspaceResultConflict: options.reportWorkspaceResultConflict, + ...(options.reportWorkspaceResultRecoveryFailure + ? { reportWorkspaceResultRecoveryFailure: options.reportWorkspaceResultRecoveryFailure } + : {}), resolveWorkspaceResultConflict: options.resolveWorkspaceResultConflict, recoverPlacementMoves: () => recoverPlacementMoves(), workspaceOperations: options.workspaceOperations, diff --git a/src/gateway/worker-environments/worker-turn-admission.ts b/src/gateway/worker-environments/worker-turn-admission.ts index a8c81f84fb0c..6120f59d0c73 100644 --- a/src/gateway/worker-environments/worker-turn-admission.ts +++ b/src/gateway/worker-environments/worker-turn-admission.ts @@ -150,7 +150,7 @@ export function requireActivePlacement( ): ActiveWorkerPlacement { const failureDetail = placement.state === "failed" - ? `: ${withCurrentWorkerBuildRemediation(placement.terminalReason ?? placement.recoveryError)}` + ? `: ${withCurrentWorkerBuildRemediation(placement.recoveryError)}` : ""; if ( placement.state !== "active" || diff --git a/src/gateway/worker-environments/worker-turn-launcher-reclaimed-placement.test.ts b/src/gateway/worker-environments/worker-turn-launcher-reclaimed-placement.test.ts index 81cf15f416b9..78f68f3832b6 100644 --- a/src/gateway/worker-environments/worker-turn-launcher-reclaimed-placement.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher-reclaimed-placement.test.ts @@ -495,6 +495,10 @@ describe("worker turn launcher reclaimed placement", () => { sessionKey: SESSION_KEY, agentId: "main", }); + placements.fail({ + sessionId: SESSION_ID, + recoveryError: "stale terminal worker failure", + }); placements.fail({ sessionId: SESSION_ID, recoveryError: "cloud worker disappeared: environment state destroyed", diff --git a/src/gateway/worker-environments/workspace-conflicts.ts b/src/gateway/worker-environments/workspace-conflicts.ts index eb1652ef2d67..9b14dca6f32d 100644 --- a/src/gateway/worker-environments/workspace-conflicts.ts +++ b/src/gateway/worker-environments/workspace-conflicts.ts @@ -4,8 +4,16 @@ export type WorkerWorkspaceResultConflict = { totalCount?: number; }; +export type WorkerWorkspaceRecoveryFailureReport = { + sessionId: string; + sessionKey: string; + agentId: string; + error: string; +}; + export const WORKSPACE_CONFLICT_TRANSCRIPT_TYPE = "cloud-workspace-conflict"; export const WORKSPACE_CONFLICT_CLEARED_TRANSCRIPT_TYPE = "cloud-workspace-conflict-cleared"; +export const WORKSPACE_RECOVERY_FAILURE_TRANSCRIPT_TYPE = "cloud-workspace-recovery-failed"; const MAX_PROJECTED_CONFLICT_PATHS = 256; const MAX_PROJECTED_CONFLICT_PATH_BYTES = 32 * 1024; diff --git a/src/gateway/worker-workspace-conflict-transcript.ts b/src/gateway/worker-workspace-conflict-transcript.ts index 4b8b2632106e..3277f9057c23 100644 --- a/src/gateway/worker-workspace-conflict-transcript.ts +++ b/src/gateway/worker-workspace-conflict-transcript.ts @@ -1,11 +1,14 @@ import { SessionManager } from "../agents/sessions/session-manager.js"; import { getRuntimeConfig } from "../config/config.js"; import { withTranscriptWriteTransaction } from "../config/sessions/session-accessor.js"; +import { boundedWorkerError } from "./worker-environments/worker-error.js"; import { formatWorkspaceConflictSummary, projectWorkspaceResultConflict, WORKSPACE_CONFLICT_CLEARED_TRANSCRIPT_TYPE, WORKSPACE_CONFLICT_TRANSCRIPT_TYPE, + WORKSPACE_RECOVERY_FAILURE_TRANSCRIPT_TYPE, + type WorkerWorkspaceRecoveryFailureReport, } from "./worker-environments/workspace-conflicts.js"; export function createWorkerWorkspaceConflictTranscriptHandlers( @@ -14,117 +17,110 @@ export function createWorkerWorkspaceConflictTranscriptHandlers( resolveGatewaySessionStoreTargetWithStore: typeof import("./session-utils.js").resolveGatewaySessionStoreTargetWithStore; }>, ) { + async function withWorkerTranscript( + identity: Pick, + run: (manager: SessionManager) => T, + missingMessage?: string, + strictIdentity = false, + ): Promise { + const runtime = await loadSessionRuntime(); + const target = runtime.resolveGatewaySessionStoreTargetWithStore({ + cfg: getRuntimeConfig(), + key: identity.sessionKey, + agentId: identity.agentId, + clone: false, + }); + return await withTranscriptWriteTransaction( + { + agentId: target.agentId, + sessionId: identity.sessionId, + sessionKey: target.canonicalKey, + storePath: target.storePath, + }, + (transcriptTarget) => { + const entry = runtime.resolveCanonicalSessionEntryFromStoreKeys( + target.store, + target.storeKeys, + ); + if ( + entry?.sessionId !== identity.sessionId || + (strictIdentity && + (target.canonicalKey !== identity.sessionKey || target.agentId !== identity.agentId)) + ) { + if (missingMessage) { + throw new Error(`${missingMessage} lost session ${identity.sessionId}`); + } + return undefined; + } + return run(SessionManager.open(transcriptTarget)); + }, + ); + } + + function latestWorkspaceReport(manager: SessionManager, ...customTypes: string[]) { + for (const entry of manager.getBranch().toReversed()) { + if (entry.type === "custom_message" && customTypes.includes(entry.customType)) { + return entry; + } + } + return undefined; + } + return { resolveWorkspaceResultConflict: async (identity: { sessionId: string; sessionKey: string; agentId: string; - }) => { - const { - resolveCanonicalSessionEntryFromStoreKeys, - resolveGatewaySessionStoreTargetWithStore, - } = await loadSessionRuntime(); - const target = resolveGatewaySessionStoreTargetWithStore({ - cfg: getRuntimeConfig(), - key: identity.sessionKey, - agentId: identity.agentId, - clone: false, - }); - const entry = resolveCanonicalSessionEntryFromStoreKeys(target.store, target.storeKeys); - if (entry?.sessionId !== identity.sessionId) { - return undefined; - } - return await withTranscriptWriteTransaction( - { - agentId: target.agentId, - sessionId: identity.sessionId, - sessionKey: target.canonicalKey, - storePath: target.storePath, - }, - (transcriptTarget) => { - for (const transcriptEntry of SessionManager.open(transcriptTarget) - .getBranch() - .toReversed()) { - if (transcriptEntry.type !== "custom_message") { - continue; - } - if (transcriptEntry.customType === WORKSPACE_CONFLICT_CLEARED_TRANSCRIPT_TYPE) { - return undefined; - } - if (transcriptEntry.customType !== WORKSPACE_CONFLICT_TRANSCRIPT_TYPE) { - continue; - } - const details = transcriptEntry.details as - | { paths?: unknown; stagedResultRef?: unknown; totalCount?: unknown } - | undefined; - if ( - Array.isArray(details?.paths) && - details.paths.length > 0 && - details.paths.every( - (entryPath): entryPath is string => - typeof entryPath === "string" && entryPath.length > 0, - ) && - typeof details.stagedResultRef === "string" && - (details.totalCount === undefined || - (Number.isSafeInteger(details.totalCount) && - (details.totalCount as number) >= details.paths.length)) && - /^refs\/openclaw\/worker-results\/[A-Za-z0-9-]+$/u.test(details.stagedResultRef) - ) { - return projectWorkspaceResultConflict( - details.paths, - details.stagedResultRef, - details.totalCount as number | undefined, - ); - } - return undefined; - } + }) => + await withWorkerTranscript(identity, (manager) => { + const transcriptEntry = latestWorkspaceReport( + manager, + WORKSPACE_CONFLICT_TRANSCRIPT_TYPE, + WORKSPACE_CONFLICT_CLEARED_TRANSCRIPT_TYPE, + ); + if (transcriptEntry?.customType !== WORKSPACE_CONFLICT_TRANSCRIPT_TYPE) { return undefined; - }, - ); - }, + } + const details = transcriptEntry.details as + | { paths?: unknown; stagedResultRef?: unknown; totalCount?: unknown } + | undefined; + if ( + Array.isArray(details?.paths) && + details.paths.length > 0 && + details.paths.every( + (entryPath): entryPath is string => + typeof entryPath === "string" && entryPath.length > 0, + ) && + typeof details.stagedResultRef === "string" && + (details.totalCount === undefined || + (Number.isSafeInteger(details.totalCount) && + (details.totalCount as number) >= details.paths.length)) && + /^refs\/openclaw\/worker-results\/[A-Za-z0-9-]+$/u.test(details.stagedResultRef) + ) { + return projectWorkspaceResultConflict( + details.paths, + details.stagedResultRef, + details.totalCount as number | undefined, + ); + } + return undefined; + }), reportWorkspaceResultConflict: async ( conflict: { sessionId: string; sessionKey: string; agentId: string } & ( | { paths: string[]; stagedResultRef: string; totalCount: number } | { cleared: true } ), ) => { - const { - resolveCanonicalSessionEntryFromStoreKeys, - resolveGatewaySessionStoreTargetWithStore, - } = await loadSessionRuntime(); - const target = resolveGatewaySessionStoreTargetWithStore({ - cfg: getRuntimeConfig(), - key: conflict.sessionKey, - agentId: conflict.agentId, - clone: false, - }); - const entry = resolveCanonicalSessionEntryFromStoreKeys(target.store, target.storeKeys); - if (entry?.sessionId !== conflict.sessionId) { - throw new Error(`Recovered cloud workspace conflict lost session ${conflict.sessionId}`); - } - await withTranscriptWriteTransaction( - { - agentId: target.agentId, - sessionId: conflict.sessionId, - sessionKey: target.canonicalKey, - storePath: target.storePath, - }, - (transcriptTarget) => { - const manager = SessionManager.open(transcriptTarget); - const latestConflictEntry = manager - .getBranch() - .toReversed() - .find( - (transcriptEntry) => - transcriptEntry.type === "custom_message" && - (transcriptEntry.customType === WORKSPACE_CONFLICT_TRANSCRIPT_TYPE || - transcriptEntry.customType === WORKSPACE_CONFLICT_CLEARED_TRANSCRIPT_TYPE), - ); + await withWorkerTranscript( + conflict, + (manager) => { + const latestConflictEntry = latestWorkspaceReport( + manager, + WORKSPACE_CONFLICT_TRANSCRIPT_TYPE, + WORKSPACE_CONFLICT_CLEARED_TRANSCRIPT_TYPE, + ); if ("cleared" in conflict) { - if ( - latestConflictEntry?.type !== "custom_message" || - latestConflictEntry.customType !== WORKSPACE_CONFLICT_CLEARED_TRANSCRIPT_TYPE - ) { + if (latestConflictEntry?.customType !== WORKSPACE_CONFLICT_CLEARED_TRANSCRIPT_TYPE) { manager.appendCustomMessageEntry( WORKSPACE_CONFLICT_CLEARED_TRANSCRIPT_TYPE, "A later cloud workspace result superseded the previous conflict.", @@ -138,15 +134,11 @@ export function createWorkerWorkspaceConflictTranscriptHandlers( conflict.stagedResultRef, conflict.totalCount, ); - const details = - latestConflictEntry?.type === "custom_message" - ? (latestConflictEntry.details as - | { paths?: unknown; stagedResultRef?: unknown; totalCount?: unknown } - | undefined) - : undefined; + const details = latestConflictEntry?.details as + | { paths?: unknown; stagedResultRef?: unknown; totalCount?: unknown } + | undefined; const alreadyReported = - latestConflictEntry?.type === "custom_message" && - latestConflictEntry.customType === WORKSPACE_CONFLICT_TRANSCRIPT_TYPE && + latestConflictEntry?.customType === WORKSPACE_CONFLICT_TRANSCRIPT_TYPE && details?.stagedResultRef === projectedConflict.stagedResultRef && details.totalCount === projectedConflict.totalCount && Array.isArray(details.paths) && @@ -164,6 +156,32 @@ export function createWorkerWorkspaceConflictTranscriptHandlers( ); } }, + "Recovered cloud workspace conflict", + ); + }, + reportWorkspaceResultRecoveryFailure: async ( + recovery: WorkerWorkspaceRecoveryFailureReport, + ) => { + await withWorkerTranscript( + recovery, + (manager) => { + const latestRecovery = latestWorkspaceReport( + manager, + WORKSPACE_RECOVERY_FAILURE_TRANSCRIPT_TYPE, + ); + const error = boundedWorkerError(recovery.error, 768); + const content = `Cloud workspace recovery attempt failed: ${error}. OpenClaw preserved the result and will retry.`; + if (latestRecovery?.content !== content) { + manager.appendCustomMessageEntry( + WORKSPACE_RECOVERY_FAILURE_TRANSCRIPT_TYPE, + content, + true, + { error }, + ); + } + }, + "Cloud workspace recovery", + true, ); }, }; diff --git a/src/gateway/worker-workspace-recovery-transcript.test.ts b/src/gateway/worker-workspace-recovery-transcript.test.ts new file mode 100644 index 000000000000..21981a0cfe79 --- /dev/null +++ b/src/gateway/worker-workspace-recovery-transcript.test.ts @@ -0,0 +1,231 @@ +import fs from "node:fs/promises"; +import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import { afterEach, describe, expect, it } from "vitest"; +import { + loadTranscriptEvents, + upsertSessionEntryCore, +} from "../config/sessions/session-accessor.js"; +import { runExclusiveSqliteSessionWrite } from "../config/sessions/session-accessor.sqlite-scope.js"; +import { runCommandWithTimeout } from "../process/exec.js"; +import { closeOpenClawAgentDatabasesForTest } from "../state/openclaw-agent-db.js"; +import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js"; +import { withOpenClawTestState } from "../test-utils/openclaw-test-state.js"; +import { + REQUEST, + type DispatchStage, +} from "./worker-environments/placement-dispatch-test-fixtures.js"; +import { createHarness } from "./worker-environments/placement-dispatch-test-harness.js"; +import { createWorkerSessionPlacementStore } from "./worker-environments/placement-store.js"; +import { WORKSPACE_RECOVERY_FAILURE_TRANSCRIPT_TYPE } from "./worker-environments/workspace-conflicts.js"; +import { createWorkerWorkspaceConflictTranscriptHandlers } from "./worker-workspace-conflict-transcript.js"; + +const IDENTITY = { + agentId: "main", + sessionId: "workspace-recovery-session", + sessionKey: "agent:main:main", +}; + +afterEach(() => { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); +}); + +function loadSessionRuntime() { + return import("./session-utils.js"); +} + +async function readRecoveryEvents(identity = IDENTITY) { + const events = await loadTranscriptEvents(identity); + return events.filter( + (event): event is Record => + isRecord(event) && + event.type === "custom_message" && + event.customType === WORKSPACE_RECOVERY_FAILURE_TRANSCRIPT_TYPE, + ); +} + +describe("worker workspace recovery transcript reporting", () => { + it("records historical recovery failures while preserving the live pending-result owner", async () => { + await withOpenClawTestState({ scenario: "minimal" }, async (state) => { + await upsertSessionEntryCore(REQUEST, { sessionId: REQUEST.sessionId, updatedAt: 1 }); + const workspacePath = state.statePath("recovery-workspace"); + await fs.mkdir(workspacePath, { recursive: true }); + expect( + ( + await runCommandWithTimeout(["git", "-C", workspacePath, "init", "--quiet"], { + timeoutMs: 10_000, + }) + ).code, + ).toBe(0); + const placements = createWorkerSessionPlacementStore(); + const harnessOptions: { failAt?: DispatchStage; workspacePath: string } = { + failAt: "workspace", + workspacePath, + }; + const harness = createHarness(placements, harnessOptions); + const active = harness.placements.seedActive(2); + if (active.state !== "active") { + throw new Error("expected active worker placement"); + } + harness.markEnvironmentOwnerEpoch(active.activeOwnerEpoch); + const claim = placements.claimTurn({ + ...REQUEST, + claimId: "workspace-recovery-claim", + runId: "workspace-recovery-run", + owner: { + kind: "worker", + environmentId: active.environmentId, + ownerEpoch: active.activeOwnerEpoch, + }, + }); + placements.markWorkspaceResultPending(claim); + placements.handoffWorkspaceResultRecovery(claim); + const { reportWorkspaceResultRecoveryFailure } = + createWorkerWorkspaceConflictTranscriptHandlers(loadSessionRuntime); + harness.reportWorkspaceResultRecoveryFailure.mockImplementation( + reportWorkspaceResultRecoveryFailure, + ); + + await harness.service.reconcile(); + await harness.service.reconcile(); + + expect(placements.get(active.sessionId)).toMatchObject({ + state: "active", + generation: active.generation, + environmentId: active.environmentId, + turnClaim: { claimId: claim.claimId, runId: claim.runId }, + }); + expect(placements.listPendingWorkspaceResults()).toHaveLength(1); + expect(harness.environments.destroy).not.toHaveBeenCalled(); + expect(await readRecoveryEvents(REQUEST)).toMatchObject([ + { + customType: WORKSPACE_RECOVERY_FAILURE_TRANSCRIPT_TYPE, + content: expect.stringContaining("workspace failed"), + display: true, + }, + ]); + + harnessOptions.failAt = undefined; + await harness.service.reconcile(); + await harness.service.reconcile(); + + expect(placements.get(active.sessionId)).toMatchObject({ state: "active", turnClaim: null }); + expect(placements.listPendingWorkspaceResults()).toEqual([]); + expect(await readRecoveryEvents(REQUEST)).toMatchObject([ + { customType: WORKSPACE_RECOVERY_FAILURE_TRANSCRIPT_TYPE, display: true }, + ]); + }); + }); + + it("persists bounded recovery failures and deduplicates identical consecutive attempts", async () => { + await withOpenClawTestState({ scenario: "minimal" }, async () => { + await upsertSessionEntryCore(IDENTITY, { sessionId: IDENTITY.sessionId, updatedAt: 1 }); + const { reportWorkspaceResultRecoveryFailure } = + createWorkerWorkspaceConflictTranscriptHandlers(loadSessionRuntime); + const secret = [ + String.fromCharCode(115, 107), + "proj", + "recovery", + "abcdefghijklmnopqrstuvwxyz", + ].join("-"); + const firstError = `snapshot rejected token=${secret} ${"detail ".repeat(200)}`; + + await reportWorkspaceResultRecoveryFailure({ ...IDENTITY, error: firstError }); + await reportWorkspaceResultRecoveryFailure({ ...IDENTITY, error: firstError }); + + const firstEvents = await readRecoveryEvents(); + expect(firstEvents).toHaveLength(1); + expect(firstEvents[0]).toMatchObject({ + customType: WORKSPACE_RECOVERY_FAILURE_TRANSCRIPT_TYPE, + display: true, + content: expect.stringMatching( + /^Cloud workspace recovery attempt failed: snapshot rejected token=.*OpenClaw preserved the result and will retry\.$/u, + ), + }); + expect(JSON.stringify(firstEvents[0])).not.toContain(secret); + expect(String(firstEvents[0]?.content).length).toBeLessThanOrEqual(1_024); + + await reportWorkspaceResultRecoveryFailure({ + ...IDENTITY, + error: "snapshot verification failed", + }); + + expect(await readRecoveryEvents()).toMatchObject([ + { customType: WORKSPACE_RECOVERY_FAILURE_TRANSCRIPT_TYPE }, + { + customType: WORKSPACE_RECOVERY_FAILURE_TRANSCRIPT_TYPE, + content: expect.stringContaining("snapshot verification failed"), + }, + ]); + }); + }); + + it("rejects a rebound session identity without touching its replacement transcript", async () => { + await withOpenClawTestState({ scenario: "minimal" }, async () => { + await upsertSessionEntryCore(IDENTITY, { sessionId: IDENTITY.sessionId, updatedAt: 1 }); + const { reportWorkspaceResultRecoveryFailure } = + createWorkerWorkspaceConflictTranscriptHandlers(loadSessionRuntime); + await upsertSessionEntryCore(IDENTITY, { + sessionId: "replacement-workspace-session", + updatedAt: 2, + }); + + await expect( + reportWorkspaceResultRecoveryFailure({ ...IDENTITY, error: "stale worker recovery" }), + ).rejects.toThrow("workspace recovery lost session"); + + expect( + await readRecoveryEvents({ ...IDENTITY, sessionId: "replacement-workspace-session" }), + ).toEqual([]); + }); + }); + + it("revalidates a rebound session after waiting for the transcript writer", async () => { + await withOpenClawTestState({ scenario: "minimal" }, async () => { + await upsertSessionEntryCore(IDENTITY, { sessionId: IDENTITY.sessionId, updatedAt: 1 }); + const runtime = await loadSessionRuntime(); + let resolvedSessionId = IDENTITY.sessionId; + const { reportWorkspaceResultRecoveryFailure } = + createWorkerWorkspaceConflictTranscriptHandlers(async () => ({ + ...runtime, + resolveCanonicalSessionEntryFromStoreKeys: (store, storeKeys) => { + const entry = runtime.resolveCanonicalSessionEntryFromStoreKeys(store, storeKeys); + return entry ? { ...entry, sessionId: resolvedSessionId } : entry; + }, + })); + let releaseWriter!: () => void; + let signalWriterHeld!: () => void; + const writerHeld = new Promise((resolve) => { + signalWriterHeld = resolve; + }); + const release = new Promise((resolve) => { + releaseWriter = resolve; + }); + const blocker = runExclusiveSqliteSessionWrite({ agentId: IDENTITY.agentId }, async () => { + signalWriterHeld(); + await release; + }); + await writerHeld; + + const reporting = reportWorkspaceResultRecoveryFailure({ + ...IDENTITY, + error: "queued stale recovery", + }).then( + () => undefined, + (error: unknown) => error, + ); + await Promise.resolve(); + await Promise.resolve(); + resolvedSessionId = "replacement-workspace-session"; + releaseWriter(); + await blocker; + + await expect(reporting).resolves.toEqual( + expect.objectContaining({ + message: expect.stringContaining("workspace recovery lost session"), + }), + ); + expect(await readRecoveryEvents()).toEqual([]); + }); + }); +}); diff --git a/ui/src/e2e/cloud-workspace-conflict.e2e.test.ts b/ui/src/e2e/cloud-workspace-conflict.e2e.test.ts index b24f2e37a457..a177dd9ed879 100644 --- a/ui/src/e2e/cloud-workspace-conflict.e2e.test.ts +++ b/ui/src/e2e/cloud-workspace-conflict.e2e.test.ts @@ -93,7 +93,7 @@ function workerRecoverySessionsList(includeError: boolean) { ...(includeError ? { recoveryError: "cloud worker disappeared: provider reported lease destroyed", - terminalReason: "cloud worker disappeared: provider reported lease destroyed", + terminalReason: "stale terminal worker failure", terminalAtMs: now, } : {}), @@ -196,6 +196,42 @@ describeControlUiE2e("Control UI cloud workspace conflict recovery", () => { } }); + it("renders historical workspace recovery failures from transcript history", async () => { + const context = await browser.newContext({ + colorScheme: "dark", + locale: "en-US", + serviceWorkers: "block", + viewport: { height: 900, width: 1440 }, + }); + const page = await context.newPage(); + await installMockGateway(page, { + historyMessages: [ + { + role: "custom", + customType: "cloud-workspace-recovery-failed", + content: + "Cloud workspace recovery attempt failed: snapshot verification failed. OpenClaw preserved the result and will retry.", + timestamp: Date.now() - 500, + }, + ], + methodResponses: { "sessions.list": workerRecoverySessionsList(false) }, + sessionKey, + }); + + try { + const response = await page.goto(controlUiSessionUrl(server.baseUrl, sessionKey)); + expect(response?.status()).toBe(200); + await page + .getByText("OpenClaw preserved the result and will retry.", { exact: false }) + .waitFor({ + timeout: 10_000, + }); + await capture(page, "04-workspace-recovery-failed-history.png"); + } finally { + await context.close(); + } + }); + it("shows a durable selected-chat alert while workspace recovery is pending", async () => { const context = await browser.newContext({ colorScheme: "dark", @@ -228,6 +264,7 @@ describeControlUiE2e("Control UI cloud workspace conflict recovery", () => { const alert = page.getByRole("alert").filter({ hasText: "Runner failed" }); await alert.waitFor({ timeout: 10_000 }); expect(await alert.textContent()).toContain("provider reported lease destroyed"); + expect(await alert.textContent()).not.toContain("stale terminal worker failure"); await capture(page, "05-after-workspace-recovery-error.png"); } finally { await context.close(); diff --git a/ui/src/pages/chat/chat-pane-render.ts b/ui/src/pages/chat/chat-pane-render.ts index cf7a8676e45a..bd25b11d1f4b 100644 --- a/ui/src/pages/chat/chat-pane-render.ts +++ b/ui/src/pages/chat/chat-pane-render.ts @@ -102,8 +102,10 @@ export class ChatPane extends ChatPaneLayoutRender { isGatewayMethodAdvertised(this.context.gateway.snapshot, "sessions.github.publish") === true; const diskSpace = placement?.state === "active" ? placement.diskSpace : undefined; const terminalReason = (placement as { terminalReason?: string } | undefined)?.terminalReason; - const placementRunError = terminalReason - ? { summary: t("chat.cloudWorkerFailed", { error: terminalReason }) } + const placementFailureReason = + placement?.state === "failed" ? placement.recoveryError : terminalReason; + const placementRunError = placementFailureReason + ? { summary: t("chat.cloudWorkerFailed", { error: placementFailureReason }) } : null; const visibleWorkspaceConflict = workspaceConflict &&