ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-26 20:35:39 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(gateway): terminalize stale worker dispatches (#124920)

Browse Source
This commit is contained in:
Peter Steinberger
2026-08-16 17:04:39 -07:00
committed by GitHub
parent bda2d54dcf
commit e2cefa8fe4
8 changed files with 129 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gateway/server-methods/chat.directive-tags.test.ts
+2 -1
View File
@@ -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 = [
{
src/gateway/worker-environments/admission.ts
+8
View File
@@ -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<WorkerAdmissionHandshake, "protocolFeatures"> | null | undefined,
src/gateway/worker-environments/credential-broker.test.ts
+2 -1
View File
@@ -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: [],
src/gateway/worker-environments/credential-broker.ts
+6 -5
View File
@@ -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;
src/gateway/worker-environments/environment-access.test.ts
+2 -1
View File
@@ -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<WorkerEnvironmentServiceError>,
);
expect(tunnelManager.start).not.toHaveBeenCalled();
src/gateway/worker-environments/environment-access.ts
+6 -5
View File
@@ -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 &&
src/gateway/worker-environments/worker-turn-launcher-failure-recovery.test.ts
+90 -1
View File
@@ -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);
src/gateway/worker-environments/worker-turn-launcher.ts
+13 -7
View File
@@ -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(