diff --git a/src/gateway/server-worker-placement-startup.ts b/src/gateway/server-worker-placement-startup.ts index a712274845bb..c0182353cd1e 100644 --- a/src/gateway/server-worker-placement-startup.ts +++ b/src/gateway/server-worker-placement-startup.ts @@ -424,7 +424,7 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme environments: params.environments, placements: params.placements, resolveWorkspacePath, - recoverPendingWorkspaceResult: async (environmentId) => + reconcileActivePlacement: async (environmentId) => await dispatchService.reconcileActive(environmentId), redispatchReclaimed: createReclaimedPlacementRedispatch({ environments: params.environments, diff --git a/src/gateway/server.public-worker-ingress.test.ts b/src/gateway/server.public-worker-ingress.test.ts index 2d44497f4e02..f28048b4735c 100644 --- a/src/gateway/server.public-worker-ingress.test.ts +++ b/src/gateway/server.public-worker-ingress.test.ts @@ -331,6 +331,16 @@ describe("public worker ingress", () => { harness.url(), workerConnect(harness.credential, { environmentId: "worker-other" }), ); + const staleBuild = await rejectWorker( + harness.url(), + workerConnect(harness.credential, { + handshake: { + ...BUILD, + bundleHash: "b".repeat(64), + protocolFeatures: [...BUILD.protocolFeatures], + }, + }), + ); harness.credentialRecord.expiresAtMs = Date.now() - 1; const expiredCredential = await rejectWorker( harness.url(), @@ -338,6 +348,7 @@ describe("public worker ingress", () => { ); expect(badCredential).toEqual(wrongEnvironment); + expect(staleBuild).toEqual(badCredential); expect(expiredCredential).toEqual(badCredential); expect(badCredential).toEqual({ response: { @@ -358,6 +369,9 @@ describe("public worker ingress", () => { expect(harness.logWsControl.warn).toHaveBeenCalledWith( "worker admission rejected reason=environment-mismatch", ); + expect(harness.logWsControl.warn).toHaveBeenCalledWith( + "worker admission rejected reason=bundle-mismatch", + ); expect(harness.logWsControl.warn).toHaveBeenCalledWith( "worker admission rejected reason=credential-expired", ); diff --git a/src/gateway/worker-environments/admission.ts b/src/gateway/worker-environments/admission.ts index 8f763084448f..03400669c637 100644 --- a/src/gateway/worker-environments/admission.ts +++ b/src/gateway/worker-environments/admission.ts @@ -19,6 +19,9 @@ import type { WorkerEnvironmentStore } from "./store.js"; export type { WorkerConnectionIdentity } from "./connection-identity.js"; 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."; + /** 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/placement-session-retirement.test.ts b/src/gateway/worker-environments/placement-session-retirement.test.ts index 671bc02c868f..0ccdcbbdd50c 100644 --- a/src/gateway/worker-environments/placement-session-retirement.test.ts +++ b/src/gateway/worker-environments/placement-session-retirement.test.ts @@ -169,10 +169,23 @@ describe("placement session retirement", () => { sessionKey: "agent:main:session-requested", agentId: "main", }); + const ownedIdentity = { + sessionId: "session-owned-requested", + sessionKey: "agent:main:session-owned-requested", + agentId: "main", + }; + const ownedClaim = placements.claimTurn({ + ...ownedIdentity, + owner: { kind: "local" }, + claimId: "requested-owner-claim", + runId: "requested-owner-run", + }); + const ownedRequested = placements.startDispatch(ownedIdentity); const retireSessionPlacement = vi.fn((input: WorkerSessionPlacementRetirement) => placements.retireSessionPlacement(input), ); const forceDestroyEnvironment = vi.fn(); + const warn = vi.fn(); const retirement = createPlacementSessionRetirement({ placements: { get: (sessionId) => placements.get(sessionId), @@ -182,7 +195,7 @@ describe("placement session retirement", () => { environments: { get: () => undefined }, forceDestroyEnvironment, createSessionEvidenceResolver: async () => async () => "absent", - warn: vi.fn(), + warn, }); try { @@ -193,7 +206,20 @@ describe("placement session retirement", () => { expectedState: "requested", expectedGeneration: requested.generation, }); + expect(retireSessionPlacement).toHaveBeenCalledOnce(); expect(placements.get(requested.sessionId)).toBeUndefined(); + expect(placements.get(ownedRequested.sessionId)).toMatchObject({ + state: "requested", + generation: ownedRequested.generation, + turnClaim: { + owner: "local", + claimId: ownedClaim.claimId, + runId: ownedClaim.runId, + }, + }); + expect(warn).toHaveBeenCalledWith( + `Retired ownerless worker placement ${requested.sessionId} because its authoritative session is absent (requested@${requested.generation})`, + ); expect(forceDestroyEnvironment).not.toHaveBeenCalled(); } finally { closeOpenClawStateDatabaseForTest(); diff --git a/src/gateway/worker-environments/placement-session-retirement.ts b/src/gateway/worker-environments/placement-session-retirement.ts index b473e3679809..6d99b5b804fb 100644 --- a/src/gateway/worker-environments/placement-session-retirement.ts +++ b/src/gateway/worker-environments/placement-session-retirement.ts @@ -46,6 +46,11 @@ export function createPlacementSessionRetirement(deps: PlacementSessionRetiremen expectedState: placement.state, expectedGeneration: placement.generation, }); + if (placement.state === "requested") { + deps.warn( + `Retired ownerless worker placement ${placement.sessionId} because its authoritative session is absent (${placement.state}@${placement.generation})`, + ); + } return true; }; diff --git a/src/gateway/worker-environments/provider-lifecycle.ts b/src/gateway/worker-environments/provider-lifecycle.ts index e8c8498e61ef..d93825affd58 100644 --- a/src/gateway/worker-environments/provider-lifecycle.ts +++ b/src/gateway/worker-environments/provider-lifecycle.ts @@ -14,7 +14,7 @@ import { type WorkerSshEndpoint, type WorkerSshIdentity, } from "../../plugins/types.js"; -import { verifyWorkerAdmissionHandshake } from "./admission.js"; +import { STALE_WORKER_BUILD_REASON, verifyWorkerAdmissionHandshake } from "./admission.js"; import type { WorkerInstallationArtifact } from "./bundle.js"; import type { WorkerCredentialBroker } from "./credential-broker.js"; import { deriveEnvironmentIntent } from "./service-contract.js"; @@ -381,6 +381,19 @@ export function createWorkerProviderLifecycle(options: WorkerProviderLifecycleOp return finishProvenDestroy(destroying); }; + const recordStaleBuildDestroy = (record: WorkerEnvironmentRecord) => { + // Attached sessions need the build cause after teardown so placement reconciliation can + // persist an actionable terminal reason instead of inferring from destroyed environment state. + return record.state === "attached" + ? store.requestDestroy({ + environmentId: record.environmentId, + state: record.state, + terminalState: "failed", + lastError: STALE_WORKER_BUILD_REASON, + }) + : record; + }; + const reconcileRecord = async (initialRecord: WorkerEnvironmentRecord): Promise => { let record = initialRecord; if (record.state === "requested" && record.destroyRequestedAtMs !== null) { @@ -492,7 +505,7 @@ export function createWorkerProviderLifecycle(options: WorkerProviderLifecycleOp // A stale node environment cannot be upgraded in place because its credential and // placement ownership bind the old build. Retire it; reprovisioning reuses the installed // content-addressed bundle without another transfer. - await finishDestroy(record, provider).catch(() => undefined); + await finishDestroy(recordStaleBuildDestroy(record), provider).catch(() => undefined); } return; } @@ -505,7 +518,7 @@ export function createWorkerProviderLifecycle(options: WorkerProviderLifecycleOp // A new Gateway build rejects the old worker at admission. This is expected lifecycle // teardown, not a bootstrap failure. `leaseId` above came from this record, so provider // inspection and destruction share the same durable lease identity. - await finishDestroy(record, provider).catch(() => undefined); + await finishDestroy(recordStaleBuildDestroy(record), provider).catch(() => undefined); } return; } diff --git a/src/gateway/worker-environments/provider-reconciliation.test.ts b/src/gateway/worker-environments/provider-reconciliation.test.ts index 0c950c3eb6a3..e4e0626cbd08 100644 --- a/src/gateway/worker-environments/provider-reconciliation.test.ts +++ b/src/gateway/worker-environments/provider-reconciliation.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it, vi } from "vitest"; +import { STALE_WORKER_BUILD_REASON } from "./admission.js"; import * as support from "./service.test-support.js"; import type { WorkerTunnelManager } from "./tunnel.js"; @@ -164,10 +165,10 @@ describe("worker environment service", () => { profile: { region: "test" }, }); expect(support.testState.store.get(environmentId)).toMatchObject({ - state: "destroyed", - leaseId: `lease:${environmentId}`, + state: "failed", + leaseId: null, attachedSessionIds: [], - lastError: null, + lastError: STALE_WORKER_BUILD_REASON, }); }); @@ -187,6 +188,11 @@ describe("worker environment service", () => { ensureNodeWorkerBundle: async () => structuredClone(support.BOOTSTRAP_RECEIPT), }); const environment = await workerService.create("development", "request-stale-node-bundle"); + await workerService.attachSession({ + environmentId: environment.environmentId, + ownerEpoch: environment.ownerEpoch, + sessionId: "session-stale-node-bundle", + }); support.testState.stateDb.db .prepare( "UPDATE worker_environments SET bootstrap_bundle_hash = ?, bootstrap_install_kind = 'local' WHERE environment_id = ?", @@ -197,8 +203,10 @@ describe("worker environment service", () => { expect(destroy).toHaveBeenCalledOnce(); expect(support.testState.store.get(environment.environmentId)).toMatchObject({ - state: "destroyed", + state: "failed", + leaseId: null, attachedSessionIds: [], + lastError: STALE_WORKER_BUILD_REASON, }); }); diff --git a/src/gateway/worker-environments/worker-turn-admission.ts b/src/gateway/worker-environments/worker-turn-admission.ts index 777d47303325..33c42af2981d 100644 --- a/src/gateway/worker-environments/worker-turn-admission.ts +++ b/src/gateway/worker-environments/worker-turn-admission.ts @@ -19,6 +19,14 @@ type ActiveWorkerPlacement = Extract { beforeEach(setupWorkerTurnLauncherTest); afterEach(cleanupWorkerTurnLauncherTest); + it("projects a stale Gateway build teardown and records its durable placement reason", async () => { + seedActivePlacement(); + const terminalReason = `cloud worker disappeared: ${STALE_WORKER_BUILD_REASON}`; + const staleEnvironment: NonNullable> = { + ...attachedEnvironment(), + state: "failed" as const, + leaseId: null, + sshEndpoint: null, + sharedHost: null, + ownerEpoch: OWNER_EPOCH + 1, + attachedSessionIds: [], + tunnelStatus: "stopped", + error: STALE_WORKER_BUILD_REASON, + }; + const environments: WorkerTurnEnvironmentService = { + ...unusedEnvironments(), + get: vi.fn(() => staleEnvironment), + }; + const reconcileActivePlacement = vi.fn(async () => { + const active = placements.get(SESSION_ID); + if (active?.state !== "active") { + throw new Error("expected active stale-build placement"); + } + const draining = placements.startDrain({ + sessionId: active.sessionId, + environmentId: active.environmentId, + ownerEpoch: active.activeOwnerEpoch, + expectedGeneration: active.generation, + }); + if (draining.state !== "draining") { + throw new Error("expected draining stale-build placement"); + } + const reconciling = placements.startReconcile({ + sessionId: draining.sessionId, + environmentId: draining.environmentId, + ownerEpoch: draining.activeOwnerEpoch, + expectedGeneration: draining.generation, + }); + if (reconciling.state !== "reconciling") { + throw new Error("expected reconciling stale-build placement"); + } + placements.fail({ + sessionId: reconciling.sessionId, + expectedGeneration: reconciling.generation, + recoveryError: terminalReason, + }); + }); + const provider = createWorkerSessionTurnPlacementProvider({ + environments, + placements, + reconcileActivePlacement, + }); + + await expect( + provider.executeTurn( + { + sessionId: SESSION_ID, + sessionKey: SESSION_KEY, + agentId: "main", + runId: "run-stale-worker-build", + }, + turn("run-stale-worker-build"), + async () => ({ meta: { durationMs: 1 } }), + ), + ).rejects.toThrow(`Worker turn rejected in placement failed: ${terminalReason}`); + expect(reconcileActivePlacement).toHaveBeenCalledWith(ENVIRONMENT_ID); + expect(placements.get(SESSION_ID)).toMatchObject({ + state: "failed", + recoveryError: terminalReason, + terminalReason, + turnClaim: 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-terminal-results.test.ts b/src/gateway/worker-environments/worker-turn-launcher-terminal-results.test.ts index 3b717e74ba0c..3e0d3c1e3b2e 100644 --- a/src/gateway/worker-environments/worker-turn-launcher-terminal-results.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher-terminal-results.test.ts @@ -88,7 +88,7 @@ describe("worker turn launcher terminal results", () => { startTunnel: vi.fn(async () => tunnel), destroy, }; - const recoverPendingWorkspaceResult = vi.fn(async () => { + const reconcileActivePlacement = vi.fn(async () => { const [pending] = placements.listPendingWorkspaceResults(); if (!pending) { throw new Error("expected pending workspace result"); @@ -98,7 +98,7 @@ describe("worker turn launcher terminal results", () => { const provider = createWorkerSessionTurnPlacementProvider({ environments, placements, - recoverPendingWorkspaceResult, + reconcileActivePlacement, }); await expect( @@ -117,7 +117,7 @@ describe("worker turn launcher terminal results", () => { "Cloud worker finished, but its workspace result could not be reconciled: workspace-transfer-failed: gateway TLS fingerprint mismatch", }); - expect(recoverPendingWorkspaceResult).toHaveBeenCalledWith(ENVIRONMENT_ID); + expect(reconcileActivePlacement).toHaveBeenCalledWith(ENVIRONMENT_ID); expect(placements.get(SESSION_ID)).toMatchObject({ state: "failed", turnClaim: null }); expect(placements.listPendingWorkspaceResults()).toHaveLength(0); expect(destroy).not.toHaveBeenCalled(); diff --git a/src/gateway/worker-environments/worker-turn-launcher.test-support.ts b/src/gateway/worker-environments/worker-turn-launcher.test-support.ts index 6d6096bf55bd..b015ded578d8 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.test-support.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.test-support.ts @@ -94,7 +94,7 @@ export function setWorkerTurnSessionTarget(target: typeof sessionTarget): typeof } type DefaultedWorkerTurnLauncherOption = - | "recoverPendingWorkspaceResult" + | "reconcileActivePlacement" | "redispatchReclaimed" | "resolveWorkspacePath" | "workspaceOperations"; @@ -104,8 +104,8 @@ export function createWorkerSessionTurnPlacementProvider( Partial>, ) { return createRawWorkerSessionTurnPlacementProvider({ - recoverPendingWorkspaceResult: async () => { - throw new Error("unexpected pending workspace recovery"); + reconcileActivePlacement: async () => { + throw new Error("unexpected active placement reconciliation"); }, redispatchReclaimed: async () => { throw new Error("unexpected reclaimed placement redispatch"); diff --git a/src/gateway/worker-environments/worker-turn-launcher.test.ts b/src/gateway/worker-environments/worker-turn-launcher.test.ts index cadae9c127a2..6756428a5f0c 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.test.ts @@ -361,7 +361,7 @@ describe("worker turn launcher local placement", () => { get: vi.fn(() => attachedEnvironment()), startTunnel: vi.fn(async () => tunnel), }; - const recoverPendingWorkspaceResult = vi.fn(async () => { + const reconcileActivePlacement = vi.fn(async () => { const placement = placements.get(SESSION_ID); if (placement?.state !== "failed" || placement.turnClaim !== null) { throw new Error("expected terminal placement before teardown recovery"); @@ -371,7 +371,7 @@ describe("worker turn launcher local placement", () => { const provider = createWorkerSessionTurnPlacementProvider({ environments, placements, - recoverPendingWorkspaceResult, + reconcileActivePlacement, }); await expect( @@ -389,7 +389,7 @@ describe("worker turn launcher local placement", () => { "Cloud worker finished, but its workspace result could not be reconciled: workspace manifest memo exceeds its entry limit", ); - expect(recoverPendingWorkspaceResult).toHaveBeenCalledWith(ENVIRONMENT_ID); + expect(reconcileActivePlacement).toHaveBeenCalledWith(ENVIRONMENT_ID); expect(placements.get(SESSION_ID)).toMatchObject({ state: "failed", turnClaim: null, diff --git a/src/gateway/worker-environments/worker-turn-launcher.ts b/src/gateway/worker-environments/worker-turn-launcher.ts index 7600bb2ade19..a0ba15c97b1a 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.ts @@ -14,7 +14,7 @@ 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 { supportsWorkerExecutionContextLaunch } from "./admission.js"; +import { STALE_WORKER_BUILD_REASON, supportsWorkerExecutionContextLaunch } from "./admission.js"; import { placementTurnOwner } from "./placement-record.js"; import { createRemoteExecPlacementSandbox } from "./placement-sandbox.js"; import type { @@ -62,7 +62,7 @@ type WorkerTurnLauncherOptions = { environments: WorkerTurnEnvironmentService; placements: WorkerSessionPlacementStore; resolveWorkspacePath: (identity: ReturnType) => Promise; - recoverPendingWorkspaceResult: (environmentId: string) => Promise; + reconcileActivePlacement: (environmentId: string) => Promise; workspaceOperations: WorkerWorkspaceOperationCoordinator; redispatchReclaimed: (placement: ReclaimedWorkerPlacement) => Promise; }; @@ -91,6 +91,7 @@ async function executeWorkerTurn(params: { onHandoff: () => void; placement: ActiveWorkerPlacement; placements: WorkerSessionPlacementStore; + reconcileActivePlacement: (environmentId: string) => Promise; workspaceOperations: WorkerWorkspaceOperationCoordinator; turn: SessionPlacementTurnParams; turnClaim: WorkerSessionTurnClaim; @@ -100,6 +101,16 @@ async function executeWorkerTurn(params: { const modelRef = assertSupportedTurn(turn); const environment = params.environments.get(placement.environmentId); const bootstrapReceipt = environment?.bootstrapReceipt; + // 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); + } if ( !environment || environment.state !== "attached" || @@ -496,6 +507,7 @@ export function createWorkerSessionTurnPlacementProvider(options: WorkerTurnLaun }, placement, placements: options.placements, + reconcileActivePlacement: options.reconcileActivePlacement, localWorkspaceDir, workspaceOperations: options.workspaceOperations, turn, @@ -524,7 +536,7 @@ export function createWorkerSessionTurnPlacementProvider(options: WorkerTurnLaun // could discard the terminal event's durably fenced file results. options.placements.handoffWorkspaceResultRecovery(turnClaim); } - await options.recoverPendingWorkspaceResult(placement.environmentId); + await options.reconcileActivePlacement(placement.environmentId); throw error; } if (