From aec8096bbc2bcd8eeb71feeb9810c737cb4a6f48 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 14 Aug 2026 20:30:34 -0700 Subject: [PATCH] refactor(gateway): remove obsolete worker test seams (#123970) --- scripts/test-projects.test-support.mts | 11 +++ .../server-methods/sessions.dispatch.test.ts | 2 + src/gateway/server-runtime-state-prepare.ts | 1 - .../server-worker-placement-startup.test.ts | 2 - .../server-worker-placement-startup.ts | 2 - .../placement-session-runtime.test.ts | 71 ------------------ ...ker-turn-launcher-failure-recovery.test.ts | 72 ------------------- ...-turn-launcher-reclaimed-placement.test.ts | 24 ------- .../worker-turn-launcher.test-support.ts | 18 ++++- .../worker-turn-launcher.test.ts | 51 ------------- .../worker-turn-launcher.ts | 30 ++------ src/node-host/node-worker-capacity.ts | 5 -- test/scripts/test-projects.test.ts | 14 ++++ 13 files changed, 50 insertions(+), 253 deletions(-) delete mode 100644 src/gateway/worker-environments/placement-session-runtime.test.ts diff --git a/scripts/test-projects.test-support.mts b/scripts/test-projects.test-support.mts index da8589903145..f87ac0b133c6 100644 --- a/scripts/test-projects.test-support.mts +++ b/scripts/test-projects.test-support.mts @@ -559,6 +559,17 @@ const PRECISE_SOURCE_TEST_TARGETS = new Map([ "extensions/slack/src/monitor/provider.auth-test-token.test.ts", ], ], + [ + "src/gateway/worker-environments/worker-turn-launcher.ts", + [ + "src/gateway/worker-environments/worker-turn-launcher.test.ts", + "src/gateway/worker-environments/worker-turn-launcher-claim-admission.test.ts", + "src/gateway/worker-environments/worker-turn-launcher-failure-recovery.test.ts", + "src/gateway/worker-environments/worker-turn-launcher-reclaimed-placement.test.ts", + "src/gateway/worker-environments/worker-turn-launcher-remote-handoff.test.ts", + "src/gateway/worker-environments/worker-turn-launcher-terminal-results.test.ts", + ], + ], ]); const DOCS_CONFIG_EXAMPLES_TEST_TARGET = "src/config/docs-config-examples.test.ts"; const RUNTIME_SIDECAR_BASELINE_OWNER_TEST_TARGETS = ["src/plugins/bundled-plugin-metadata.test.ts"]; diff --git a/src/gateway/server-methods/sessions.dispatch.test.ts b/src/gateway/server-methods/sessions.dispatch.test.ts index e364437ed5f8..234611f2120b 100644 --- a/src/gateway/server-methods/sessions.dispatch.test.ts +++ b/src/gateway/server-methods/sessions.dispatch.test.ts @@ -466,6 +466,7 @@ describe("sessions.dispatch", () => { mocks.resolveTarget.mockReturnValue( targetWithEntry({ sessionId, + agentRuntimeOverride: "openclaw", worktree: { id: "worktree-1", branch: "openclaw/cloud-test", repoRoot: "/repo" }, }), ); @@ -526,6 +527,7 @@ describe("sessions.dispatch", () => { sessionId, sessionKey, agentId: "main", + executionMode: "worker-turn", profileId: "test", }), expect.any(Function), diff --git a/src/gateway/server-runtime-state-prepare.ts b/src/gateway/server-runtime-state-prepare.ts index 86527c25f517..1202348d1665 100644 --- a/src/gateway/server-runtime-state-prepare.ts +++ b/src/gateway/server-runtime-state-prepare.ts @@ -184,7 +184,6 @@ export async function prepareGatewayKernelState(params: { placements: workerEnvironmentStartup.placementStore, environments: workerEnvironmentService, gatewayNamespace: nodeWorkerGatewayNamespace, - admitNewPlacements: true, revokeSessionAuthority: (request) => workerDispatchAuthority.revoke(request), warn: (message) => log.warn(message), }); diff --git a/src/gateway/server-worker-placement-startup.test.ts b/src/gateway/server-worker-placement-startup.test.ts index 99818c881a66..13a6693bb33f 100644 --- a/src/gateway/server-worker-placement-startup.test.ts +++ b/src/gateway/server-worker-placement-startup.test.ts @@ -73,7 +73,6 @@ describe("worker placement startup health lifetime", () => { } as never, environments: environments as never, gatewayNamespace: "gateway-test", - admitNewPlacements: true, revokeSessionAuthority: vi.fn(), warn, }); @@ -159,7 +158,6 @@ describe("worker placement startup health lifetime", () => { } as never, environments: environments as never, gatewayNamespace: "gateway-test", - admitNewPlacements: true, revokeSessionAuthority: vi.fn(), warn: vi.fn(), }); diff --git a/src/gateway/server-worker-placement-startup.ts b/src/gateway/server-worker-placement-startup.ts index a327f3734e58..2009b96ca88c 100644 --- a/src/gateway/server-worker-placement-startup.ts +++ b/src/gateway/server-worker-placement-startup.ts @@ -117,7 +117,6 @@ export type GatewayWorkerPlacementRuntimeParams = { placements: WorkerSessionPlacementStore; environments: WorkerEnvironmentService; gatewayNamespace: string; - admitNewPlacements: boolean; revokeSessionAuthority: (request: { sessionId: string; sessionKeys: readonly string[] }) => void; warn: (message: string) => void; }; @@ -419,7 +418,6 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme const admissionProvider = createWorkerSessionTurnPlacementProvider({ environments: params.environments, placements: params.placements, - admitNewPlacements: params.admitNewPlacements, resolveWorkspacePath, recoverPendingWorkspaceResult: async (environmentId) => await dispatchService.reconcileActive(environmentId), diff --git a/src/gateway/worker-environments/placement-session-runtime.test.ts b/src/gateway/worker-environments/placement-session-runtime.test.ts deleted file mode 100644 index c86e6bcece7e..000000000000 --- a/src/gateway/worker-environments/placement-session-runtime.test.ts +++ /dev/null @@ -1,71 +0,0 @@ -import { afterEach, describe, expect, it } from "vitest"; -import { registerAgentHarness } from "../../agents/harness/registry.js"; -import { createEmptyPluginRegistry } from "../../plugins/registry-empty.js"; -import { - getActivePluginRegistry, - resetPluginRuntimeStateForTest, - setActivePluginRegistry, -} from "../../plugins/runtime.js"; -import { - resolveWorkerPlacementExecutionMode, - resolveWorkerPlacementSessionRuntime, -} from "./placement-session-runtime.js"; - -describe("worker placement session runtime", () => { - const originalRegistry = getActivePluginRegistry(); - - afterEach(() => { - if (originalRegistry) { - setActivePluginRegistry(originalRegistry, "placement-runtime-test", "default"); - } else { - resetPluginRuntimeStateForTest(); - } - }); - - it.each([ - ["openclaw", "worker-turn"], - ["unknown", undefined], - ] as const)("resolves %s placement mode", (runtime, expected) => { - setActivePluginRegistry(createEmptyPluginRegistry(), "placement-runtime-test", "default"); - expect(resolveWorkerPlacementExecutionMode(runtime)).toBe(expected); - }); - - it("resolves a registered harness capability", () => { - setActivePluginRegistry(createEmptyPluginRegistry(), "placement-runtime-test", "default"); - const harness = { - id: "codex", - label: "Codex", - cloudPlacement: { mode: "remote-exec" }, - supports: () => ({ supported: true, priority: 10 }), - async runAttempt() { - throw new Error("not used"); - }, - } as const; - registerAgentHarness(harness); - - expect(resolveWorkerPlacementExecutionMode("codex")).toBe("remote-exec"); - }); - - it("uses a persisted runtime before model policy", () => { - setActivePluginRegistry(createEmptyPluginRegistry(), "placement-runtime-test", "default"); - const harness = { - id: "codex", - label: "Codex", - cloudPlacement: { mode: "remote-exec" }, - supports: () => ({ supported: true, priority: 10 }), - async runAttempt() { - throw new Error("not used"); - }, - } as const; - registerAgentHarness(harness); - const runtime = resolveWorkerPlacementSessionRuntime({ - cfg: {}, - entry: { sessionId: "persisted-runtime", updatedAt: 1, agentRuntimeOverride: "codex" }, - agentId: "main", - sessionKey: "agent:main:persisted-runtime", - }); - - expect(runtime).toBe("codex"); - expect(resolveWorkerPlacementExecutionMode(runtime)).toBe("remote-exec"); - }); -}); 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 a409223da388..0ddf75b05649 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 @@ -8,7 +8,6 @@ import { makeAgentAssistantMessage } from "../../agents/test-helpers/agent-messa import type { SpawnResult } from "../../process/exec.js"; import { WORKER_PROVIDER_REPLAY_LOCAL_RETRY_MESSAGE } from "../../worker/transcript-message.js"; import type { WorkerSessionPlacementStore } from "./placement-store.js"; -import { createWorkerSessionPlacementGate } from "./placement-worker-gate.js"; import { WorkerRunnerUnavailableError, type WorkerTunnelHandle } from "./tunnel-contract.js"; import { ENVIRONMENT_ID, @@ -156,77 +155,6 @@ describe("worker turn launcher failure recovery", () => { expect(placements.get(SESSION_ID)).toMatchObject({ state: "active", turnClaim: null }); }); - it("preserves a terminal workspace result when the worker child later exits nonzero", async () => { - seedActivePlacement(); - const destroy = vi.fn(async () => attachedEnvironment()); - const launchTurn = vi.fn( - async (request: Parameters[0]): Promise => { - request.onDispatchReady?.(); - createWorkerSessionPlacementGate(placements).updateAckCursors({ - sessionId: SESSION_ID, - environmentId: ENVIRONMENT_ID, - ownerEpoch: OWNER_EPOCH, - runId: "run-terminal-child-failure", - liveSeq: 1, - }); - return { - stdout: "", - stderr: "child cleanup failed", - code: 1, - signal: null, - killed: false, - termination: "exit", - }; - }, - ); - const environments: WorkerTurnEnvironmentService = { - get: vi.fn(() => attachedEnvironment()), - acquireTurnCredential: vi.fn(async () => credential()), - acknowledgeCredentialDelivery: vi.fn(() => true), - startTunnel: vi.fn(async () => ({ - environmentId: ENVIRONMENT_ID, - ownerEpoch: OWNER_EPOCH, - quiesceWorkspace: vi.fn(), - runWorkspaceCommand: vi.fn(), - launchTurn, - syncWorkspace: vi.fn(), - reconcileWorkspace: vi.fn(), - stop: vi.fn(async () => {}), - })), - stopTunnel: vi.fn(async () => {}), - destroy, - }; - const provider = createWorkerSessionTurnPlacementProvider({ environments, placements }); - - await expect( - provider.executeTurn( - { - sessionId: SESSION_ID, - sessionKey: SESSION_KEY, - agentId: "main", - runId: "run-terminal-child-failure", - }, - turn("run-terminal-child-failure"), - async () => ({ meta: { durationMs: 1 } }), - ), - ).rejects.toThrow("child cleanup failed"); - - expect(launchTurn).toHaveBeenCalledOnce(); - expect(destroy).not.toHaveBeenCalled(); - expect(placements.listPendingWorkspaceResults()).toMatchObject([ - { - sessionId: SESSION_ID, - runId: "run-terminal-child-failure", - gatewayInstanceId: placements.workspaceResultInstanceId(), - recoveryRequestedAtMs: expect.any(Number), - }, - ]); - expect(placements.get(SESSION_ID)).toMatchObject({ - state: "active", - turnClaim: { owner: "worker", runId: "run-terminal-child-failure" }, - }); - }); - it("preserves an unresolved rollback journal when pre-launch recovery conflicts", async () => { seedActivePlacement(); const active = placements.get(SESSION_ID); 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 83dd688f595c..419ba5fab4f2 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 @@ -412,30 +412,6 @@ describe("worker turn launcher reclaimed placement", () => { } }); - it("rejects a reclaimed placement when redispatch is unavailable", async () => { - seedReclaimedPlacement(); - const provider = createWorkerSessionTurnPlacementProvider({ - environments: unusedEnvironments(), - placements, - }); - const runLocal = vi.fn(async () => ({ meta: { durationMs: 1 } })); - - await expect( - provider.executeTurn( - { - sessionId: SESSION_ID, - sessionKey: SESSION_KEY, - agentId: "main", - runId: "run-reclaimed-unavailable", - }, - turn("run-reclaimed-unavailable"), - runLocal, - ), - ).rejects.toThrow("Reclaimed worker placement requires redispatch"); - expect(runLocal).not.toHaveBeenCalled(); - expect(placements.get(SESSION_ID)).toMatchObject({ state: "reclaimed", turnClaim: null }); - }); - it("does not fall back locally when reclaimed redispatch fails", async () => { seedReclaimedPlacement(); const provider = createWorkerSessionTurnPlacementProvider({ 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 df4f9904dfd2..6d6096bf55bd 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.test-support.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.test-support.ts @@ -23,6 +23,7 @@ import { type WorkerSessionPlacementStore, } from "./placement-store.js"; import { createWorkerSessionTurnPlacementProvider as createRawWorkerSessionTurnPlacementProvider } from "./worker-turn-launcher.js"; +import { createWorkerWorkspaceOperationCoordinator } from "./workspace-operation-coordinator.js"; export type WorkerTurnLauncherOptions = Parameters< typeof createRawWorkerSessionTurnPlacementProvider @@ -92,12 +93,25 @@ export function setWorkerTurnSessionTarget(target: typeof sessionTarget): typeof return target; } +type DefaultedWorkerTurnLauncherOption = + | "recoverPendingWorkspaceResult" + | "redispatchReclaimed" + | "resolveWorkspacePath" + | "workspaceOperations"; + export function createWorkerSessionTurnPlacementProvider( - options: Omit & - Partial>, + options: Omit & + Partial>, ) { return createRawWorkerSessionTurnPlacementProvider({ + recoverPendingWorkspaceResult: async () => { + throw new Error("unexpected pending workspace recovery"); + }, + redispatchReclaimed: async () => { + throw new Error("unexpected reclaimed placement redispatch"); + }, resolveWorkspacePath: async () => root, + workspaceOperations: createWorkerWorkspaceOperationCoordinator(), ...options, }); } diff --git a/src/gateway/worker-environments/worker-turn-launcher.test.ts b/src/gateway/worker-environments/worker-turn-launcher.test.ts index 04c29de27b59..753893123d35 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.test.ts @@ -123,57 +123,6 @@ describe("worker turn launcher local placement", () => { expect(placements.list()).toEqual([]); }); - it("keeps recovery-only admission invisible for sessions without durable placement", async () => { - const provider = createWorkerSessionTurnPlacementProvider({ - admitNewPlacements: false, - environments: unusedEnvironments(), - placements, - }); - - await provider.executeTurn( - { sessionId: SESSION_ID, sessionKey: SESSION_KEY, agentId: "main", runId: "run-local" }, - turn("run-local"), - async () => ({ meta: { durationMs: 1 } }), - ); - await provider.executeLocalTurn( - { sessionId: SESSION_ID, sessionKey: SESSION_KEY, agentId: "main", runId: "run-cli" }, - async () => ({ kind: "cli" }), - ); - - expect(placements.list()).toEqual([]); - }); - - it("still admits an existing local placement in recovery-only mode", async () => { - const seedClaim = placements.claimTurn({ - sessionId: SESSION_ID, - sessionKey: SESSION_KEY, - agentId: "main", - claimId: "seed-local-placement", - runId: "seed-local-placement", - owner: { kind: "local" }, - }); - placements.releaseTurn(seedClaim); - const provider = createWorkerSessionTurnPlacementProvider({ - admitNewPlacements: false, - environments: unusedEnvironments(), - placements, - }); - - await provider.executeTurn( - { sessionId: SESSION_ID, runId: "run-existing-local" }, - turn("run-existing-local"), - async () => { - expect(placements.get(SESSION_ID)?.turnClaim).toMatchObject({ - owner: "local", - runId: "run-existing-local", - }); - return { meta: { durationMs: 1 } }; - }, - ); - - expect(placements.get(SESSION_ID)).toMatchObject({ state: "local", turnClaim: null }); - }); - it("holds a local placement claim around CLI execution", async () => { const environments = unusedEnvironments(); const provider = createWorkerSessionTurnPlacementProvider({ environments, placements }); diff --git a/src/gateway/worker-environments/worker-turn-launcher.ts b/src/gateway/worker-environments/worker-turn-launcher.ts index 53c2ca804553..10ecbffd4a2e 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.ts @@ -47,10 +47,7 @@ import { windowInitialMessages, } from "./worker-turn-payload.js"; import { resolveWorkerTurnTranscriptTarget } from "./worker-turn-transcript-target.js"; -import { - createWorkerWorkspaceOperationCoordinator, - type WorkerWorkspaceOperationCoordinator, -} from "./workspace-operation-coordinator.js"; +import type { WorkerWorkspaceOperationCoordinator } from "./workspace-operation-coordinator.js"; import { executeRemoteExecTurn, reconcileWorkspaceAfterTurn, @@ -61,13 +58,12 @@ import { type ReclaimedWorkerPlacement = Extract; type WorkerTurnLauncherOptions = { - admitNewPlacements?: boolean; environments: WorkerTurnEnvironmentService; placements: WorkerSessionPlacementStore; resolveWorkspacePath: (identity: ReturnType) => Promise; - recoverPendingWorkspaceResult?: (environmentId: string) => Promise; - workspaceOperations?: WorkerWorkspaceOperationCoordinator; - redispatchReclaimed?: (placement: ReclaimedWorkerPlacement) => Promise; + recoverPendingWorkspaceResult: (environmentId: string) => Promise; + workspaceOperations: WorkerWorkspaceOperationCoordinator; + redispatchReclaimed: (placement: ReclaimedWorkerPlacement) => Promise; }; async function executeLocalTurn(params: { @@ -382,8 +378,6 @@ async function executeWorkerTurn(params: { } export function createWorkerSessionTurnPlacementProvider(options: WorkerTurnLauncherOptions) { - const workspaceOperations = - options.workspaceOperations ?? createWorkerWorkspaceOperationCoordinator(); const provider: SessionPlacementAdmissionProvider & { resolveSandbox(params: { agentId: string; @@ -433,18 +427,11 @@ export function createWorkerSessionTurnPlacementProvider(options: WorkerTurnLaun return sandbox; }, async executeLocalTurn(claim: LocalTurnPlacementClaim, runLocal: () => Promise) { - if (!options.placements.get(claim.sessionId) && options.admitNewPlacements === false) { - return await runLocal(); - } return await executeLocalTurn({ claim, placements: options.placements, runLocal }); }, async executeTurn(claim, turn, runLocal, onAdmitted) { const current = options.placements.get(claim.sessionId); - if ( - !current && - (options.admitNewPlacements === false || - (turn.modelRun === true && !claim.sessionKey?.trim())) - ) { + if (!current && turn.modelRun === true && !claim.sessionKey?.trim()) { return await runLocal(); } if (!current || current.state === "local") { @@ -452,9 +439,6 @@ export function createWorkerSessionTurnPlacementProvider(options: WorkerTurnLaun } let routablePlacement = current; if (routablePlacement.state === "reclaimed") { - if (!options.redispatchReclaimed) { - throw new Error("Reclaimed worker placement requires redispatch"); - } emitAgentRunStatusEvent({ runId: claim.runId, phase: "provisioning_environment", @@ -516,7 +500,7 @@ export function createWorkerSessionTurnPlacementProvider(options: WorkerTurnLaun placement, placements: options.placements, localWorkspaceDir, - workspaceOperations, + workspaceOperations: options.workspaceOperations, turn, turnClaim, }; @@ -537,7 +521,7 @@ export function createWorkerSessionTurnPlacementProvider(options: WorkerTurnLaun // A recovery sweep owns the still-live worker claim. Teardown here // could discard the terminal event's durably fenced file results. options.placements.handoffWorkspaceResultRecovery(turnClaim); - await options.recoverPendingWorkspaceResult?.(placement.environmentId); + await options.recoverPendingWorkspaceResult(placement.environmentId); throw error; } if (error instanceof WorkerRunnerUnavailableError && !handedOff) { diff --git a/src/node-host/node-worker-capacity.ts b/src/node-host/node-worker-capacity.ts index 550f1b989533..714c9bbc704e 100644 --- a/src/node-host/node-worker-capacity.ts +++ b/src/node-host/node-worker-capacity.ts @@ -18,7 +18,6 @@ type NodeWorkerCapacityOptions = { capacity?: number; capacityWaitMs?: number; onAvailabilityChanged?: (available: boolean) => void; - onTerminal?: () => void; }; function capacityAbortReason(signal: AbortSignal): Error { @@ -41,7 +40,6 @@ export class NodeWorkerCapacity { private readonly capacity: number; private readonly waitMs: number; private readonly onAvailabilityChanged?: (available: boolean) => void; - private readonly onTerminal?: () => void; private readonly waiters = new Set<() => void>(); private readonly closeAbort = new AbortController(); private availability?: boolean; @@ -53,7 +51,6 @@ export class NodeWorkerCapacity { this.capacity = options.capacity ?? DEFAULT_WORKER_CAPACITY; this.waitMs = options.capacityWaitMs ?? DEFAULT_CAPACITY_WAIT_MS; this.onAvailabilityChanged = options.onAvailabilityChanged; - this.onTerminal = options.onTerminal; if (!Number.isSafeInteger(this.capacity) || this.capacity < 1) { throw new Error("node worker capacity must be a positive safe integer"); } @@ -117,7 +114,6 @@ export class NodeWorkerCapacity { const receipt = this.store.finish(params); if (notify && receipt.state !== "pending" && receipt.state !== "running") { this.changed(); - this.onTerminal?.(); } return receipt; } @@ -128,7 +124,6 @@ export class NodeWorkerCapacity { const receipt = this.store.finishCancelled(params); if (receipt && receipt.state !== "pending" && receipt.state !== "running") { this.changed(); - this.onTerminal?.(); } return receipt; } diff --git a/test/scripts/test-projects.test.ts b/test/scripts/test-projects.test.ts index 68364c9594cd..240492e9aafa 100644 --- a/test/scripts/test-projects.test.ts +++ b/test/scripts/test-projects.test.ts @@ -1988,6 +1988,20 @@ describe("scripts/test-projects changed-target routing", () => { ); }); + it("routes worker launcher changes through every split owner suite", () => { + expectChangedTargets( + ["src/gateway/worker-environments/worker-turn-launcher.ts"], + [ + "src/gateway/worker-environments/worker-turn-launcher.test.ts", + "src/gateway/worker-environments/worker-turn-launcher-claim-admission.test.ts", + "src/gateway/worker-environments/worker-turn-launcher-failure-recovery.test.ts", + "src/gateway/worker-environments/worker-turn-launcher-reclaimed-placement.test.ts", + "src/gateway/worker-environments/worker-turn-launcher-remote-handoff.test.ts", + "src/gateway/worker-environments/worker-turn-launcher-terminal-results.test.ts", + ], + ); + }); + it("keeps unknown root surfaces cheap by default", () => { expect( resolveChangedTargetArgs(["--changed", "origin/main"], process.cwd(), () => [