fix(gateway): preserve device capacity failures (#124574)

src/gateway/server-methods/sessions.dispatch.device.test.ts

@@ -1,6 +1,26 @@
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
 import { beforeEach, describe, expect, it, vi } from "vitest";
+import {
+  GATEWAY_CLIENT_IDS,
+  GATEWAY_CLIENT_MODES,
+} from "../../../packages/gateway-protocol/src/client-info.js";
 import { ErrorCodes } from "../../../packages/gateway-protocol/src/index.js";
+import type { PairedDevice } from "../../infra/device-pairing.types.js";
+import { NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE } from "../../infra/node-runner-inventory.js";
+import {
+  closeOpenClawStateDatabaseForTest,
+  openOpenClawStateDatabase,
+} from "../../state/openclaw-state-db.js";
+import type { NodeWorkerSupervisorNodeProof } from "../node-registry-private.js";
+import {
+  bindDeviceWorkerAvailability,
+  createDeviceWorkerRuntime,
+} from "../worker-environments/device-provider.js";
+import { createHarness } from "../worker-environments/placement-dispatch-test-harness.js";
 import type { WorkerSessionPlacementRecord } from "../worker-environments/placement-store.js";
+import { createWorkerSessionPlacementStore } from "../worker-environments/placement-store.js";
 import {
   dispatchTestSessionId,
   dispatchTestSessionKey,
@@ -12,6 +32,39 @@ import {
 
 const dispatchTestMocks = getDispatchTestMocks();
 
+function pairedNode(deviceId: string): PairedDevice {
+  return {
+    deviceId,
+    publicKey: `public-key-${deviceId}`,
+    role: "node",
+    roles: ["node"],
+    tokens: {
+      node: {
+        token: "fixture-token",
+        role: "node",
+        scopes: [],
+        createdAtMs: 1,
+      },
+    },
+    createdAtMs: 1,
+    approvedAtMs: 1,
+  };
+}
+
+function connectedNode(deviceId: string, capacity: "available" | "full") {
+  return {
+    nodeId: deviceId,
+    connId: `conn-${deviceId}`,
+    pairingIdentity: `identity-${deviceId}`,
+    pairingGeneration: `generation-${deviceId}`,
+    clientId: GATEWAY_CLIENT_IDS.NODE_HOST,
+    clientMode: GATEWAY_CLIENT_MODES.NODE,
+    protocolFeature: NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE,
+    workerHost: { enabled: true, capacity },
+    commands: ["system.run"],
+  } satisfies NodeWorkerSupervisorNodeProof;
+}
+
 describe("sessions.dispatch device targets", () => {
   beforeEach(() => {
     vi.clearAllMocks();
@@ -81,7 +134,7 @@ describe("sessions.dispatch device targets", () => {
     );
   });
 
-  it("rejects a device target without a connected session-capable pairing", async () => {
+  it("returns a device dispatch failure to the operator", async () => {
     dispatchTestMocks.resolveTarget.mockReturnValue(
       makeSessionTarget({
         sessionId: dispatchTestSessionId,
@@ -96,9 +149,7 @@ describe("sessions.dispatch device targets", () => {
     const dispatch = vi
       .fn()
       .mockRejectedValue(
-        new Error(
-          "device worker requires a connected current node host; reconnect or reprovision: device-1",
-        ),
+        new Error("device worker node is not connected: device-1; reconnect it before retrying"),
       );
     const respond = await invokeSessionDispatch(
       makeDispatchTestContext({
@@ -114,8 +165,87 @@ describe("sessions.dispatch device targets", () => {
       undefined,
       expect.objectContaining({
         code: ErrorCodes.UNAVAILABLE,
-        message: expect.stringContaining("reconnect or reprovision"),
+        message: expect.stringContaining("reconnect"),
       }),
     );
   });
+
+  it.each([
+    {
+      name: "full",
+      nodes: [connectedNode("device-1", "full")],
+      expectedMessage: "at capacity (all worker slots in use)",
+      rejectedMessage: "reconnect",
+    },
+    {
+      name: "disconnected",
+      nodes: [],
+      expectedMessage: "reconnect",
+      rejectedMessage: "at capacity",
+    },
+  ])(
+    "carries a $name node rejection through the placement row and operator response",
+    async ({ nodes, expectedMessage, rejectedMessage }) => {
+      const root = await fs.mkdtemp(
+        path.join(await fs.realpath(os.tmpdir()), "openclaw-session-dispatch-device-"),
+      );
+      try {
+        const database = openOpenClawStateDatabase({ env: { OPENCLAW_STATE_DIR: root } });
+        const placements = createWorkerSessionPlacementStore({ database, now: () => 1_000 });
+        const harness = createHarness(placements);
+        const runtime = createDeviceWorkerRuntime({
+          getPairedDevice: async (deviceId) => pairedNode(deviceId),
+        });
+        runtime.bindNodeTransport({
+          listCurrentNodes: async () => nodes,
+          isCurrent: () => true,
+          invoke: async () => ({ ok: false }),
+        });
+        bindDeviceWorkerAvailability(harness.environments, runtime.resolveAvailability);
+
+        dispatchTestMocks.resolveTarget.mockReturnValue(
+          makeSessionTarget({
+            sessionId: dispatchTestSessionId,
+            worktree: { id: "worktree-1", branch: "openclaw/device-test", repoRoot: "/repo" },
+          }),
+        );
+        dispatchTestMocks.findLiveByOwner.mockReturnValue({
+          id: "worktree-1",
+          ownerKind: "session",
+          ownerId: dispatchTestSessionKey,
+        });
+        const respond = await invokeSessionDispatch(
+          makeDispatchTestContext({
+            workerPlacementDispatchService: harness.service,
+            workerSessionPlacementService: placements,
+          }),
+          { deviceId: "device-1" },
+        );
+
+        const placement = placements.get(dispatchTestSessionId);
+        expect(placement).toMatchObject({
+          state: "failed",
+          recoveryError: expect.stringContaining(expectedMessage),
+          terminalReason: expect.stringContaining(expectedMessage),
+        });
+        expect(placement?.recoveryError).not.toContain(rejectedMessage);
+        expect(respond).toHaveBeenCalledWith(
+          false,
+          undefined,
+          expect.objectContaining({
+            code: ErrorCodes.UNAVAILABLE,
+            message: expect.stringContaining(expectedMessage),
+          }),
+        );
+        expect(respond).not.toHaveBeenCalledWith(
+          false,
+          undefined,
+          expect.objectContaining({ message: expect.stringContaining(rejectedMessage) }),
+        );
+      } finally {
+        closeOpenClawStateDatabaseForTest();
+        await fs.rm(root, { recursive: true, force: true });
+      }
+    },
+  );
 });

src/gateway/worker-environments/device-provider.test.ts

@@ -118,7 +118,7 @@ describe("device worker provider", () => {
       name: "offline device",
       getPairedDevice: async () => pairedDevice(),
       listCurrentNodes: async () => [],
-      expectedMessage: `device worker node is not connected: ${DEVICE_ID}`,
+      expectedMessage: `device worker node is not connected: ${DEVICE_ID}; reconnect it before retrying`,
     },
     {
       name: "connected node at capacity",

src/gateway/worker-environments/device-provider.ts

@@ -49,6 +49,22 @@ export async function resolveDeviceWorkerAvailability(
   return resolveAvailability ? await resolveAvailability(deviceId) : { available: false };
 }
 
+export function deviceUnavailableText(deviceId: string, availability: DeviceWorkerAvailability) {
+  if (availability.issue) {
+    return formatNodeRunnerUpdateRequired(deviceId, availability.issue);
+  }
+  switch (availability.unavailableReason) {
+    case "unpaired":
+      return `device worker is not a paired node host: ${deviceId}`;
+    case "disconnected":
+      return `device worker node is not connected: ${deviceId}; reconnect it before retrying`;
+    case "at-capacity":
+      return `device worker is at capacity (all worker slots in use): ${deviceId}; retry after a running turn completes`;
+    default:
+      return `device worker availability is unknown: ${deviceId}; verify the node host is paired and connected, then retry`;
+  }
+}
+
 export function bindDeviceWorkerReconciliation(
   service: object,
   reconcile: DeviceWorkerReconciliation,
@@ -127,20 +143,7 @@ export function createDeviceWorkerRuntime(options: DeviceWorkerRuntimeOptions) {
       const deviceId = requireDeviceId(profile);
       const availability = await resolveAvailability(deviceId);
       if (!availability.available) {
-        if (availability.issue) {
-          throw new WorkerProviderError(
-            formatNodeRunnerUpdateRequired(deviceId, availability.issue),
-          );
-        }
-        if (availability.unavailableReason === "unpaired") {
-          throw new WorkerProviderError(`device worker is not a paired node host: ${deviceId}`);
-        }
-        if (availability.unavailableReason === "disconnected") {
-          throw new WorkerProviderError(`device worker node is not connected: ${deviceId}`);
-        }
-        throw new WorkerProviderError(
-          `device worker is at capacity (all worker slots in use): ${deviceId}; retry after a running turn completes`,
-        );
+        throw new WorkerProviderError(deviceUnavailableText(deviceId, availability));
       }
       return {
         leaseId: deviceLeaseId(deviceId, operationId),

src/gateway/worker-environments/placement-dispatch.ts

@@ -1,7 +1,6 @@
 import { randomUUID } from "node:crypto";
-import { formatNodeRunnerUpdateRequired } from "../../infra/node-runner-inventory.js";
 import { supportsWorkerExecutionContextLaunch } from "./admission.js";
-import { resolveDeviceWorkerAvailability } from "./device-provider.js";
+import * as device from "./device-provider.js";
 import {
   createPlacementFailureActions,
   isUnavailableEnvironment,
@@ -165,15 +164,14 @@ export function createWorkerPlacementDispatchService(options: WorkerPlacementDis
           return placement;
         },
       });
-      const deviceAvailability = request.deviceId
-        ? await resolveDeviceWorkerAvailability(environments, request.deviceId)
-        : undefined;
-      if (request.deviceId && !deviceAvailability?.available) {
-        throw new Error(
-          deviceAvailability?.issue
-            ? formatNodeRunnerUpdateRequired(request.deviceId, deviceAvailability.issue)
-            : `device worker requires a connected current node host; reconnect or reprovision: ${request.deviceId}`,
+      if (request.deviceId) {
+        const availability = await device.resolveDeviceWorkerAvailability(
+          environments,
+          request.deviceId,
         );
+        if (!availability.available) {
+          throw new Error(device.deviceUnavailableText(request.deviceId, availability));
+        }
       }
       const localPath = await options.resolveWorkspacePath(request);
       const idempotencyKey = `session-dispatch:${request.sessionId}:${placement.generation}`;