From e2cefa8fe4f15816802f475347c5928db5120313 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 16 Aug 2026 17:04:39 -0700 Subject: [PATCH] fix(gateway): terminalize stale worker dispatches (#124920) --- .../chat.directive-tags.test.ts | 3 +- src/gateway/worker-environments/admission.ts | 8 ++ .../credential-broker.test.ts | 3 +- .../worker-environments/credential-broker.ts | 11 ++- .../environment-access.test.ts | 3 +- .../worker-environments/environment-access.ts | 11 ++- ...ker-turn-launcher-failure-recovery.test.ts | 91 ++++++++++++++++++- .../worker-turn-launcher.ts | 20 ++-- 8 files changed, 129 insertions(+), 21 deletions(-) diff --git a/src/gateway/server-methods/chat.directive-tags.test.ts b/src/gateway/server-methods/chat.directive-tags.test.ts index cc22e5e36733..8d19ed3256fc 100644 --- a/src/gateway/server-methods/chat.directive-tags.test.ts +++ b/src/gateway/server-methods/chat.directive-tags.test.ts @@ -60,6 +60,7 @@ import { withEnvAsync } from "../../test-utils/env.js"; import { normalizeSessionDeliveryState } from "../../utils/delivery-context.shared.js"; import { consumeCronCreatorAuthorityGrant } from "../cron-creator-authority-grant.js"; import { createChatRunState } from "../server-chat-state.js"; +import { STALE_WORKER_BUILD_REASON } from "../worker-environments/admission.js"; import { handleChatSend } from "./chat-send-handler.js"; import type { GatewayRequestContext } from "./types.js"; @@ -4528,7 +4529,7 @@ describe("chat directive tag stripping for non-streaming final payloads", () => await createGatewayUserTurnSqliteFixture("openclaw-chat-send-agent-returned-error-"); const errorMessage = agentStarted ? "LLM idle timeout (120s): no response from model" - : "Worker must bootstrap the current build before continuing"; + : STALE_WORKER_BUILD_REASON; mockState.triggerAgentRunStart = agentStarted; mockState.dispatchedReplies = [ { diff --git a/src/gateway/worker-environments/admission.ts b/src/gateway/worker-environments/admission.ts index 03400669c637..6aa53226f83b 100644 --- a/src/gateway/worker-environments/admission.ts +++ b/src/gateway/worker-environments/admission.ts @@ -22,6 +22,14 @@ export type { ExpectedWorkerBuild } from "../../worker/worker-build-identity.js" export const STALE_WORKER_BUILD_REASON = "Worker build does not match the current Gateway build; redispatch the session so its worker can bootstrap the current build before retrying."; +export class StaleWorkerBuildError extends Error { + readonly code = "invalid_state"; + + constructor() { + super(STALE_WORKER_BUILD_REASON); + } +} + /** True only for bundles that accept the exact admitted execution carrier. */ export function supportsWorkerExecutionContextLaunch( handshake: Pick | null | undefined, diff --git a/src/gateway/worker-environments/credential-broker.test.ts b/src/gateway/worker-environments/credential-broker.test.ts index 387b6584899f..bcc2db636705 100644 --- a/src/gateway/worker-environments/credential-broker.test.ts +++ b/src/gateway/worker-environments/credential-broker.test.ts @@ -3,6 +3,7 @@ import { closeOpenClawStateDatabaseForTest, openOpenClawStateDatabase, } from "../../state/openclaw-state-db.js"; +import { STALE_WORKER_BUILD_REASON } from "./admission.js"; import * as support from "./service.test-support.js"; import { createWorkerEnvironmentStore } from "./store.js"; import type { WorkerTunnelManager } from "./tunnel.js"; @@ -60,7 +61,7 @@ describe("worker environment service", () => { ownerEpoch: 1, sessionId: "session-1", }), - ).rejects.toThrow("must bootstrap the current build"); + ).rejects.toThrow(STALE_WORKER_BUILD_REASON); expect(support.testState.store.get(staleId)).toMatchObject({ state: "ready", attachedSessionIds: [], diff --git a/src/gateway/worker-environments/credential-broker.ts b/src/gateway/worker-environments/credential-broker.ts index 9b84ffa18583..aa86f739a451 100644 --- a/src/gateway/worker-environments/credential-broker.ts +++ b/src/gateway/worker-environments/credential-broker.ts @@ -2,7 +2,11 @@ import { type WorkerAdmissionHandshake, WORKER_RPC_SET_VERSION, } from "../../../packages/gateway-protocol/src/schema/worker-admission.js"; -import { verifyWorkerAdmissionHandshake, type ExpectedWorkerBuild } from "./admission.js"; +import { + StaleWorkerBuildError, + verifyWorkerAdmissionHandshake, + type ExpectedWorkerBuild, +} from "./admission.js"; import type { WorkerInstallationArtifact } from "./bundle.js"; import { createWorkerCredentialMaterial, @@ -224,10 +228,7 @@ export function createWorkerCredentialBroker(options: WorkerCredentialBrokerOpti !current.bootstrapReceipt || !verifyWorkerAdmissionHandshake(current.bootstrapReceipt, currentBuild) ) { - throw serviceError( - "invalid_state", - "Worker must bootstrap the current build before attach", - ); + throw new StaleWorkerBuildError(); } const material = credentialMaterial(); let attached: WorkerEnvironmentRecord; diff --git a/src/gateway/worker-environments/environment-access.test.ts b/src/gateway/worker-environments/environment-access.test.ts index 6110458ab92b..ca7cac4d60e9 100644 --- a/src/gateway/worker-environments/environment-access.test.ts +++ b/src/gateway/worker-environments/environment-access.test.ts @@ -3,6 +3,7 @@ import { closeOpenClawStateDatabaseForTest, openOpenClawStateDatabase, } from "../../state/openclaw-state-db.js"; +import { STALE_WORKER_BUILD_REASON } from "./admission.js"; import * as support from "./service.test-support.js"; import { createWorkerEnvironmentStore } from "./store.js"; import type { WorkerTunnelManager } from "./tunnel.js"; @@ -98,7 +99,7 @@ describe("worker environment service", () => { code: "invalid_state", message: prepareError ? "Current worker build identity is unavailable" - : "Worker must bootstrap the current build before continuing", + : STALE_WORKER_BUILD_REASON, } satisfies Partial, ); expect(tunnelManager.start).not.toHaveBeenCalled(); diff --git a/src/gateway/worker-environments/environment-access.ts b/src/gateway/worker-environments/environment-access.ts index 1cc724cda834..0692a750ef15 100644 --- a/src/gateway/worker-environments/environment-access.ts +++ b/src/gateway/worker-environments/environment-access.ts @@ -2,7 +2,11 @@ import { isRecord } from "@openclaw/normalization-core/record-coerce"; import type { OpenClawConfig } from "../../config/types.js"; import { withTimeout } from "../../infra/fs-safe.js"; import type { WorkerProvider } from "../../plugins/types.js"; -import { verifyWorkerAdmissionHandshake, type ExpectedWorkerBuild } from "./admission.js"; +import { + StaleWorkerBuildError, + verifyWorkerAdmissionHandshake, + type ExpectedWorkerBuild, +} from "./admission.js"; import { DEVICE_WORKER_PROVIDER_ID } from "./device-provider.js"; import type { NodeWorkerTunnelManager } from "./node-worker-tunnel.js"; import type { WorkerDesktopLaunchResult, WorkerDesktopObserveResult } from "./service-contract.js"; @@ -127,10 +131,7 @@ export function createWorkerEnvironmentAccess(options: WorkerEnvironmentAccessOp throw serviceError("invalid_state", "Current worker build identity is unavailable"); } if (!verifyWorkerAdmissionHandshake(record.bootstrapReceipt, currentBundle)) { - throw serviceError( - "invalid_state", - "Worker must bootstrap the current build before continuing", - ); + throw new StaleWorkerBuildError(); } const nodeBundle = record.providerId === DEVICE_WORKER_PROVIDER_ID && diff --git a/src/gateway/worker-environments/worker-turn-launcher-failure-recovery.test.ts b/src/gateway/worker-environments/worker-turn-launcher-failure-recovery.test.ts index c4452aa843d5..c6416031a926 100644 --- a/src/gateway/worker-environments/worker-turn-launcher-failure-recovery.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher-failure-recovery.test.ts @@ -7,7 +7,9 @@ import { createDeferred } from "../../../test/helpers/promise.js"; import { makeAgentAssistantMessage } from "../../agents/test-helpers/agent-message-fixtures.js"; import type { SpawnResult } from "../../process/exec.js"; import { WORKER_PROVIDER_REPLAY_LOCAL_RETRY_MESSAGE } from "../../worker/transcript-message.js"; -import { STALE_WORKER_BUILD_REASON } from "./admission.js"; +import { STALE_WORKER_BUILD_REASON, StaleWorkerBuildError } from "./admission.js"; +import type { WorkerDispatchEnvironmentService } from "./placement-dispatch-failure.js"; +import { createWorkerPlacementDispatchService } from "./placement-dispatch.js"; import type { WorkerSessionPlacementStore } from "./placement-store.js"; import { WorkerRunnerCapacityError, @@ -35,6 +37,7 @@ import { type WorkerTurnEnvironmentService, type WorkerTurnLauncherOptions, } from "./worker-turn-launcher.test-support.js"; +import { createWorkerWorkspaceOperationCoordinator } from "./workspace-operation-coordinator.js"; describe("worker turn launcher failure recovery", () => { beforeEach(setupWorkerTurnLauncherTest); @@ -114,6 +117,92 @@ describe("worker turn launcher failure recovery", () => { }); }); + it("terminalizes a stale build rejected during live tunnel admission", async () => { + seedActivePlacement(); + const terminalReason = `cloud worker disappeared: ${STALE_WORKER_BUILD_REASON}`; + let environment = attachedEnvironment(); + const stopTunnel = vi.fn(async () => {}); + const destroy = vi.fn(async () => environment); + const reconcileOnce = vi.fn(async () => { + environment = { + ...environment, + state: "failed", + leaseId: null, + sshEndpoint: null, + sharedHost: null, + ownerEpoch: OWNER_EPOCH + 1, + attachedSessionIds: [], + tunnelStatus: "stopped", + error: STALE_WORKER_BUILD_REASON, + }; + }); + const environments: WorkerTurnEnvironmentService & WorkerDispatchEnvironmentService = { + ...unusedEnvironments(), + get: vi.fn(() => environment), + acquireTurnCredential: vi.fn(async () => credential()), + acknowledgeCredentialDelivery: vi.fn(() => true), + startTunnel: vi.fn(async () => { + throw new StaleWorkerBuildError(); + }), + stopTunnel, + destroy, + attachSession: vi.fn(async () => { + throw new Error("unexpected worker session attachment"); + }), + create: vi.fn(async () => { + throw new Error("unexpected worker environment creation"); + }), + createFromProfileSnapshot: vi.fn(async () => { + throw new Error("unexpected inherited worker environment creation"); + }), + reconcileOnce, + }; + const workspaceOperations = createWorkerWorkspaceOperationCoordinator(); + const dispatch = createWorkerPlacementDispatchService({ + placements, + environments, + runLocalBarrier: async ({ startDispatch }) => startDispatch(), + runActivationBarrier: async ({ activate }) => activate(), + runReclaimBarrier: async ({ reclaim }) => await reclaim(root), + workspaceOperations, + resolveWorkspacePath: async () => root, + reportWorkspaceResultConflict: async () => {}, + resolveWorkspaceResultConflict: async () => undefined, + }); + const provider = createWorkerSessionTurnPlacementProvider({ + environments, + placements, + reconcileActivePlacement: dispatch.reconcileActive, + workspaceOperations, + }); + + await expect( + provider.executeTurn( + { + sessionId: SESSION_ID, + sessionKey: SESSION_KEY, + agentId: "main", + runId: "run-live-stale-worker-build", + }, + turn("run-live-stale-worker-build"), + async () => ({ meta: { durationMs: 1 } }), + ), + ).rejects.toThrow(`Worker turn rejected in placement failed: ${terminalReason}`); + + expect(reconcileOnce).toHaveBeenCalledOnce(); + expect(placements.get(SESSION_ID)).toMatchObject({ + state: "failed", + recoveryError: terminalReason, + terminalReason, + turnClaim: null, + }); + expect(placements.get(SESSION_ID)).not.toMatchObject({ + state: "active", + recoveryError: null, + terminalReason: null, + }); + }); + it("keeps an active placement when tunnel startup fails before remote handoff", async () => { seedActivePlacement(); const acknowledgeCredentialDelivery = vi.fn(() => true); diff --git a/src/gateway/worker-environments/worker-turn-launcher.ts b/src/gateway/worker-environments/worker-turn-launcher.ts index a0ba15c97b1a..87857223361d 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.ts @@ -14,7 +14,11 @@ import { emitAgentRunStatusEvent } from "../../infra/agent-run-status-events.js" import { redactSensitiveText } from "../../logging/redact.js"; import { parseWorkerLaunchPlan } from "../../worker/launch-descriptor.js"; import { WORKER_PROVIDER_REPLAY_LOCAL_RETRY_MESSAGE } from "../../worker/transcript-message.js"; -import { STALE_WORKER_BUILD_REASON, supportsWorkerExecutionContextLaunch } from "./admission.js"; +import { + STALE_WORKER_BUILD_REASON, + StaleWorkerBuildError, + supportsWorkerExecutionContextLaunch, +} from "./admission.js"; import { placementTurnOwner } from "./placement-record.js"; import { createRemoteExecPlacementSandbox } from "./placement-sandbox.js"; import type { @@ -104,12 +108,7 @@ async function executeWorkerTurn(params: { // Provider reconciliation records current-build teardown before placement repair. Consume // that fact before launch so canonical reconciliation can persist the same cause. if (environment?.error === STALE_WORKER_BUILD_REASON) { - await params.reconcileActivePlacement(placement.environmentId); - const reconciled = params.placements.get(placement.sessionId); - if (reconciled) { - requireActivePlacement(reconciled); - } - throw new Error(STALE_WORKER_BUILD_REASON); + throw new StaleWorkerBuildError(); } if ( !environment || @@ -518,6 +517,13 @@ export function createWorkerSessionTurnPlacementProvider(options: WorkerTurnLaun : await executeWorkerTurn(executionParams); return result; } catch (error) { + if (error instanceof StaleWorkerBuildError) { + await options.reconcileActivePlacement(placement.environmentId); + const reconciled = options.placements.get(placement.sessionId); + if (reconciled) { + requireActivePlacement(reconciled); + } + } const pendingWorkspaceResult = options.placements .listPendingWorkspaceResults() .find(