diff --git a/docs/nodes/index.md b/docs/nodes/index.md index 65defb2acc31..508494bfca64 100644 --- a/docs/nodes/index.md +++ b/docs/nodes/index.md @@ -513,6 +513,11 @@ visible but disabled with an actionable reason. Enable hosting with setting, then restart the node host. Update-required hosts must be upgraded and restarted before selection. +While node inventory refreshes, or if that refresh fails, the picker keeps known +devices visible but disables remote selection and Start until fresh inventory +arrives. Local remains selectable; cached worker slots never authorize a new +remote session. + Choose **Any available node** to let the Gateway select an eligible paired, connected session host. For OpenClaw worker turns, it selects the host with the most available worker slots and breaks ties by device ID. Runtimes that do not diff --git a/src/cli/update-cli.test.ts b/src/cli/update-cli.test.ts index 7b4b4b3567a5..dd817f155b5d 100644 --- a/src/cli/update-cli.test.ts +++ b/src/cli/update-cli.test.ts @@ -48,6 +48,9 @@ const serviceLoaded = vi.fn(); const serviceEnabled = vi.fn(); const serviceStop = vi.fn(); const serviceRestart = vi.fn(); +// A fixed Gateway PID can collide with the updater and trigger its self-stop safeguard. +const gatewayFixturePid = process.pid + 1; +const unrelatedGatewayFixturePid = process.pid + 2; const isDefaultInstallIdentity = vi.hoisted(() => vi.fn<(env?: NodeJS.ProcessEnv, homedir?: () => string, platform?: NodeJS.Platform) => boolean>( () => true, @@ -1350,7 +1353,11 @@ describe("update-cli", () => { }, }); serviceLoaded.mockResolvedValue(true); - serviceReadRuntime.mockResolvedValue({ status: "running", pid: 4242, state: "running" }); + serviceReadRuntime.mockResolvedValue({ + status: "running", + pid: gatewayFixturePid, + state: "running", + }); }; const mockStoppedManagedGitGateway = () => { @@ -1363,7 +1370,11 @@ describe("update-cli", () => { serviceLoaded.mockResolvedValue(false); serviceLoaded.mockResolvedValueOnce(true); serviceReadRuntime.mockResolvedValue({ status: "stopped", pid: null, state: "stopped" }); - serviceReadRuntime.mockResolvedValueOnce({ status: "running", pid: 4242, state: "running" }); + serviceReadRuntime.mockResolvedValueOnce({ + status: "running", + pid: gatewayFixturePid, + state: "running", + }); }; const expectFailedManagedGitRestart = (message: string) => { @@ -1618,7 +1629,7 @@ describe("update-cli", () => { ); serviceReadRuntime.mockResolvedValue({ status: "running", - pid: 4242, + pid: gatewayFixturePid, state: "running", }); mockGetSelfAndAncestorPidsSync.mockReturnValue(new Set([process.pid])); @@ -1627,7 +1638,7 @@ describe("update-cli", () => { inspectPortUsage.mockResolvedValue({ port: 18789, status: "busy", - listeners: [{ pid: 4242, command: "openclaw-gateway" }], + listeners: [{ pid: gatewayFixturePid, command: "openclaw-gateway" }], hints: [], }); classifyPortListener.mockReturnValue("gateway"); @@ -1865,10 +1876,14 @@ describe("update-cli", () => { OPENCLAW_GATEWAY_PORT: "19222", OPENCLAW_SERVICE_MARKER: "openclaw", OPENCLAW_SERVICE_KIND: "gateway", - [GATEWAY_SERVICE_RUNTIME_PID_ENV]: "7777", + [GATEWAY_SERVICE_RUNTIME_PID_ENV]: String(unrelatedGatewayFixturePid), }); serviceLoaded.mockResolvedValue(true); - serviceReadRuntime.mockResolvedValue({ status: "running", pid: 4242, state: "running" }); + serviceReadRuntime.mockResolvedValue({ + status: "running", + pid: gatewayFixturePid, + state: "running", + }); vi.mocked(readConfigFileSnapshot).mockImplementation(async () => process.env.OPENCLAW_PROFILE === "work" ? managedSnapshot : baseSnapshot, ); @@ -1984,7 +1999,11 @@ describe("update-cli", () => { OPENCLAW_GATEWAY_PORT: "19222", }); serviceLoaded.mockResolvedValue(true); - serviceReadRuntime.mockResolvedValue({ status: "running", pid: 4242, state: "running" }); + serviceReadRuntime.mockResolvedValue({ + status: "running", + pid: gatewayFixturePid, + state: "running", + }); vi.mocked(resolveGatewayInstallEntrypoint).mockResolvedValueOnce(undefined); vi.mocked(resolveGatewayInstallEntrypoint).mockResolvedValueOnce( "/tmp/openclaw-updated-entry.mjs", @@ -2039,7 +2058,11 @@ describe("update-cli", () => { OPENCLAW_GATEWAY_PORT: "19222", }); serviceLoaded.mockResolvedValue(true); - serviceReadRuntime.mockResolvedValue({ status: "running", pid: 4242, state: "running" }); + serviceReadRuntime.mockResolvedValue({ + status: "running", + pid: gatewayFixturePid, + state: "running", + }); prepareRestartScript.mockResolvedValue(null); vi.mocked(runDaemonRestart).mockResolvedValue(true); let doctorProfile: string | undefined; @@ -2202,7 +2225,10 @@ describe("update-cli", () => { it("does not carry gateway service markers into the post-core update process", async () => { setupUpdatedRootRefresh(); - await runWithGatewayServiceEnv({ yes: true }, { [GATEWAY_SERVICE_RUNTIME_PID_ENV]: "7777" }); + await runWithGatewayServiceEnv( + { yes: true }, + { [GATEWAY_SERVICE_RUNTIME_PID_ENV]: String(unrelatedGatewayFixturePid) }, + ); const spawnEnv = spawnCall()?.[2]?.env; expect(spawnEnv?.OPENCLAW_SERVICE_MARKER).toBeUndefined(); @@ -4324,7 +4350,7 @@ describe("update-cli", () => { serviceReadCommand.mockResolvedValue(null); serviceReadRuntime.mockResolvedValueOnce({ status: "running", - pid: 4242, + pid: gatewayFixturePid, state: "running", }); @@ -4345,7 +4371,9 @@ describe("update-cli", () => { it("refuses package updates from inside the active gateway process tree", async () => { mockPackageInstallStatus(createCaseDir("openclaw-update")); serviceLoaded.mockResolvedValue(true); - mockGetSelfAndAncestorPidsSync.mockReturnValue(new Set([process.pid, 4242])); + mockGetSelfAndAncestorPidsSync.mockReturnValue( + new Set([process.pid, gatewayFixturePid]), + ); await updateCommand({ yes: true }); @@ -4353,7 +4381,7 @@ describe("update-cli", () => { expect(errors).toContain( "openclaw update detected it is running inside the gateway process tree.", ); - expect(errors).toContain("Gateway PID 4242 is an ancestor"); + expect(errors).toContain(`Gateway PID ${gatewayFixturePid} is an ancestor`); expect(defaultRuntime.exit).toHaveBeenCalledWith(1); expect(serviceStop).not.toHaveBeenCalled(); expect(packageInstallCommandCall()).toBeUndefined(); @@ -4364,18 +4392,21 @@ describe("update-cli", () => { serviceLoaded.mockResolvedValue(true); serviceReadRuntime.mockResolvedValue({ status: "running", - pid: 4242, + pid: gatewayFixturePid, state: "running", }); mockGetSelfAndAncestorPidsSync.mockReturnValue(new Set([process.pid])); - await runWithGatewayServiceEnv({ yes: true }, { [GATEWAY_SERVICE_RUNTIME_PID_ENV]: "4242" }); + await runWithGatewayServiceEnv( + { yes: true }, + { [GATEWAY_SERVICE_RUNTIME_PID_ENV]: String(gatewayFixturePid) }, + ); const errors = getErrorOutput(); expect(errors).toContain( "openclaw update detected it is running inside the gateway process tree.", ); - expect(errors).toContain("Gateway PID 4242 is an ancestor"); + expect(errors).toContain(`Gateway PID ${gatewayFixturePid} is an ancestor`); expect(defaultRuntime.exit).toHaveBeenCalledWith(1); expect(serviceStop).not.toHaveBeenCalled(); expect(packageInstallCommandCall()).toBeUndefined(); @@ -5177,7 +5208,7 @@ describe("update-cli", () => { }); serviceReadRuntime.mockResolvedValue( runtimeStatus === "running" - ? { status: "running", state: "running", pid: 4242 } + ? { status: "running", state: "running", pid: gatewayFixturePid } : { status: "stopped", state: "stopped" }, ); suspendScheduledTaskAutoStartForUpdate.mockResolvedValue(true); @@ -5264,7 +5295,7 @@ describe("update-cli", () => { portUsage: { port: 18789, status: "busy", - listeners: [{ pid: 4242, command: "openclaw-gateway" }], + listeners: [{ pid: gatewayFixturePid, command: "openclaw-gateway" }], hints: [], }, healthy: true, @@ -6953,11 +6984,11 @@ describe("update-cli", () => { prepareRestartScript.mockResolvedValue(null); serviceLoaded.mockResolvedValue(true); restartHealthTestControl.snapshot = { - runtime: { status: "running", pid: 4242 }, + runtime: { status: "running", pid: gatewayFixturePid }, portUsage: { port: 18789, status: "busy", - listeners: [{ pid: 4242, command: "openclaw-gateway" }], + listeners: [{ pid: gatewayFixturePid, command: "openclaw-gateway" }], hints: [], }, healthy: false, diff --git a/src/gateway/server-worker-placement-reclaim.ts b/src/gateway/server-worker-placement-reclaim.ts index 7b94bed27cc8..3a6fbbc9de67 100644 --- a/src/gateway/server-worker-placement-reclaim.ts +++ b/src/gateway/server-worker-placement-reclaim.ts @@ -54,6 +54,7 @@ export function createGatewayWorkerPlacementReclaimBarriers( sessionKey, agentId, authorize, + beforeDrain, begin, reclaim, }) => { @@ -68,6 +69,7 @@ export function createGatewayWorkerPlacementReclaimBarriers( scope: target.storePath, identities: lifecycleIdentities, prepare: async () => { + beforeDrain?.(); const { worktree } = resolveWorkerPlacementSessionTarget({ sessionRuntime, config: getRuntimeConfig(), @@ -84,7 +86,6 @@ export function createGatewayWorkerPlacementReclaimBarriers( ); } worktreePath = worktree.path; - // Automatic reclaim must recheck admission after fencing, before interrupting live work. authorize?.(); const released = await interruptSessionWorkAdmissions({ scope: target.storePath, @@ -108,6 +109,8 @@ export function createGatewayWorkerPlacementReclaimBarriers( // Sharing mutations use this lifecycle fence too. Reauthorize after every wait and // immediately before drain so revoked callers cannot commit stale placement authority. authorize?.(); + // Eligibility ends at this operation's drain, unlike caller authority during teardown. + beforeDrain?.(); const drainingPlacement = begin(); reclaimedPlacement = await reclaim(worktreePath, drainingPlacement, authorize); params.revokeSessionAuthority({ sessionId, sessionKeys: lifecycleIdentities }); diff --git a/src/gateway/server.sessions.recover.test.ts b/src/gateway/server.sessions.recover.test.ts index f7eef74b75e7..a22670c9adab 100644 --- a/src/gateway/server.sessions.recover.test.ts +++ b/src/gateway/server.sessions.recover.test.ts @@ -13,7 +13,10 @@ import { replaceSessionEntry, } from "../config/sessions/session-accessor.js"; import { addSessionMember, removeSessionMember } from "../config/sessions/session-sharing-store.js"; -import { runExclusiveSessionLifecycleMutation } from "../sessions/session-lifecycle-admission.js"; +import { + beginSessionWorkAdmission, + runExclusiveSessionLifecycleMutation, +} from "../sessions/session-lifecycle-admission.js"; import { createDeferredCore } from "../shared/deferred.js"; import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js"; import { ensureProfileForEmail, setUserProfileRole } from "../state/user-profiles.js"; @@ -240,6 +243,120 @@ test("sessions.recover settles its active placement before archiving a real sess expect(reclaim).toHaveBeenCalledOnce(); }); +test.each(["before-interrupt", "before-drain"] as const)( + "automatic reclaim rechecks eligibility after waiting %s", + async (phase) => { + const { dir, storePath } = await createSessionStoreDir(); + const sessionKey = `agent:main:dashboard:idle-reclaim-${phase}`; + const sessionId = `idle-reclaim-${phase}`; + const stateDir = process.env.OPENCLAW_STATE_DIR; + if (!stateDir) { + throw new Error("gateway test state directory is unavailable"); + } + const repoRoot = await initializeManagedWorktreeTestRepository(dir); + const worktree = await materializeManagedWorktreeFixture({ + env: process.env, + name: sessionId, + now: Date.now(), + ownerKind: "session", + ownerId: sessionKey, + repoRoot, + stateDir, + }); + await writeSessionStore({ + entries: { + [sessionKey]: sessionStoreEntry(sessionId, { + spawnedCwd: worktree.path, + worktree: { + id: worktree.id, + branch: worktree.branch, + repoRoot, + canonicalWorkspaceDir: repoRoot, + }, + }), + }, + }); + const placement = recoveryWorkerPlacement({ sessionId, sessionKey, state: "active" }); + if (placement.state !== "active") { + throw new Error("expected active worker placement"); + } + const enteredWait = createDeferredCore(); + const releaseWait = createDeferredCore(); + const wait = async () => { + enteredWait.resolve(); + await releaseWait.promise; + }; + const onInterrupt = vi.fn(); + let releaseAdmission = () => {}; + const admission = + phase === "before-interrupt" + ? await beginSessionWorkAdmission({ + scope: storePath, + identities: [sessionId, sessionKey], + assertAllowed: () => {}, + onInterrupt: () => { + onInterrupt(); + releaseAdmission(); + }, + }) + : undefined; + releaseAdmission = () => admission?.release(); + const begin = vi.fn(() => ({ ...placement, state: "draining" as const })); + const reclaim = vi.fn(async () => { + throw new Error("ineligible worker must not be reclaimed"); + }); + const barriers = createGatewayWorkerPlacementReclaimBarriers({ + placements: { + get: () => placement, + waitForTurnClaimRelease: async () => { + if (phase === "before-drain") { + await wait(); + } + }, + }, + loadSessionRuntime: async () => { + if (phase === "before-interrupt") { + await wait(); + } + return { + managedWorktrees, + resolveCanonicalSessionEntryFromStoreKeys, + resolveGatewaySessionStoreTargetWithStore, + }; + }, + revokeSessionAuthority: vi.fn(), + }); + let eligible = true; + const eligibilityError = new Error("worker is no longer idle"); + const reclaiming = barriers.runReclaimBarrier({ + sessionId, + sessionKey, + agentId: "main", + beforeDrain: () => { + if (!eligible) { + throw eligibilityError; + } + }, + begin, + reclaim, + }); + const rejected = expect(reclaiming).rejects.toBe(eligibilityError); + try { + await enteredWait.promise; + eligible = false; + releaseWait.resolve(); + await rejected; + expect(onInterrupt).not.toHaveBeenCalled(); + expect(begin).not.toHaveBeenCalled(); + expect(reclaim).not.toHaveBeenCalled(); + } finally { + releaseWait.resolve(); + admission?.release(); + await Promise.allSettled([reclaiming]); + } + }, +); + test.each(["rejected", "unavailable", "stale-result"] as const)( "sessions.recover leaves its source and successor untouched when cloud reclaim is %s", async (failure) => { diff --git a/src/gateway/worker-environments/placement-dispatch-coordinator.ts b/src/gateway/worker-environments/placement-dispatch-coordinator.ts index 04b173f09a80..ad73bffa0df6 100644 --- a/src/gateway/worker-environments/placement-dispatch-coordinator.ts +++ b/src/gateway/worker-environments/placement-dispatch-coordinator.ts @@ -175,8 +175,8 @@ export function coordinateWorkerPlacementDispatch( } } }, - reclaim: async (request, authorize) => - await runExclusivePlacementOperation(() => service.reclaim(request, authorize)), + reclaim: async (request, authorize, beforeDrain) => + await runExclusivePlacementOperation(() => service.reclaim(request, authorize, beforeDrain)), reconcile: (mode) => runReconciliation(() => service.reconcile(mode)), reconcileActive: (environmentId) => environmentId === undefined diff --git a/src/gateway/worker-environments/placement-dispatch-test-harness.ts b/src/gateway/worker-environments/placement-dispatch-test-harness.ts index ad1977c3abaa..1a2acad9234f 100644 --- a/src/gateway/worker-environments/placement-dispatch-test-harness.ts +++ b/src/gateway/worker-environments/placement-dispatch-test-harness.ts @@ -452,8 +452,9 @@ export function createHarness( } : { requiredNodeCommands: [], consumesWorkerSlot: true }, isCurrentNodePlacement: options.isCurrentNodePlacement ?? (() => true), - runReclaimBarrier: async ({ authorize, begin, reclaim }) => { + runReclaimBarrier: async ({ authorize, beforeDrain, begin, reclaim }) => { authorize?.(); + beforeDrain?.(); return await reclaim(options.workspacePath ?? "/gateway/workspace", begin(), authorize); }, runFailedReclaimBarrier: async ({ authorize, reclaim }) => { diff --git a/src/gateway/worker-environments/placement-dispatch.ts b/src/gateway/worker-environments/placement-dispatch.ts index 7d882fe686f0..18cb7d987dd5 100644 --- a/src/gateway/worker-environments/placement-dispatch.ts +++ b/src/gateway/worker-environments/placement-dispatch.ts @@ -74,6 +74,7 @@ type WorkerReclaimPlacement = Extract WorkerDrainingDispatchPlacement; reclaim: ( localPath: string, @@ -336,10 +337,12 @@ export function createWorkerPlacementDispatchService(options: WorkerPlacementDis request: WorkerPlacementReclaimRequest, moveIntent?: WorkerPlacementMoveIntent, authorize?: WorkerPlacementAuthorization, + beforeDrain?: WorkerPlacementAuthorization, ): Promise => await options.runReclaimBarrier({ ...request, authorize, + beforeDrain, begin: () => { const current = placements.get(request.sessionId); if ((current?.state !== "active" && current?.state !== "draining") || current.turnClaim) { @@ -617,7 +620,9 @@ export function createWorkerPlacementDispatchService(options: WorkerPlacementDis const reclaim = async ( request: WorkerPlacementReclaimRequest, authorize?: WorkerPlacementAuthorization, + beforeDrain?: WorkerPlacementAuthorization, ): Promise => { + beforeDrain?.(); const current = placements.get(request.sessionId); if (current?.state === "reclaimed") { return current; @@ -663,7 +668,7 @@ export function createWorkerPlacementDispatchService(options: WorkerPlacementDis }, }); } - return await reclaimOnce(request, undefined, authorize); + return await reclaimOnce(request, undefined, authorize, beforeDrain); })().catch((error: unknown) => { // Another teardown path can win after this call has crossed its durable completion fence. // Report the committed terminal state instead of leaking a stale tunnel error to callers. diff --git a/src/gateway/worker-environments/placement-idle-sweep.test.ts b/src/gateway/worker-environments/placement-idle-sweep.test.ts index 167ba20ec8e7..dab138e9444e 100644 --- a/src/gateway/worker-environments/placement-idle-sweep.test.ts +++ b/src/gateway/worker-environments/placement-idle-sweep.test.ts @@ -31,6 +31,7 @@ describe("worker placement idle suspension", () => { options: { suspendAfter?: string | null; destroyFails?: boolean; + reclaim?: Parameters[0]["dispatch"]["reclaim"]; isPlacementOperationInFlight?: (sessionId: string) => boolean; getSessionWorkAdmissionCheck?: (identity: { sessionId: string; @@ -53,7 +54,7 @@ describe("worker placement idle suspension", () => { const idleSweep = createWorkerPlacementIdleSweep({ placements, environments: harness.environments, - dispatch: harness.service, + dispatch: { reclaim: options.reclaim ?? harness.service.reclaim }, getConfig: () => ({ cloudWorkers: { profiles: { @@ -307,6 +308,91 @@ describe("worker placement idle suspension", () => { expect(warn).not.toHaveBeenCalled(); }); + it("keeps a recently used worker when automatic reclaim waited behind another dispatch", async () => { + const dispatchStarted = createDeferredCore(); + const releaseDispatch = createDeferredCore(); + const reclaimQueued = createDeferredCore(); + const { harness, idleSweep, info, warn } = createIdleFixture({ + reclaim: (request, authorize, beforeDrain) => { + const pending = coordinated.reclaim(request, authorize, beforeDrain); + reclaimQueued.resolve(); + return pending; + }, + isPlacementOperationInFlight: (sessionId) => + coordinated.isPlacementOperationInFlight(sessionId), + getSessionWorkAdmissionCheck: async () => () => false, + }); + const active = await harness.service.dispatch(REQUEST); + const coordinated = coordinateWorkerPlacementDispatch({ + ...harness.service, + dispatch: async () => { + dispatchStarted.resolve(); + await releaseDispatch.promise; + return active; + }, + }); + const unrelatedDispatch = coordinated.dispatch({ + ...REQUEST, + sessionId: "another-session", + sessionKey: "agent:main:another-session", + }); + let sweeping: Promise | undefined; + try { + await dispatchStarted.promise; + nowMs += 60_000; + sweeping = idleSweep.sweep(); + await reclaimQueued.promise; + + const claim = claimWorkerTurn("turn-during-idle-reclaim-wait"); + nowMs += 1_000; + placements.releaseTurn(claim); + releaseDispatch.resolve(); + await Promise.all([unrelatedDispatch, sweeping]); + + expect(placements.get(REQUEST.sessionId)).toMatchObject({ + state: "active", + turnClaim: null, + updatedAtMs: nowMs, + }); + expect(harness.environments.destroy).not.toHaveBeenCalled(); + expect(info).not.toHaveBeenCalled(); + expect(warn).not.toHaveBeenCalled(); + + nowMs += 60_000; + await idleSweep.sweep(); + expect(placements.get(REQUEST.sessionId)?.state).toBe("reclaimed"); + expect(harness.environments.destroy).toHaveBeenCalledOnce(); + } finally { + releaseDispatch.resolve(); + await Promise.allSettled([unrelatedDispatch, sweeping]); + } + }); + + it("finishes its owned drain without rechecking idle eligibility during teardown", async () => { + let hasSessionWork = false; + const { harness, idleSweep, info, warn } = createIdleFixture({ + getSessionWorkAdmissionCheck: async () => () => hasSessionWork, + }); + await harness.service.dispatch(REQUEST); + const startTunnel = vi.mocked(harness.environments.startTunnel).getMockImplementation(); + if (!startTunnel) { + throw new Error("expected the fixture tunnel implementation"); + } + vi.mocked(harness.environments.startTunnel).mockImplementationOnce(async (...args) => { + expect(placements.get(REQUEST.sessionId)?.state).toBe("draining"); + hasSessionWork = true; + return await startTunnel(...args); + }); + nowMs += 60_000; + + await idleSweep.sweep(); + + expect(placements.get(REQUEST.sessionId)?.state).toBe("reclaimed"); + expect(harness.environments.destroy).toHaveBeenCalledOnce(); + expect(info).toHaveBeenCalledOnce(); + expect(warn).not.toHaveBeenCalled(); + }); + it("logs a failed reclaim once without immediately retrying provider teardown", async () => { const { harness, idleSweep, info, warn } = createIdleFixture({ destroyFails: true }); await harness.service.dispatch(REQUEST); diff --git a/src/gateway/worker-environments/placement-idle-sweep.ts b/src/gateway/worker-environments/placement-idle-sweep.ts index c176e7953864..9f38805f51ae 100644 --- a/src/gateway/worker-environments/placement-idle-sweep.ts +++ b/src/gateway/worker-environments/placement-idle-sweep.ts @@ -84,28 +84,21 @@ export function createWorkerPlacementIdleSweep(options: { agentId: placement.agentId, }; const hasSessionWork = await getSessionWorkAdmissionCheck?.(request); - const current = options.placements.get(placement.sessionId); - if ( - hasSessionWork?.() || - current?.state !== "active" || - current.generation !== placement.generation || - current.updatedAtMs !== placement.updatedAtMs || - current.turnClaim - ) { - continue; - } - const authorize = hasSessionWork - ? () => { - // Once drain starts, its lifecycle fence owns teardown; never abort it midway. - if ( - options.placements.get(placement.sessionId)?.state === "active" && - hasSessionWork() - ) { - throw new WorkerPlacementAutoSuspendBusyError(); - } - } - : undefined; - await options.dispatch.reclaim(request, authorize); + const beforeDrain = () => { + const current = options.placements.get(placement.sessionId); + if ( + hasSessionWork?.() || + current?.state !== "active" || + current.generation !== placement.generation || + current.environmentId !== placement.environmentId || + current.activeOwnerEpoch !== placement.activeOwnerEpoch || + current.updatedAtMs !== placement.updatedAtMs || + current.turnClaim + ) { + throw new WorkerPlacementAutoSuspendBusyError(); + } + }; + await options.dispatch.reclaim(request, undefined, beforeDrain); options.info( `auto-suspended ${placement.sessionKey} after ${suspendAfter} idle; wakes on next message`, ); diff --git a/src/gateway/worker-environments/service-contract.ts b/src/gateway/worker-environments/service-contract.ts index 307676f76c67..9a76c21c59bf 100644 --- a/src/gateway/worker-environments/service-contract.ts +++ b/src/gateway/worker-environments/service-contract.ts @@ -145,6 +145,7 @@ export type WorkerPlacementDispatchContract = { reclaim?( request: WorkerPlacementReclaimRequest, authorize?: WorkerPlacementAuthorization, + beforeDrain?: WorkerPlacementAuthorization, ): Promise>; forceDestroyEnvironment?( environmentId: string, diff --git a/ui/src/e2e/new-session-page.device-dispatch.e2e.test.ts b/ui/src/e2e/new-session-page.device-dispatch.e2e.test.ts index 23069954c113..1c22bbf45567 100644 --- a/ui/src/e2e/new-session-page.device-dispatch.e2e.test.ts +++ b/ui/src/e2e/new-session-page.device-dispatch.e2e.test.ts @@ -1,5 +1,6 @@ import { gatewayOriginScope } from "@openclaw/gateway-client/browser"; import { expect, it } from "vitest"; +import { CLOUD_PROFILE_RETRY_DELAYS_MS } from "../pages/new-session/cloud-profile-discovery.ts"; import { WORKSPACE, captureDeviceRuntimeUiProof, @@ -102,9 +103,15 @@ suite.define(() => { }); it.each(deviceTargets)( - "does not dispatch the $name device from stale capacity during a topology refresh", + "does not dispatch the $name device from stale capacity during a failed topology refresh", async ({ value }) => { - const context = await suite.browser.newContext({ locale: "en-US", serviceWorkers: "block" }); + const context = await suite.browser.newContext({ + locale: "en-US", + serviceWorkers: "block", + ...(process.env.OPENCLAW_CAPTURE_UI_PROOF === "1" + ? { recordVideo: { dir: ".artifacts/control-ui-e2e/device-runtime-gating" } } + : {}), + }); const page = await context.newPage(); const environment = { id: "node:paired-runner", @@ -133,7 +140,15 @@ suite.define(() => { await page.locator(".new-session-page__message").fill("require current worker capacity"); const start = page.getByRole("button", { name: "Start session" }); await expect.poll(() => start.isEnabled()).toBe(true); + await where.click(); + const selectedDevice = page.locator('[data-value="device:paired-runner"]'); + const automaticDevice = page.locator('[data-value="auto-device"]'); + const localDevice = page.locator('[data-value="gateway"]'); + await selectedDevice.waitFor({ state: "visible" }); + const clockTime = Date.now(); + await page.clock.install({ time: clockTime }); + await page.clock.pauseAt(clockTime + 1_000); await gateway.deferNext("environments.list"); const requestsBeforeRefresh = (await gateway.getRequests("environments.list")).length; await gateway.emitGatewayEvent("node.runnerInventory.changed", { @@ -148,6 +163,24 @@ suite.define(() => { ) .toBe(value === "auto-device" ? "true" : "paired-runner"); + await gateway.rejectDeferred("environments.list", { + code: "UNAVAILABLE", + message: "worker inventory is temporarily unavailable", + }); + await page.clock.runFor(CLOUD_PROFILE_RETRY_DELAYS_MS[0] - 1); + expect(await gateway.getRequests("environments.list")).toHaveLength( + requestsBeforeRefresh + 1, + ); + await captureDeviceRuntimeUiProof(page, `failed-topology-${value.replace(":", "-")}.png`); + expect(await start.isDisabled()).toBe(true); + expect(await selectedDevice.isDisabled()).toBe(true); + expect(await automaticDevice.isDisabled()).toBe(true); + expect(await localDevice.isEnabled()).toBe(true); + expect(await gateway.getRequests("sessions.create")).toHaveLength(0); + + await gateway.deferNext("environments.list"); + await page.clock.runFor(1); + await gateway.waitForRequest("environments.list", { after: requestsBeforeRefresh + 1 }); await gateway.resolveDeferred("environments.list", { environments: [{ ...environment, workerSlots: { total: 2, available: 0 } }], profiles: [], @@ -157,6 +190,15 @@ suite.define(() => { .poll(() => start.locator("xpath=..").getAttribute("content")) .toContain("No worker slots are available"); expect(await gateway.getRequests("sessions.create")).toHaveLength(0); + + await page.clock.resume(); + await gateway.emitGatewayEvent("node.runnerInventory.changed", { + nodeId: "paired-runner", + }); + await gateway.waitForRequest("environments.list", { after: requestsBeforeRefresh + 2 }); + await expect.poll(() => start.isEnabled()).toBe(true); + expect(await selectedDevice.isEnabled()).toBe(true); + expect(await automaticDevice.isEnabled()).toBe(true); } finally { await context.close(); } diff --git a/ui/src/pages/new-session/device-placement.ts b/ui/src/pages/new-session/device-placement.ts index bb4072e4e77c..8cb3b9df95a9 100644 --- a/ui/src/pages/new-session/device-placement.ts +++ b/ui/src/pages/new-session/device-placement.ts @@ -58,6 +58,7 @@ function unavailableReason( export function projectDevicePlacements( environments: readonly DraftEnvironment[] | null, requirement: DevicePlacementRequirement = DEFAULT_DEVICE_PLACEMENT, + placementDisabledReason?: string, ): DevicePlacementOption[] { const devices = (environments ?? []) .flatMap((environment) => { @@ -68,7 +69,7 @@ export function projectDevicePlacements( if (!deviceId) { return []; } - const disabledReason = unavailableReason(environment, requirement); + const disabledReason = placementDisabledReason ?? unavailableReason(environment, requirement); const facts = environmentMenuFacts(environment, { connected: environment.status === "available", }); @@ -84,7 +85,7 @@ export function projectDevicePlacements( { deviceId, label: environment.label ?? deviceId, - facts: visibleFacts, + facts: placementDisabledReason ? [placementDisabledReason] : visibleFacts, selectable: disabledReason === undefined, ...(disabledReason ? { disabledReason } : {}), }, diff --git a/ui/src/pages/new-session/draft-gateway-state.ts b/ui/src/pages/new-session/draft-gateway-state.ts index b12b86fa6c32..34395d067b74 100644 --- a/ui/src/pages/new-session/draft-gateway-state.ts +++ b/ui/src/pages/new-session/draft-gateway-state.ts @@ -161,6 +161,13 @@ export class DraftGatewayState { return this.cloudProfileTask.status === TaskStatus.PENDING; } + get deviceCatalogDisabledReason(): string | undefined { + // Cached cloud profiles survive refresh failures; live node capacity does not. + return this.cloudProfilesReadyValue && this.cloudProfileTask.status === TaskStatus.COMPLETE + ? undefined + : t("newSession.placementNotReady"); + } + get catalogRetrying(): boolean { return this.catalogRetryingValue; } diff --git a/ui/src/pages/new-session/draft-place-state.ts b/ui/src/pages/new-session/draft-place-state.ts index 2d381416635e..62281ad1d05c 100644 --- a/ui/src/pages/new-session/draft-place-state.ts +++ b/ui/src/pages/new-session/draft-place-state.ts @@ -178,7 +178,11 @@ export class DraftPlaceState { } devices() { - return projectDevicePlacements(this.gateway.environments, this.devicePlacementRequirement()); + return projectDevicePlacements( + this.gateway.environments, + this.devicePlacementRequirement(), + this.gateway.deviceCatalogDisabledReason, + ); } private findDevice(deviceId: string) { diff --git a/ui/src/pages/new-session/new-session-page.ts b/ui/src/pages/new-session/new-session-page.ts index 5c8b9a9a1999..d16ccfeee83d 100644 --- a/ui/src/pages/new-session/new-session-page.ts +++ b/ui/src/pages/new-session/new-session-page.ts @@ -397,7 +397,9 @@ export class NewSessionPage extends OpenClawLightDomElement { deviceId: this.place.deviceId, autoDevice: this.place.autoDevice, devicePlacement: this.place.devicePlacementRequirement(), - deviceDisabledReason: this.place.modelControl.devicePlacementUnsupportedReason(), + deviceDisabledReason: + this.place.modelControl.devicePlacementUnsupportedReason() ?? + this.gateway.deviceCatalogDisabledReason, }); const projectState = resolveProjectChip({ folder: this.place.folder, diff --git a/ui/src/pages/new-session/where-chip.ts b/ui/src/pages/new-session/where-chip.ts index 73a29d159fbe..0cb0ebca5a62 100644 --- a/ui/src/pages/new-session/where-chip.ts +++ b/ui/src/pages/new-session/where-chip.ts @@ -22,7 +22,6 @@ type WhereChipState = Readonly<{ cloudProfiles: readonly DraftCloudProfile[]; cloudMachines: readonly DraftMachineOption[]; selectedMachineId: string; - deviceDisabledReason?: string; autoDeviceDisabledReason?: string; }>; @@ -36,7 +35,11 @@ export function resolveWhereChip(params: { devicePlacement?: DevicePlacementRequirement; deviceDisabledReason?: string; }): WhereChipState { - const devices = projectDevicePlacements(params.environments, params.devicePlacement); + const devices = projectDevicePlacements( + params.environments, + params.devicePlacement, + params.deviceDisabledReason, + ); const autoDeviceDisabledReason = resolveAutomaticDevicePlacementDisabledReason( params.environments, devices, @@ -62,7 +65,6 @@ export function resolveWhereChip(params: { selectedMachineId: selectedMachine?.id ?? "", devices, cloudProfiles: params.cloudProfiles, - deviceDisabledReason: params.deviceDisabledReason, autoDeviceDisabledReason, }; } @@ -74,7 +76,6 @@ export function resolveWhereChip(params: { selectedMachineId: "", devices, cloudProfiles: params.cloudProfiles, - deviceDisabledReason: params.deviceDisabledReason, autoDeviceDisabledReason, }; } @@ -86,7 +87,6 @@ export function resolveWhereChip(params: { selectedMachineId: "", devices, cloudProfiles: params.cloudProfiles, - deviceDisabledReason: params.deviceDisabledReason, autoDeviceDisabledReason, }; } @@ -97,7 +97,6 @@ export function resolveWhereChip(params: { selectedMachineId: "", devices, cloudProfiles: params.cloudProfiles, - deviceDisabledReason: params.deviceDisabledReason, autoDeviceDisabledReason, }; } @@ -199,19 +198,16 @@ export function renderWhereChip(params: { params.submitting, )} ${params.state.devices.map((device) => { - const disabledReason = params.state.deviceDisabledReason ?? device.disabledReason; return renderSessionMenuItem( { value: `device:${device.deviceId}`, label: device.label, sub: device.subtitle, icon: icons.monitor, - facts: params.state.deviceDisabledReason - ? [params.state.deviceDisabledReason] - : device.facts, + facts: device.facts, checked: params.deviceId === device.deviceId, - disabled: Boolean(params.state.deviceDisabledReason) || !device.selectable, - title: disabledReason, + disabled: !device.selectable, + title: device.disabledReason, onSelect: () => params.onSelectDevice(device.deviceId), }, params.submitting,