From 4d872fbfbc74eb17a6e69d46df6f3876b136cefe Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 16 Aug 2026 10:04:46 -0700 Subject: [PATCH] feat(nodes): expose installed worker bundle status (#124640) * feat(nodes): expose installed worker bundle status * perf(nodes): defer bundle status validation * fix(nodes): tighten bundle status type contracts * docs(gateway): document node worker bundle status * refactor(gateway): split runner inventory runtime * test(ui): keep healthy device status quiet * fix(ui): preserve steer target ordering * test(ui): isolate catalog handoff lifecycle --- .../OpenClawProtocol/GatewayModels.swift | 16 + docs/gateway/protocol.md | 2 +- docs/nodes/index.md | 6 + docs/plan/runners.md | 11 +- .../src/schema/environments.test.ts | 36 ++ .../src/schema/environments.ts | 6 + .../gateway-protocol/src/schema/frames.ts | 1 + packages/sdk/src/index.ts | 1 + packages/sdk/src/types.ts | 5 + src/gateway/node-catalog.ts | 16 +- src/gateway/node-registry-private.ts | 196 +++++------ src/gateway/node-runner-inventory-runtime.ts | 106 ++++++ .../server-methods/environments.test.ts | 23 ++ src/gateway/server-methods/environments.ts | 7 + src/gateway/server-methods/nodes.read.ts | 6 + .../nodes.runner-inventory.test.ts | 122 +++++++ src/gateway/server-methods/nodes.test.ts | 50 ++- .../server.auth.default-token.suite.ts | 3 + .../server/ws-connection/connect-hello.ts | 1 + .../node-workspace-retain-coordinator.test.ts | 327 +++++++++++++++++- .../node-workspace-retain-coordinator.ts | 74 +++- src/infra/node-runner-inventory.ts | 15 +- .../invoke-worker-supervisor.test.ts | 51 ++- .../node-worker-bundle-installer.test.ts | 34 ++ src/node-host/node-worker-bundle-installer.ts | 21 +- .../node-worker-supervisor-commands.ts | 34 +- src/node-host/runner.test.ts | 26 ++ src/node-host/runner.ts | 9 + src/shared/node-list-types.ts | 5 + .../node-workspace-retain-protocol.test.ts | 20 ++ src/worker/node-workspace-retain-protocol.ts | 18 + test/vitest/vitest.ui-isolated-paths.mjs | 1 + ui/src/e2e/operator-admin.e2e.test.ts | 2 +- ui/src/i18n/locales/en.ts | 4 + ui/src/lib/nodes/inventory.test.ts | 38 ++ ui/src/lib/nodes/inventory.ts | 18 + ui/src/pages/chat/chat-gateway.test.ts | 3 +- ui/src/pages/chat/chat-send.test.ts | 1 + ui/src/pages/chat/steer-lifecycle.ts | 7 +- ui/src/pages/devices/devices-page.test.ts | 35 +- ui/src/pages/devices/devices-page.ts | 6 +- ui/src/pages/devices/view-inventory.ts | 13 +- ui/src/pages/devices/view.devices.test.ts | 43 ++- 43 files changed, 1272 insertions(+), 147 deletions(-) create mode 100644 src/gateway/node-runner-inventory-runtime.ts diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift index 5dc41cc2e7ec..478418bf2471 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift @@ -1956,6 +1956,7 @@ public struct EnvironmentSummary: Codable, Sendable { public let status: EnvironmentStatus public let platform: String? public let sessionhost: Bool? + public let workerbundle: AnyCodable? public let lastconnectedatms: Int? public let lastdisconnectedatms: Int? public let lastseenatms: Int? @@ -1973,6 +1974,7 @@ public struct EnvironmentSummary: Codable, Sendable { status: EnvironmentStatus, platform: String? = nil, sessionhost: Bool? = nil, + workerbundle: AnyCodable? = nil, lastconnectedatms: Int? = nil, lastdisconnectedatms: Int? = nil, lastseenatms: Int? = nil, @@ -1989,6 +1991,7 @@ public struct EnvironmentSummary: Codable, Sendable { self.status = status self.platform = platform self.sessionhost = sessionhost + self.workerbundle = workerbundle self.lastconnectedatms = lastconnectedatms self.lastdisconnectedatms = lastdisconnectedatms self.lastseenatms = lastseenatms @@ -2007,6 +2010,7 @@ public struct EnvironmentSummary: Codable, Sendable { case status case platform case sessionhost = "sessionHost" + case workerbundle = "workerBundle" case lastconnectedatms = "lastConnectedAtMs" case lastdisconnectedatms = "lastDisconnectedAtMs" case lastseenatms = "lastSeenAtMs" @@ -2044,6 +2048,7 @@ public struct EnvironmentsCreateResult: Codable, Sendable { public let status: EnvironmentStatus public let platform: String? public let sessionhost: Bool? + public let workerbundle: AnyCodable? public let lastconnectedatms: Int? public let lastdisconnectedatms: Int? public let lastseenatms: Int? @@ -2061,6 +2066,7 @@ public struct EnvironmentsCreateResult: Codable, Sendable { status: EnvironmentStatus, platform: String? = nil, sessionhost: Bool? = nil, + workerbundle: AnyCodable? = nil, lastconnectedatms: Int? = nil, lastdisconnectedatms: Int? = nil, lastseenatms: Int? = nil, @@ -2077,6 +2083,7 @@ public struct EnvironmentsCreateResult: Codable, Sendable { self.status = status self.platform = platform self.sessionhost = sessionhost + self.workerbundle = workerbundle self.lastconnectedatms = lastconnectedatms self.lastdisconnectedatms = lastdisconnectedatms self.lastseenatms = lastseenatms @@ -2095,6 +2102,7 @@ public struct EnvironmentsCreateResult: Codable, Sendable { case status case platform case sessionhost = "sessionHost" + case workerbundle = "workerBundle" case lastconnectedatms = "lastConnectedAtMs" case lastdisconnectedatms = "lastDisconnectedAtMs" case lastseenatms = "lastSeenAtMs" @@ -2132,6 +2140,7 @@ public struct EnvironmentsDestroyResult: Codable, Sendable { public let status: EnvironmentStatus public let platform: String? public let sessionhost: Bool? + public let workerbundle: AnyCodable? public let lastconnectedatms: Int? public let lastdisconnectedatms: Int? public let lastseenatms: Int? @@ -2149,6 +2158,7 @@ public struct EnvironmentsDestroyResult: Codable, Sendable { status: EnvironmentStatus, platform: String? = nil, sessionhost: Bool? = nil, + workerbundle: AnyCodable? = nil, lastconnectedatms: Int? = nil, lastdisconnectedatms: Int? = nil, lastseenatms: Int? = nil, @@ -2165,6 +2175,7 @@ public struct EnvironmentsDestroyResult: Codable, Sendable { self.status = status self.platform = platform self.sessionhost = sessionhost + self.workerbundle = workerbundle self.lastconnectedatms = lastconnectedatms self.lastdisconnectedatms = lastdisconnectedatms self.lastseenatms = lastseenatms @@ -2183,6 +2194,7 @@ public struct EnvironmentsDestroyResult: Codable, Sendable { case status case platform case sessionhost = "sessionHost" + case workerbundle = "workerBundle" case lastconnectedatms = "lastConnectedAtMs" case lastdisconnectedatms = "lastDisconnectedAtMs" case lastseenatms = "lastSeenAtMs" @@ -2236,6 +2248,7 @@ public struct EnvironmentsStatusResult: Codable, Sendable { public let status: EnvironmentStatus public let platform: String? public let sessionhost: Bool? + public let workerbundle: AnyCodable? public let lastconnectedatms: Int? public let lastdisconnectedatms: Int? public let lastseenatms: Int? @@ -2253,6 +2266,7 @@ public struct EnvironmentsStatusResult: Codable, Sendable { status: EnvironmentStatus, platform: String? = nil, sessionhost: Bool? = nil, + workerbundle: AnyCodable? = nil, lastconnectedatms: Int? = nil, lastdisconnectedatms: Int? = nil, lastseenatms: Int? = nil, @@ -2269,6 +2283,7 @@ public struct EnvironmentsStatusResult: Codable, Sendable { self.status = status self.platform = platform self.sessionhost = sessionhost + self.workerbundle = workerbundle self.lastconnectedatms = lastconnectedatms self.lastdisconnectedatms = lastdisconnectedatms self.lastseenatms = lastseenatms @@ -2287,6 +2302,7 @@ public struct EnvironmentsStatusResult: Codable, Sendable { case status case platform case sessionhost = "sessionHost" + case workerbundle = "workerBundle" case lastconnectedatms = "lastConnectedAtMs" case lastdisconnectedatms = "lastDisconnectedAtMs" case lastseenatms = "lastSeenAtMs" diff --git a/docs/gateway/protocol.md b/docs/gateway/protocol.md index 31c65e118f44..80a098abd01c 100644 --- a/docs/gateway/protocol.md +++ b/docs/gateway/protocol.md @@ -634,7 +634,7 @@ methods. Treat this as feature discovery, not a full enumeration of - `agents.workspace.list` and `agents.workspace.get` (`operator.read`) expose read-only, paginated browsing of an agent's workspace directory for clients in the trusted operator domain described in [Operator scopes](/gateway/operator-scopes). Requests accept workspace-relative paths only; reads stay confined to the realpathed workspace root (symlink and hardlink escapes rejected), size-capped, and limited to UTF-8 text plus common image types (base64). Responses do not expose the host workspace path. There are no write operations in this namespace. - `tasks.list`, `tasks.get`, and `tasks.cancel` expose the gateway task ledger to SDK and operator clients. See [Task ledger RPCs](#task-ledger-rpcs) below. - `artifacts.list`, `artifacts.get`, and `artifacts.download` expose transcript-derived artifact summaries and downloads for an explicit `sessionKey`, `runId`, or `taskId` scope. Run and task queries resolve the owning session server-side and only return transcript media with matching provenance; unsafe or local URL sources return unsupported downloads instead of fetching server-side. - - `environments.list` and `environments.status` remain available without cloud-worker profiles and preserve gateway-local and node environment discovery. Configured cloud workers and durable records left by earlier profiles add `worker` metadata with `providerId`, optional `leaseId`, `state`, `ageMs`, optional `idleMs`, and `attachedSessionIds`. Worker lifecycle states are `requested`, `provisioning`, `bootstrapping`, `ready`, `attached`, `idle`, `draining`, `destroying`, `destroyed`, `failed`, and `orphaned`. + - `environments.list` and `environments.status` remain available without cloud-worker profiles and preserve gateway-local and node environment discovery. Configured cloud workers and durable records left by earlier profiles add `worker` metadata with `providerId`, optional `leaseId`, `state`, `ageMs`, optional `idleMs`, and `attachedSessionIds`. Worker lifecycle states are `requested`, `provisioning`, `bootstrapping`, `ready`, `attached`, `idle`, `draining`, `destroying`, `destroyed`, `failed`, and `orphaned`. A connected node may also include `workerBundle: { status: "installed", version }` or `workerBundle: { status: "missing" }`. This optional observation is reconnect-scoped and reports validation of one Gateway-retained bundle; it is not launch authority. The public result never exposes the bundle hash, Gateway namespace, node filesystem path, receipt, or protocol-feature details. - `environments.create` (`{ profileId, idempotencyKey }`) provisions a worker from a configured plugin provider profile; retries with the same key reuse the durable operation. `environments.destroy` (`{ environmentId }`) requests idempotent teardown of a durable worker environment. Both require `operator.admin`, are control-plane writes, and return the same environment summary shape used by status responses. - `worker.desktop.observe` (`{ environmentId, control? }`, `operator.admin`) starts or reuses the environment's desktop forward and returns `{ transport, wsPath, expiresAtMs, control, vncPassword? }`. `wsPath` carries a single-use 60-second token for the Gateway's desktop observer WebSocket; reconnecting requires a fresh observe call. Environments with an observable desktop advertise `worker.desktop: true` in `environments.list`. The method is advertised only when the `cloudWorkers.desktop` lab is enabled. See [Cloud workers](/gateway/cloud-workers#desktop-interactive). - `agent.identity.get` returns the effective assistant identity for an agent or session. diff --git a/docs/nodes/index.md b/docs/nodes/index.md index a2437e396b11..be070256972b 100644 --- a/docs/nodes/index.md +++ b/docs/nodes/index.md @@ -438,6 +438,12 @@ contains its complete JavaScript dependency closure; the node does not install packages or execute lifecycle scripts. Later turns reuse the immutable artifact while its receipt still matches the Gateway's current build. +The Devices page shows the validated Gateway-owned worker version in the node's +metadata. If the retained artifact is missing or fails validation, Devices shows +a **worker missing** warning; start a new session on that device to reinstall the +current bundle. This status is observational and reconnect-scoped: launch still +requires the exact durable receipt and current node authority. + Node hosts must support the current private worker-supervisor dialect before they can host sessions. An older connected host remains visible but disabled in the session picker. Update OpenClaw on that device and reconnect it; for a diff --git a/docs/plan/runners.md b/docs/plan/runners.md index 787ac73e6bf6..3545ea795a38 100644 --- a/docs/plan/runners.md +++ b/docs/plan/runners.md @@ -233,9 +233,10 @@ bounded terminal receipts, and the Gateway launch replay/poll/cancel adapter. A node publishes one atomic, reconnect-scoped private runner inventory with the supervisor dialect, explicit local consent, and current capacity. Milestone 7 removes the temporary local-package scanner and connect-time build claim; the -durable Gateway bundle receipt is the sole execution authority. Public node and environment projections expose -only `sessionHost`; a read-scoped topology invalidation makes clients refetch -without exposing build identity. Status and cancellation reacquire the current +durable Gateway bundle receipt is the sole execution authority. Public node and +environment projections expose `sessionHost` plus a redacted installed/missing +bundle status; a read-scoped topology invalidation makes clients refetch without +exposing hashes, paths, or receipt details. Status and cancellation reacquire the current supervisor proof and use the durable launch identity so an upgrade cannot strand an existing worker. Node-local opt-in advertises capacity; default nodes remain non-hosts. The supervisor owns two atomic durable capacity slots, bounded @@ -327,7 +328,9 @@ local-package execution. The cleanup that follows separates inventory consent an capacity from installed bundle state and deletes the obsolete local build scan. The retention slice (#124590) reuses the authoritative maintenance snapshot to prune superseded node bundles in bounded generation-acknowledged passes. The -remaining functional slice exposes the validated installed version on Devices. +installed-status slice validates one retained hash on the node, keeps that fact +reconnect-scoped and proof-bound in the Gateway, and shows the Gateway-owned +version quietly on Devices or a remediation warning when the bundle is missing. ### Projects read model (milestone 4 foundation) diff --git a/packages/gateway-protocol/src/schema/environments.test.ts b/packages/gateway-protocol/src/schema/environments.test.ts index 7583a4391376..5f83544028d7 100644 --- a/packages/gateway-protocol/src/schema/environments.test.ts +++ b/packages/gateway-protocol/src/schema/environments.test.ts @@ -114,6 +114,42 @@ describe("worker environment protocol schemas", () => { ).toBe(true); }); + it("accepts only redacted node worker bundle status", () => { + const node = { + id: "node:build-mac", + type: "node", + status: "available", + }; + expect( + Value.Check(EnvironmentSummarySchema, { + ...node, + workerBundle: { status: "installed", version: "2026.8.9" }, + }), + ).toBe(true); + expect( + Value.Check(EnvironmentSummarySchema, { + ...node, + workerBundle: { status: "missing" }, + }), + ).toBe(true); + expect( + Value.Check(EnvironmentSummarySchema, { + ...node, + workerBundle: { + status: "installed", + version: "2026.8.9", + bundleHash: "a".repeat(64), + }, + }), + ).toBe(false); + expect( + Value.Check(EnvironmentSummarySchema, { + ...node, + workerBundle: { status: "installed", version: "" }, + }), + ).toBe(false); + }); + it("accepts bounded node lifecycle history and rejects malformed timestamps", () => { const node = { id: "node:build-mac", diff --git a/packages/gateway-protocol/src/schema/environments.ts b/packages/gateway-protocol/src/schema/environments.ts index d94fd3ae320e..0a28e9418abd 100644 --- a/packages/gateway-protocol/src/schema/environments.ts +++ b/packages/gateway-protocol/src/schema/environments.ts @@ -55,6 +55,11 @@ export const RuntimeTargetIssueSchema = closedObject({ headlessReconnectCommand: Type.Literal("openclaw node restart"), }); +const NodeWorkerBundleStatusSchema = Type.Union([ + closedObject({ status: Type.Literal("installed"), version: NonEmptyString }), + closedObject({ status: Type.Literal("missing") }), +]); + /** Worker-only lifecycle metadata layered onto the existing environment projection. */ export const WorkerEnvironmentMetadataSchema = closedObject({ providerId: NonEmptyString, @@ -79,6 +84,7 @@ function createEnvironmentSummarySchema() { status: EnvironmentStatusSchema, platform: Type.Optional(NonEmptyString), sessionHost: Type.Optional(Type.Boolean()), + workerBundle: Type.Optional(NodeWorkerBundleStatusSchema), lastConnectedAtMs: Type.Optional(Type.Integer({ minimum: 0 })), lastDisconnectedAtMs: Type.Optional(Type.Integer({ minimum: 0 })), lastSeenAtMs: Type.Optional(Type.Integer({ minimum: 0 })), diff --git a/packages/gateway-protocol/src/schema/frames.ts b/packages/gateway-protocol/src/schema/frames.ts index ea63dd80e7d6..b48c40cc4f4e 100644 --- a/packages/gateway-protocol/src/schema/frames.ts +++ b/packages/gateway-protocol/src/schema/frames.ts @@ -12,6 +12,7 @@ export const GATEWAY_SERVER_CAPS = { CHAT_SEND_ROUTING_CONTRACT: "chat-send-routing-contract", GATEWAY_RESTART_TARGET_SAFE: "gateway-restart-target-safe-v1", NODE_WORKER_BUNDLE_RETENTION: "node-worker-bundle-retention-v1", + NODE_WORKER_BUNDLE_STATUS: "node-worker-bundle-status-v1", SYSTEM_AGENT_WIZARD_CANCEL: "openclaw-chat-wizard-cancel", SYSTEM_AGENT_SETUP_MODEL_REF: "openclaw-setup-model-ref", TASK_SUGGESTIONS_ACCEPT_MODES: "taskSuggestions.acceptModes", diff --git a/packages/sdk/src/index.ts b/packages/sdk/src/index.ts index 4d4196b347ef..5932e141e354 100644 --- a/packages/sdk/src/index.ts +++ b/packages/sdk/src/index.ts @@ -36,6 +36,7 @@ export type { EnvironmentSelection, EnvironmentSummary, EnvironmentsListResult, + NodeWorkerBundleStatus, GatewayEvent, GatewayRequestOptions, JsonObject, diff --git a/packages/sdk/src/types.ts b/packages/sdk/src/types.ts index dc4e6508dce0..02e6924c1eef 100644 --- a/packages/sdk/src/types.ts +++ b/packages/sdk/src/types.ts @@ -74,6 +74,10 @@ export type WorkerEnvironmentMetadata = { error?: string; }; +export type NodeWorkerBundleStatus = + | { status: "installed"; version: string } + | { status: "missing" }; + export type EnvironmentSummary = { id: string; type: "local" | "gateway" | "node" | "managed" | "ephemeral" | (string & {}); @@ -81,6 +85,7 @@ export type EnvironmentSummary = { status: "available" | "unavailable" | "starting" | "stopping" | "error"; platform?: string; sessionHost?: boolean; + workerBundle?: NodeWorkerBundleStatus; lastConnectedAtMs?: number; lastDisconnectedAtMs?: number; lastSeenAtMs?: number; diff --git a/src/gateway/node-catalog.ts b/src/gateway/node-catalog.ts index 3d252e7d972f..a019296f9610 100644 --- a/src/gateway/node-catalog.ts +++ b/src/gateway/node-catalog.ts @@ -226,10 +226,19 @@ function buildEffectiveKnownNode(entry: { pendingNodePairing?: KnownNodePendingSource; live?: NodeSession; sessionHost: boolean; + workerBundle?: NodeListNode["workerBundle"]; issues?: NodeListNode["issues"]; }): NodeListNode { - const { nodeId, devicePairing, nodePairing, pendingNodePairing, live, sessionHost, issues } = - entry; + const { + nodeId, + devicePairing, + nodePairing, + pendingNodePairing, + live, + sessionHost, + workerBundle, + issues, + } = entry; const lastSeen = resolveEffectiveLastSeen({ live, devicePairing, nodePairing }); const lastConnectedAtMs = maxDefinedTimestamp( nodePairing?.lastConnectedAtMs, @@ -299,6 +308,7 @@ function buildEffectiveKnownNode(entry: { ), computerUse: live?.computerUse, sessionHost, + ...(live && workerBundle ? { workerBundle: structuredClone(workerBundle) } : {}), ...(issues?.length ? { issues: [...issues] } : {}), nodePluginTools: live?.nodePluginTools, pathEnv: live?.pathEnv, @@ -349,6 +359,7 @@ export function createKnownNodeCatalog(params: { pendingNodes?: readonly NodePairingPendingRequest[]; connectedNodes: readonly NodeSession[]; sessionHostNodeIds?: ReadonlySet; + workerBundleByNodeId?: ReadonlyMap>; issuesByNodeId?: ReadonlyMap; }): KnownNodeCatalog { const devicePairingById = new Map( @@ -403,6 +414,7 @@ export function createKnownNodeCatalog(params: { pendingNodePairing, live, sessionHost: params.sessionHostNodeIds?.has(nodeId) === true, + workerBundle: params.workerBundleByNodeId?.get(nodeId), issues: params.issuesByNodeId?.get(nodeId), }), }); diff --git a/src/gateway/node-registry-private.ts b/src/gateway/node-registry-private.ts index 73dd728b46bd..3272ee97bb7d 100644 --- a/src/gateway/node-registry-private.ts +++ b/src/gateway/node-registry-private.ts @@ -8,14 +8,12 @@ import { NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND, } from "../infra/node-commands.js"; import { - NODE_RUNNER_UPDATE_REQUIRED_ISSUE, - NODE_WORKER_SUPERVISOR_BUILD_PROTOCOL_FEATURE, - NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE, - NODE_WORKER_SUPERVISOR_LEGACY_PROTOCOL_FEATURE, + NODE_WORKER_BUNDLE_RETENTION_VERSION, + NODE_WORKER_BUNDLE_STATUS_VERSION, type NodeRunnerInventoryIssue, type NodeRunnerInventoryDeclaration, - type NodeWorkerHostDeclaration, } from "../infra/node-runner-inventory.js"; +import type { NodeWorkerBundleStatus } from "../shared/node-list-types.js"; import { sameWorkerProtocolFeatures } from "../worker/worker-build-identity.js"; import { NODE_INVOKE_PAIRING_CHANGED_ABORT } from "./node-registry-private-token.js"; import type { @@ -24,17 +22,18 @@ import type { PendingSystemRunEvent, } from "./node-registry.invoke-stream.js"; import { normalizeSystemRunTimeoutMs } from "./node-registry.system-run.js"; +import { + resolveNodeRunnerInventoryIssue, + resolveNodeWorkerSupervisorProof, + sameNodeWorkerHostDeclaration, + type NodeRunnerInventoryRecord, + type NodeRunnerRegistrySession, + type NodeWorkerSupervisorNodeProof, +} from "./node-runner-inventory-runtime.js"; -type NodeRegistryPrivateSession = { - nodeId: string; - connId: string; - pairingIdentity?: string; - pairingGeneration?: string; - client: { invalidated?: boolean }; - clientId?: string; - clientMode?: string; - commands: string[]; -}; +export type { NodeWorkerSupervisorNodeProof } from "./node-runner-inventory-runtime.js"; + +type NodeRegistryPrivateSession = NodeRunnerRegistrySession; type NodeInvokeResult = { ok: boolean; @@ -67,21 +66,19 @@ type NodeInvokeParams = { type NodeWorkerPrivateCommand = (typeof NODE_WORKER_PRIVATE_COMMANDS)[number]; -export type NodeWorkerSupervisorNodeProof = { - nodeId: string; - connId: string; - pairingIdentity: string; - pairingGeneration: string; - clientId: typeof GATEWAY_CLIENT_IDS.NODE_HOST; - clientMode: "node"; - protocolFeature: typeof NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE; - workerHost: Extract; - commands: readonly string[]; +type NodeWorkerBundleStatusObservation = { + bundleHash: string; + status: NodeWorkerBundleStatus; }; export type NodeWorkerSupervisorTransport = { listCurrentNodes(): Promise; getIssue?(nodeId: string): NodeRunnerInventoryIssue | undefined; + getBundleStatus?(nodeId: string): NodeWorkerBundleStatusObservation | undefined; + acceptBundleStatus?( + node: NodeWorkerSupervisorNodeProof, + observation: NodeWorkerBundleStatusObservation | undefined, + ): boolean; isCurrent(node: NodeWorkerSupervisorNodeProof, requireLaunchEligibility?: boolean): boolean; invoke(params: { node: NodeWorkerSupervisorNodeProof; @@ -95,14 +92,6 @@ export type NodeWorkerSupervisorTransport = { }): Promise; }; -type NodeRunnerInventoryRecord = Omit< - NodeWorkerSupervisorNodeProof, - "commands" | "pairingGeneration" | "protocolFeature" | "workerHost" -> & { - protocolFeatures: readonly string[]; - workerHost?: NodeWorkerHostDeclaration; -}; - type NodeRegistryPrivateContext = { getNode: (nodeId: string) => PairingBoundNodeSession | undefined; listCurrentConnected: () => Promise; @@ -137,6 +126,7 @@ type NodeRunnerInventoryUpdateResult = { type NodeRegistryPrivateState = { context: NodeRegistryPrivateContext; runnerInventoryByConn: Map; + bundleStatusByConn: Map; generationBoundInvokes: WeakMap; publishRunnerInventoryChanged: (nodeId: string) => void; invokeCore: (params: NodeInvokeParams, allowPrivateCommand: boolean) => Promise; @@ -194,71 +184,18 @@ function normalizeSystemRunInvokeParams(params: { command: string; params?: unkn return normalized; } -function sameWorkerHostDeclaration( - left: NodeWorkerHostDeclaration | undefined, - right: NodeWorkerHostDeclaration | undefined, +function sameBundleStatusObservation( + left: NodeWorkerBundleStatusObservation | undefined, + right: NodeWorkerBundleStatusObservation | undefined, ): boolean { return ( - left?.enabled === right?.enabled && - (left?.enabled !== true || - (right?.enabled === true && - left.capacity === right.capacity && - left.bundlePrewarm === right.bundlePrewarm && - left.bundleRetention === right.bundleRetention)) + left?.bundleHash === right?.bundleHash && + left?.status.status === right?.status.status && + (left?.status.status !== "installed" || + (right?.status.status === "installed" && left.status.version === right.status.version)) ); } -function resolveWorkerSupervisorProof( - node: NodeRegistryPrivateSession, - runnerInventoryByConn: ReadonlyMap, -): NodeWorkerSupervisorNodeProof | undefined { - const declaration = runnerInventoryByConn.get(node.connId); - if ( - !declaration || - !node.pairingIdentity || - !node.pairingGeneration || - node.clientId !== GATEWAY_CLIENT_IDS.NODE_HOST || - node.clientMode !== "node" || - declaration.nodeId !== node.nodeId || - declaration.pairingIdentity !== node.pairingIdentity || - declaration.clientId !== node.clientId || - declaration.clientMode !== node.clientMode || - !declaration.protocolFeatures.includes(NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE) || - declaration.workerHost?.enabled !== true - ) { - return undefined; - } - return { - nodeId: node.nodeId, - connId: node.connId, - pairingIdentity: node.pairingIdentity, - pairingGeneration: node.pairingGeneration, - clientId: GATEWAY_CLIENT_IDS.NODE_HOST, - clientMode: "node", - protocolFeature: NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE, - workerHost: { ...declaration.workerHost }, - commands: [...node.commands], - }; -} - -function resolveNodeRunnerIssue( - node: NodeRegistryPrivateSession, - runnerInventoryByConn: ReadonlyMap, -): NodeRunnerInventoryIssue | undefined { - const declaration = runnerInventoryByConn.get(node.connId); - return declaration && - node.client.invalidated !== true && - declaration.nodeId === node.nodeId && - declaration.pairingIdentity === node.pairingIdentity && - declaration.clientId === GATEWAY_CLIENT_IDS.NODE_HOST && - declaration.clientMode === "node" && - declaration.protocolFeatures.length === 1 && - (declaration.protocolFeatures[0] === NODE_WORKER_SUPERVISOR_LEGACY_PROTOCOL_FEATURE || - declaration.protocolFeatures[0] === NODE_WORKER_SUPERVISOR_BUILD_PROTOCOL_FEATURE) - ? NODE_RUNNER_UPDATE_REQUIRED_ISSUE - : undefined; -} - function isWorkerSupervisorProofCurrent( state: NodeRegistryPrivateState, proof: NodeWorkerSupervisorNodeProof, @@ -268,7 +205,7 @@ function isWorkerSupervisorProofCurrent( if (!node || node.client.invalidated === true || node.connId !== proof.connId) { return false; } - const current = resolveWorkerSupervisorProof(node, state.runnerInventoryByConn); + const current = resolveNodeWorkerSupervisorProof(node, state.runnerInventoryByConn); return ( current?.pairingIdentity === proof.pairingIdentity && current.pairingGeneration === proof.pairingGeneration && @@ -300,7 +237,9 @@ function updateWorkerRunnerInventory( } const previous = state.runnerInventoryByConn.get(node.connId); if (!publishesRunnerDialect) { - const changed = state.runnerInventoryByConn.delete(node.connId); + const inventoryChanged = state.runnerInventoryByConn.delete(node.connId); + const statusChanged = state.bundleStatusByConn.delete(node.connId); + const changed = inventoryChanged || statusChanged; if (changed) { state.context.publishActiveNodeContext(); state.publishRunnerInventoryChanged(node.nodeId); @@ -317,10 +256,17 @@ function updateWorkerRunnerInventory( protocolFeatures: [...params.declaration.protocolFeatures], ...(workerHost ? { workerHost: { ...workerHost } } : {}), }; + const statusCleared = + next.workerHost?.enabled !== true || + next.workerHost.bundleRetention === undefined || + next.workerHost.bundleStatus === undefined + ? state.bundleStatusByConn.delete(node.connId) + : false; const changed = !previous || !sameWorkerProtocolFeatures(previous.protocolFeatures, next.protocolFeatures) || - !sameWorkerHostDeclaration(previous.workerHost, next.workerHost); + !sameNodeWorkerHostDeclaration(previous.workerHost, next.workerHost) || + statusCleared; if (changed) { state.runnerInventoryByConn.set(node.connId, next); state.context.publishActiveNodeContext(); @@ -493,6 +439,7 @@ export function registerNodeRegistryPrivateRuntime( const state = {} as NodeRegistryPrivateState; state.context = context; state.runnerInventoryByConn = new Map(); + state.bundleStatusByConn = new Map(); state.generationBoundInvokes = new WeakMap(); state.publishRunnerInventoryChanged = () => {}; state.invokeCore = async (params, allowPrivateCommand) => @@ -502,13 +449,43 @@ export function registerNodeRegistryPrivateRuntime( listCurrentNodes: async () => { const current = await context.listCurrentConnected(); return current.flatMap((node) => { - const proof = resolveWorkerSupervisorProof(node, state.runnerInventoryByConn); + const proof = resolveNodeWorkerSupervisorProof(node, state.runnerInventoryByConn); return proof ? [proof] : []; }); }, getIssue: (nodeId) => { const node = context.getNode(nodeId); - return node ? resolveNodeRunnerIssue(node, state.runnerInventoryByConn) : undefined; + return node ? resolveNodeRunnerInventoryIssue(node, state.runnerInventoryByConn) : undefined; + }, + getBundleStatus: (nodeId) => { + const node = context.getNode(nodeId); + const observation = node ? state.bundleStatusByConn.get(node.connId) : undefined; + return observation ? structuredClone(observation) : undefined; + }, + acceptBundleStatus: (node, observation) => { + if (!isWorkerSupervisorProofCurrent(state, node, false)) { + return false; + } + const currentNode = state.context.getNode(node.nodeId); + const currentProof = currentNode + ? resolveNodeWorkerSupervisorProof(currentNode, state.runnerInventoryByConn) + : undefined; + if ( + currentProof?.workerHost.bundleRetention !== NODE_WORKER_BUNDLE_RETENTION_VERSION || + currentProof.workerHost.bundleStatus !== NODE_WORKER_BUNDLE_STATUS_VERSION + ) { + return false; + } + const previous = state.bundleStatusByConn.get(node.connId); + if (observation) { + state.bundleStatusByConn.set(node.connId, structuredClone(observation)); + } else { + state.bundleStatusByConn.delete(node.connId); + } + if (!sameBundleStatusObservation(previous, observation)) { + state.publishRunnerInventoryChanged(node.nodeId); + } + return true; }, isCurrent: (node, requireLaunchEligibility = false) => isWorkerSupervisorProofCurrent(state, node, requireLaunchEligibility), @@ -615,6 +592,7 @@ export function forgetNodeRunnerInventory(nodeRegistry: object, connId: string): if (!state || !declaration || !state.runnerInventoryByConn.delete(connId)) { return; } + state.bundleStatusByConn.delete(connId); state.publishRunnerInventoryChanged(declaration.nodeId); } @@ -629,7 +607,7 @@ export function isNodeRunnerSessionHost(params: { if (!state || !node || node.connId !== params.connId) { return false; } - const proof = resolveWorkerSupervisorProof(node, state.runnerInventoryByConn); + const proof = resolveNodeWorkerSupervisorProof(node, state.runnerInventoryByConn); return Boolean( proof && proof.pairingGeneration === params.pairingGeneration && @@ -645,10 +623,25 @@ function getNodeRunnerInventoryIssue(params: { const state = NODE_REGISTRY_PRIVATE_STATES.get(params.registry); const node = state?.context.getNode(params.nodeId); return state && node?.connId === params.connId - ? resolveNodeRunnerIssue(node, state.runnerInventoryByConn) + ? resolveNodeRunnerInventoryIssue(node, state.runnerInventoryByConn) : undefined; } +export function collectNodeWorkerBundleStatusByNodeId( + registry: object, + connectedNodes: ReadonlyArray<{ nodeId: string; connId: string }>, +): Map { + const state = NODE_REGISTRY_PRIVATE_STATES.get(registry); + return new Map( + connectedNodes.flatMap((node) => { + const current = state?.context.getNode(node.nodeId); + const observation = + current?.connId === node.connId ? state?.bundleStatusByConn.get(node.connId) : undefined; + return observation ? [[node.nodeId, structuredClone(observation.status)] as const] : []; + }), + ); +} + /** Shared node/environments read-projection shape: nodeId -> runner issues. */ export function collectNodeRunnerIssuesByNodeId( registry: object, @@ -689,6 +682,9 @@ export function settleNodeRegistryPairingGenerationChange(params: { if (!state) { return; } + if (state.bundleStatusByConn.delete(params.connId)) { + state.publishRunnerInventoryChanged(params.nodeId); + } for (const pending of state.context.pendingInvokes.values()) { const binding = state.generationBoundInvokes.get(pending); if ( diff --git a/src/gateway/node-runner-inventory-runtime.ts b/src/gateway/node-runner-inventory-runtime.ts new file mode 100644 index 000000000000..fc4bc6da92bd --- /dev/null +++ b/src/gateway/node-runner-inventory-runtime.ts @@ -0,0 +1,106 @@ +import { GATEWAY_CLIENT_IDS } from "../../packages/gateway-protocol/src/client-info.js"; +import { + NODE_RUNNER_UPDATE_REQUIRED_ISSUE, + NODE_WORKER_SUPERVISOR_BUILD_PROTOCOL_FEATURE, + NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE, + NODE_WORKER_SUPERVISOR_LEGACY_PROTOCOL_FEATURE, + type NodeRunnerInventoryIssue, + type NodeWorkerHostDeclaration, +} from "../infra/node-runner-inventory.js"; + +export type NodeRunnerRegistrySession = { + nodeId: string; + connId: string; + pairingIdentity?: string; + pairingGeneration?: string; + client: { invalidated?: boolean }; + clientId?: string; + clientMode?: string; + commands: string[]; +}; + +export type NodeWorkerSupervisorNodeProof = { + nodeId: string; + connId: string; + pairingIdentity: string; + pairingGeneration: string; + clientId: typeof GATEWAY_CLIENT_IDS.NODE_HOST; + clientMode: "node"; + protocolFeature: typeof NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE; + workerHost: Extract; + commands: readonly string[]; +}; + +export type NodeRunnerInventoryRecord = Omit< + NodeWorkerSupervisorNodeProof, + "commands" | "pairingGeneration" | "protocolFeature" | "workerHost" +> & { + protocolFeatures: readonly string[]; + workerHost?: NodeWorkerHostDeclaration; +}; + +export function sameNodeWorkerHostDeclaration( + left: NodeWorkerHostDeclaration | undefined, + right: NodeWorkerHostDeclaration | undefined, +): boolean { + return ( + left?.enabled === right?.enabled && + (left?.enabled !== true || + (right?.enabled === true && + left.capacity === right.capacity && + left.bundlePrewarm === right.bundlePrewarm && + left.bundleRetention === right.bundleRetention && + left.bundleStatus === right.bundleStatus)) + ); +} + +export function resolveNodeWorkerSupervisorProof( + node: NodeRunnerRegistrySession, + runnerInventoryByConn: ReadonlyMap, +): NodeWorkerSupervisorNodeProof | undefined { + const declaration = runnerInventoryByConn.get(node.connId); + if ( + !declaration || + !node.pairingIdentity || + !node.pairingGeneration || + node.clientId !== GATEWAY_CLIENT_IDS.NODE_HOST || + node.clientMode !== "node" || + declaration.nodeId !== node.nodeId || + declaration.pairingIdentity !== node.pairingIdentity || + declaration.clientId !== node.clientId || + declaration.clientMode !== node.clientMode || + !declaration.protocolFeatures.includes(NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE) || + declaration.workerHost?.enabled !== true + ) { + return undefined; + } + return { + nodeId: node.nodeId, + connId: node.connId, + pairingIdentity: node.pairingIdentity, + pairingGeneration: node.pairingGeneration, + clientId: GATEWAY_CLIENT_IDS.NODE_HOST, + clientMode: "node", + protocolFeature: NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE, + workerHost: { ...declaration.workerHost }, + commands: [...node.commands], + }; +} + +export function resolveNodeRunnerInventoryIssue( + node: NodeRunnerRegistrySession, + runnerInventoryByConn: ReadonlyMap, +): NodeRunnerInventoryIssue | undefined { + const declaration = runnerInventoryByConn.get(node.connId); + return declaration && + node.client.invalidated !== true && + declaration.nodeId === node.nodeId && + declaration.pairingIdentity === node.pairingIdentity && + declaration.clientId === GATEWAY_CLIENT_IDS.NODE_HOST && + declaration.clientMode === "node" && + declaration.protocolFeatures.length === 1 && + (declaration.protocolFeatures[0] === NODE_WORKER_SUPERVISOR_LEGACY_PROTOCOL_FEATURE || + declaration.protocolFeatures[0] === NODE_WORKER_SUPERVISOR_BUILD_PROTOCOL_FEATURE) + ? NODE_RUNNER_UPDATE_REQUIRED_ISSUE + : undefined; +} diff --git a/src/gateway/server-methods/environments.test.ts b/src/gateway/server-methods/environments.test.ts index 51658bf62a6d..3312081fd853 100644 --- a/src/gateway/server-methods/environments.test.ts +++ b/src/gateway/server-methods/environments.test.ts @@ -9,6 +9,7 @@ import { NODE_RUNNER_UPDATE_REQUIRED_ISSUE } from "../../infra/node-runner-inven import { NODE_DESKTOP_STREAM_COMMAND } from "../../shared/node-desktop-stream.js"; import { collectNodeRunnerIssuesByNodeId, + collectNodeWorkerBundleStatusByNodeId, isNodeRunnerSessionHost, } from "../node-registry-private.js"; import type { WorkerEnvironmentServiceRecord } from "../worker-environments/service-contract.js"; @@ -26,6 +27,7 @@ vi.mock("../../infra/device-pairing-node.js", () => ({ vi.mock("../node-registry-private.js", () => ({ collectNodeRunnerIssuesByNodeId: vi.fn(() => new Map()), + collectNodeWorkerBundleStatusByNodeId: vi.fn(() => new Map()), isNodeRunnerSessionHost: vi.fn(() => false), })); @@ -202,6 +204,7 @@ beforeEach(() => { vi.spyOn(Date, "now").mockReturnValue(NOW); vi.mocked(isNodeRunnerSessionHost).mockReturnValue(false); vi.mocked(collectNodeRunnerIssuesByNodeId).mockReturnValue(new Map()); + vi.mocked(collectNodeWorkerBundleStatusByNodeId).mockReturnValue(new Map()); vi.mocked(listDevicePairing).mockResolvedValue({ paired: [] } as never); vi.mocked(listNodePairing).mockResolvedValue({ paired: [ @@ -308,6 +311,26 @@ describe("environment gateway methods", () => { }); }); + it("projects the same redacted worker bundle status through list and status", async () => { + vi.mocked(collectNodeWorkerBundleStatusByNodeId).mockReturnValue( + new Map([["node-live", { status: "installed", version: "2026.8.9" }]]), + ); + + const [, listPayload] = await callEnvironmentMethod("environments.list", {}); + const [, statusPayload] = await callEnvironmentMethod("environments.status", { + environmentId: "node:node-live", + }); + const listed = ( + listPayload as { environments: Array<{ id: string; workerBundle?: unknown }> } + ).environments.find((environment) => environment.id === "node:node-live"); + + expect(listed?.workerBundle).toEqual({ status: "installed", version: "2026.8.9" }); + expect(statusPayload).toMatchObject({ + workerBundle: { status: "installed", version: "2026.8.9" }, + }); + expect(JSON.stringify({ listed, statusPayload })).not.toContain("bundleHash"); + }); + it("projects the same current-node update issue through list and status", async () => { vi.mocked(collectNodeRunnerIssuesByNodeId).mockReturnValue( new Map([["node-live", [NODE_RUNNER_UPDATE_REQUIRED_ISSUE]]]), diff --git a/src/gateway/server-methods/environments.ts b/src/gateway/server-methods/environments.ts index bd6981376ae0..87e95e8f15bd 100644 --- a/src/gateway/server-methods/environments.ts +++ b/src/gateway/server-methods/environments.ts @@ -23,6 +23,7 @@ import { createKnownNodeCatalog, listKnownNodes } from "../node-catalog.js"; import { isNodeCommandAllowed, resolveNodeCommandAllowlist } from "../node-command-policy.js"; import { collectNodeRunnerIssuesByNodeId, + collectNodeWorkerBundleStatusByNodeId, isNodeRunnerSessionHost, } from "../node-registry-private.js"; import type { WorkerEnvironmentServiceRecord } from "../worker-environments/service-contract.js"; @@ -91,6 +92,7 @@ function summarizeNodeEnvironment( status: node.connected ? "available" : "unavailable", ...(platform ? { platform } : {}), sessionHost: node.connected === true && node.sessionHost === true, + ...(node.workerBundle ? { workerBundle: structuredClone(node.workerBundle) } : {}), ...(node.lastConnectedAtMs !== undefined ? { lastConnectedAtMs: node.lastConnectedAtMs } : {}), ...(node.lastDisconnectedAtMs !== undefined ? { lastDisconnectedAtMs: node.lastDisconnectedAtMs } @@ -160,11 +162,16 @@ async function listEnvironments(context: GatewayRequestContext): Promise { diff --git a/src/gateway/server-methods/nodes.runner-inventory.test.ts b/src/gateway/server-methods/nodes.runner-inventory.test.ts index 0688c90d0673..0de923672042 100644 --- a/src/gateway/server-methods/nodes.runner-inventory.test.ts +++ b/src/gateway/server-methods/nodes.runner-inventory.test.ts @@ -9,6 +9,7 @@ import { NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE, } from "../../infra/node-runner-inventory.js"; import { + collectNodeWorkerBundleStatusByNodeId, createNodeRegistryRuntime, setNodeRunnerInventoryChangedListener, } from "../node-registry-private.js"; @@ -59,6 +60,17 @@ const fullHost = { workerHost: { enabled: true, capacity: "full", bundlePrewarm: 1 }, } as const; +const retainedHost = { + protocolFeatures: [NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE], + workerHost: { + enabled: true, + capacity: "available", + bundlePrewarm: 1, + bundleRetention: 1, + bundleStatus: 1, + }, +} as const; + describe("nodeHandlers node.runnerInventory.update", () => { it("publishes explicit runner consent and launch capacity for the authenticated node", async () => { const inventoryChanged = vi.fn(); @@ -90,6 +102,102 @@ describe("nodeHandlers node.runnerInventory.update", () => { runtime.nodeRegistry.unregister("conn-1"); }); + it("stores bundle status only for the exact current node proof", async () => { + const runtime = createNodeRegistryRuntime(() => new NodeRegistry()); + const client = createWorkerSupervisorNodeClient(); + runtime.nodeRegistry.register(client, { + pairingIdentity: "identity-1", + pairingGeneration: "generation-1", + }); + await runnerInventoryHandler( + runnerInventoryOptions({ + nodeRegistry: runtime.nodeRegistry, + client, + declaration: retainedHost, + }), + ); + const [proof] = await runtime.nodeWorkerSupervisorTransport.listCurrentNodes(); + if (!proof) { + throw new Error("expected current node proof"); + } + + expect( + runtime.nodeWorkerSupervisorTransport.acceptBundleStatus?.(proof, { + bundleHash: "a".repeat(64), + status: { status: "installed", version: "2026.8.9" }, + }), + ).toBe(true); + expect(runtime.nodeWorkerSupervisorTransport.getBundleStatus?.("node-1")).toEqual({ + bundleHash: "a".repeat(64), + status: { status: "installed", version: "2026.8.9" }, + }); + expect( + collectNodeWorkerBundleStatusByNodeId(runtime.nodeRegistry, [ + { nodeId: "node-1", connId: "conn-1" }, + ]), + ).toEqual(new Map([["node-1", { status: "installed", version: "2026.8.9" }]])); + + expect( + runtime.nodeRegistry.updateSurface( + "node-1", + { commands: ["system.run"] }, + { + expectedConnId: "conn-1", + expectedPairingIdentity: "identity-1", + expectedPairingGeneration: "generation-1", + nextPairingGeneration: "generation-2", + }, + ), + ).not.toBeNull(); + expect( + runtime.nodeWorkerSupervisorTransport.acceptBundleStatus?.(proof, { + bundleHash: "b".repeat(64), + status: { status: "missing" }, + }), + ).toBe(false); + expect( + collectNodeWorkerBundleStatusByNodeId(runtime.nodeRegistry, [ + { nodeId: "node-1", connId: "conn-1" }, + ]), + ).toEqual(new Map()); + + const [currentProof] = await runtime.nodeWorkerSupervisorTransport.listCurrentNodes(); + if (!currentProof) { + throw new Error("expected promoted node proof"); + } + expect( + runtime.nodeWorkerSupervisorTransport.acceptBundleStatus?.(currentProof, { + bundleHash: "b".repeat(64), + status: { status: "missing" }, + }), + ).toBe(true); + await runnerInventoryHandler( + runnerInventoryOptions({ + nodeRegistry: runtime.nodeRegistry, + client, + declaration: availableHost, + }), + ); + expect( + runtime.nodeWorkerSupervisorTransport.acceptBundleStatus?.(currentProof, { + bundleHash: "c".repeat(64), + status: { status: "installed", version: "2026.8.9" }, + }), + ).toBe(false); + expect( + collectNodeWorkerBundleStatusByNodeId(runtime.nodeRegistry, [ + { nodeId: "node-1", connId: "conn-1" }, + ]), + ).toEqual(new Map()); + + runtime.nodeRegistry.unregister("conn-1"); + expect( + collectNodeWorkerBundleStatusByNodeId(runtime.nodeRegistry, [ + { nodeId: "node-1", connId: "conn-1" }, + ]), + ).toEqual(new Map()); + }); + it("retains the supervisor proof while full but rejects new launches", async () => { const runtime = createNodeRegistryRuntime(() => new NodeRegistry()); const client = createWorkerSupervisorNodeClient(); @@ -345,6 +453,20 @@ describe("nodeHandlers node.runnerInventory.update", () => { workerHost: { enabled: true, capacity: "available", bundleRetention: 2 }, }, }, + { + name: "unsupported bundle status version", + params: { + protocolFeatures: [NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE], + workerHost: { enabled: true, capacity: "available", bundleStatus: 2 }, + }, + }, + { + name: "bundle status without bundle retention", + params: { + protocolFeatures: [NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE], + workerHost: { enabled: true, capacity: "available", bundleStatus: 1 }, + }, + }, ])("rejects $name without changing private eligibility", async ({ params }) => { const runtime = createNodeRegistryRuntime(() => new NodeRegistry()); const client = createWorkerSupervisorNodeClient(); diff --git a/src/gateway/server-methods/nodes.test.ts b/src/gateway/server-methods/nodes.test.ts index 7bcab6a2aa92..1ab453bbd5e0 100644 --- a/src/gateway/server-methods/nodes.test.ts +++ b/src/gateway/server-methods/nodes.test.ts @@ -157,18 +157,20 @@ function createOptions( ): { context: ReturnType; opts: GatewayRequestHandlerOptions; + respond: ReturnType; } { const context = createContext(); + const respond = vi.fn(); const opts = { req: { type: "req", id: "req-1", method: "node.pair.remove", params }, params, client: createClient(["operator.pairing", "operator.admin"]), isWebchatConnect: () => false, - respond: vi.fn(), + respond, context, ...overrides, } as unknown as GatewayRequestHandlerOptions; - return { context, opts }; + return { context, opts, respond }; } describe("nodeHandlers node.skills.update", () => { @@ -293,7 +295,12 @@ describe("nodeHandlers node.describe", () => { const publication = createOptions( { protocolFeatures: [NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE], - workerHost: { enabled: true, capacity: "available" }, + workerHost: { + enabled: true, + capacity: "available", + bundleRetention: 1, + bundleStatus: 1, + }, }, { client: nodeClient as never }, ); @@ -302,7 +309,22 @@ describe("nodeHandlers node.describe", () => { nodeHandlers["node.runnerInventory.update"], 'nodeHandlers["node.runnerInventory.update"] test invariant', )(publication.opts); + const [proof] = await runtime.nodeWorkerSupervisorTransport.listCurrentNodes(); + expect(proof).toBeDefined(); + expect( + proof && + runtime.nodeWorkerSupervisorTransport.acceptBundleStatus?.(proof, { + bundleHash: "a".repeat(64), + status: { status: "installed", version: "2026.8.9" }, + }), + ).toBe(true); + const listCall = createOptions({}); + Object.assign(listCall.context, { nodeRegistry: runtime.nodeRegistry }); + await expectDefined( + nodeHandlers["node.list"], + 'nodeHandlers["node.list"] test invariant', + )(listCall.opts); const describeCall = createOptions({ nodeId }); Object.assign(describeCall.context, { nodeRegistry: runtime.nodeRegistry }); await expectDefined( @@ -310,11 +332,29 @@ describe("nodeHandlers node.describe", () => { 'nodeHandlers["node.describe"] test invariant', )(describeCall.opts); - expect(describeCall.opts.respond).toHaveBeenCalledWith( + expect(listCall.respond).toHaveBeenCalledWith( true, - expect.objectContaining({ nodeId, sessionHost: true }), + expect.objectContaining({ + nodes: expect.arrayContaining([ + expect.objectContaining({ + nodeId, + workerBundle: { status: "installed", version: "2026.8.9" }, + }), + ]), + }), undefined, ); + expect(describeCall.respond).toHaveBeenCalledWith( + true, + expect.objectContaining({ + nodeId, + sessionHost: true, + workerBundle: { status: "installed", version: "2026.8.9" }, + }), + undefined, + ); + expect(JSON.stringify(listCall.respond.mock.calls)).not.toContain("bundleHash"); + expect(JSON.stringify(describeCall.respond.mock.calls)).not.toContain("bundleHash"); runtime.nodeRegistry.unregister(nodeClient.connId); }); }); diff --git a/src/gateway/server.auth.default-token.suite.ts b/src/gateway/server.auth.default-token.suite.ts index 14e06777c46d..1814dc44e5e1 100644 --- a/src/gateway/server.auth.default-token.suite.ts +++ b/src/gateway/server.auth.default-token.suite.ts @@ -165,6 +165,9 @@ export function registerDefaultAuthTokenSuite(): void { expect(payload?.features?.capabilities).toContain( GATEWAY_SERVER_CAPS.NODE_WORKER_BUNDLE_RETENTION, ); + expect(payload?.features?.capabilities).toContain( + GATEWAY_SERVER_CAPS.NODE_WORKER_BUNDLE_STATUS, + ); expect(payload?.features?.capabilities).toContain( GATEWAY_SERVER_CAPS.SYSTEM_AGENT_WIZARD_CANCEL, ); diff --git a/src/gateway/server/ws-connection/connect-hello.ts b/src/gateway/server/ws-connection/connect-hello.ts index 5b7558d52ce9..e6a0ff3f7f7c 100644 --- a/src/gateway/server/ws-connection/connect-hello.ts +++ b/src/gateway/server/ws-connection/connect-hello.ts @@ -136,6 +136,7 @@ export async function sendGatewayHello( GATEWAY_SERVER_CAPS.CHAT_SEND_ROUTING_CONTRACT, GATEWAY_SERVER_CAPS.GATEWAY_RESTART_TARGET_SAFE, GATEWAY_SERVER_CAPS.NODE_WORKER_BUNDLE_RETENTION, + GATEWAY_SERVER_CAPS.NODE_WORKER_BUNDLE_STATUS, GATEWAY_SERVER_CAPS.SYSTEM_AGENT_WIZARD_CANCEL, GATEWAY_SERVER_CAPS.SYSTEM_AGENT_SETUP_MODEL_REF, GATEWAY_SERVER_CAPS.TASK_SUGGESTIONS_ACCEPT_MODES, diff --git a/src/gateway/worker-environments/node-workspace-retain-coordinator.test.ts b/src/gateway/worker-environments/node-workspace-retain-coordinator.test.ts index 811663dfc036..4d645fdac305 100644 --- a/src/gateway/worker-environments/node-workspace-retain-coordinator.test.ts +++ b/src/gateway/worker-environments/node-workspace-retain-coordinator.test.ts @@ -1,7 +1,10 @@ import { describe, expect, it, vi } from "vitest"; import { NODE_WORKER_WORKSPACE_RETAIN_COMMAND } from "../../infra/node-commands.js"; import { NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE } from "../../infra/node-runner-inventory.js"; -import { NODE_WORKER_BUNDLE_RETAIN_MAX_HASHES } from "../../worker/node-workspace-retain-protocol.js"; +import { + NODE_WORKER_BUNDLE_RETAIN_MAX_HASHES, + NODE_WORKER_RETAIN_REQUEST_MAX_BYTES, +} from "../../worker/node-workspace-retain-protocol.js"; import type { NodeWorkerSupervisorNodeProof, NodeWorkerSupervisorTransport, @@ -10,6 +13,10 @@ import { createNodeWorkspaceRetainCoordinator } from "./node-workspace-retain-co import type { WorkerSessionPlacementStore } from "./placement-store.js"; import type { WorkerEnvironmentService } from "./service.js"; +type NodeWorkerBundleStatusObservation = NonNullable< + ReturnType> +>; + const node = { nodeId: "node-1", connId: "connection-1", @@ -18,7 +25,12 @@ const node = { clientId: "node-host", clientMode: "node", protocolFeature: NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE, - workerHost: { enabled: true, capacity: "available", bundleRetention: 1 }, + workerHost: { + enabled: true, + capacity: "available", + bundleRetention: 1, + bundleStatus: 1, + }, commands: [], } as const; @@ -85,17 +97,40 @@ function createHarness( deleted: number; hasMore: boolean; bundleGeneration?: number; + bundleStatus?: { bundleHash: string; status: "installed" | "missing" }; }>; node?: NodeWorkerSupervisorNodeProof; + currentBundleStatus?: NodeWorkerBundleStatusObservation; + invokeError?: string; + onInvoke?: (index: number) => void; } = {}, ) { const results = [...(params.results ?? [{ applied: true, deleted: 0, hasMore: false }])]; - const invoke = vi.fn(async () => ({ - ok: true, - payloadJSON: JSON.stringify(results.shift() ?? { applied: true, deleted: 0, hasMore: false }), - })); + let invokeIndex = 0; + const invoke = vi.fn(async () => { + params.onInvoke?.(invokeIndex++); + if (params.invokeError) { + return { ok: false, error: { code: "UNAVAILABLE", message: params.invokeError } }; + } + return { + ok: true, + payloadJSON: JSON.stringify(results.shift() ?? { applied: true, deleted: 0, hasMore: false }), + }; + }); + let currentBundleStatus = params.currentBundleStatus; + const acceptBundleStatus = vi.fn( + ( + _node: NodeWorkerSupervisorNodeProof, + observation: NodeWorkerBundleStatusObservation | undefined, + ) => { + currentBundleStatus = observation; + return true; + }, + ); const transport: NodeWorkerSupervisorTransport = { listCurrentNodes: async () => [params.node ?? node], + getBundleStatus: () => currentBundleStatus, + acceptBundleStatus, isCurrent: () => true, invoke, }; @@ -111,7 +146,7 @@ function createHarness( warn, }); coordinator.bindTransport(transport); - return { coordinator, invoke, warn }; + return { coordinator, invoke, warn, acceptBundleStatus }; } describe("node workspace retain coordinator", () => { @@ -152,6 +187,228 @@ describe("node workspace retain coordinator", () => { await coordinator.stop(); }); + it("keeps prior retention nodes compatible without sending a status query", async () => { + const { coordinator, invoke, acceptBundleStatus } = createHarness({ + node: { + ...node, + workerHost: { + enabled: true, + capacity: "available", + bundleRetention: 1, + }, + }, + environments: [ + environment({ + bootstrapReceipt: { + bundleHash: "b".repeat(64), + openclawVersion: "2026.8.9", + protocolFeatures: [], + installKind: "bundle", + }, + }), + ], + }); + + await coordinator.start(); + + expect(invoke.mock.calls[0]?.[0].params).toMatchObject({ + bundleHashes: ["b".repeat(64)], + }); + expect(invoke.mock.calls[0]?.[0].params).not.toHaveProperty("bundleStatusHash"); + expect(acceptBundleStatus).toHaveBeenCalledWith(expect.any(Object), undefined); + await coordinator.stop(); + }); + + it("accepts a validated installed bundle status with the Gateway-owned version", async () => { + const bundleHash = "b".repeat(64); + const { coordinator, invoke, acceptBundleStatus } = createHarness({ + currentBundleStatus: { + bundleHash, + status: { status: "installed", version: "2026.8.9" }, + }, + environments: [ + environment({ + bootstrapReceipt: { + bundleHash, + openclawVersion: "2026.8.9", + protocolFeatures: [], + installKind: "bundle", + }, + }), + ], + results: [ + { + applied: true, + deleted: 0, + hasMore: false, + bundleGeneration: 3, + bundleStatus: { bundleHash, status: "installed" }, + }, + ], + }); + + await coordinator.start(); + + expect(invoke.mock.calls[0]?.[0].params).toMatchObject({ bundleStatusHash: bundleHash }); + expect(acceptBundleStatus).toHaveBeenCalledWith(node, { + bundleHash, + status: { status: "installed", version: "2026.8.9" }, + }); + await coordinator.stop(); + }); + + it("accepts status only from the final pass for the exact requested hash", async () => { + const bundleHash = "b".repeat(64); + const { coordinator, acceptBundleStatus } = createHarness({ + environments: [ + environment({ + bootstrapReceipt: { + bundleHash, + openclawVersion: "2026.8.9", + protocolFeatures: [], + installKind: "bundle", + }, + }), + ], + results: [ + { + applied: true, + deleted: 1, + hasMore: true, + bundleStatus: { bundleHash, status: "installed" }, + }, + { + applied: true, + deleted: 0, + hasMore: false, + bundleStatus: { bundleHash, status: "missing" }, + }, + ], + }); + + await coordinator.start(); + + expect(acceptBundleStatus).toHaveBeenCalledTimes(1); + expect(acceptBundleStatus).toHaveBeenCalledWith(node, { + bundleHash, + status: { status: "missing" }, + }); + await coordinator.stop(); + }); + + it("clears the previous hash before a new authoritative inspection can fail", async () => { + const previousHash = "b".repeat(64); + const currentHash = "c".repeat(64); + const { coordinator, acceptBundleStatus, warn } = createHarness({ + currentBundleStatus: { + bundleHash: previousHash, + status: { status: "installed", version: "2026.8.8" }, + }, + environments: [ + environment({ + bootstrapReceipt: { + bundleHash: currentHash, + openclawVersion: "2026.8.9", + protocolFeatures: [], + installKind: "bundle", + }, + }), + ], + invokeError: "maintenance unavailable", + }); + + await coordinator.start(); + + expect(acceptBundleStatus).toHaveBeenCalledWith(node, undefined); + expect(warn).toHaveBeenCalledWith(expect.stringContaining("maintenance unavailable")); + await coordinator.stop(); + }); + + it("clears status when a newer environment becomes authoritative during cleanup", async () => { + const bundleHash = "b".repeat(64); + const environments = [ + environment({ + bootstrapReceipt: { + bundleHash, + openclawVersion: "2026.8.9", + protocolFeatures: [], + installKind: "bundle", + }, + }), + ]; + const { coordinator, acceptBundleStatus } = createHarness({ + environments, + results: [ + { + applied: true, + deleted: 1, + hasMore: true, + bundleStatus: { bundleHash, status: "installed" }, + }, + { + applied: true, + deleted: 0, + hasMore: false, + bundleStatus: { bundleHash, status: "installed" }, + }, + ], + onInvoke: (index) => { + if (index !== 0) { + return; + } + environments.splice( + 0, + 1, + environment({ + environmentId: "environment-new", + createdAtMs: 3, + bootstrapReceipt: { + bundleHash: "c".repeat(64), + openclawVersion: "2026.8.10", + protocolFeatures: [], + installKind: "bundle", + }, + }), + ); + }, + }); + + await coordinator.start(); + + expect(acceptBundleStatus).toHaveBeenCalledTimes(1); + expect(acceptBundleStatus).toHaveBeenCalledWith(node, undefined); + await coordinator.stop(); + }); + + it("clears status when the node echoes a different bundle hash", async () => { + const bundleHash = "b".repeat(64); + const { coordinator, acceptBundleStatus } = createHarness({ + environments: [ + environment({ + bootstrapReceipt: { + bundleHash, + openclawVersion: "2026.8.9", + protocolFeatures: [], + installKind: "bundle", + }, + }), + ], + results: [ + { + applied: true, + deleted: 0, + hasMore: false, + bundleStatus: { bundleHash: "c".repeat(64), status: "installed" }, + }, + ], + }); + + await coordinator.start(); + + expect(acceptBundleStatus).toHaveBeenCalledWith(node, undefined); + await coordinator.stop(); + }); + it("keeps workspace retention compatible when bundle cleanup is not advertised", async () => { const { coordinator, invoke } = createHarness({ node: { @@ -192,6 +449,62 @@ describe("node workspace retain coordinator", () => { await coordinator.stop(); }); + it("keeps bounded bundle retention when only the status query exceeds one MiB", async () => { + const attachedCount = 1_241; + const environments = Array.from( + { length: NODE_WORKER_BUNDLE_RETAIN_MAX_HASHES }, + (_, index) => { + const suffix = index.toString(16).padStart(8, "0"); + const attached = index < attachedCount; + const environmentPadding = + index < attachedCount - 1 ? 220 : index === attachedCount - 1 ? 31 : 0; + const sessionPadding = + index < attachedCount - 1 ? 224 : index === attachedCount - 1 ? 31 : 0; + return environment({ + environmentId: `environment-${"e".repeat(environmentPadding)}-${suffix}`, + attachedSessionIds: attached ? [`session-${"s".repeat(sessionPadding)}-${suffix}`] : [], + createdAtMs: index === NODE_WORKER_BUNDLE_RETAIN_MAX_HASHES - 1 ? 10 : 1, + bootstrapReceipt: { + bundleHash: index.toString(16).padStart(64, "0"), + openclawVersion: "2026.8.9", + protocolFeatures: [], + installKind: "bundle", + }, + }); + }, + ); + const placements = environments.slice(0, attachedCount).map((entry, index) => + placement({ + sessionId: entry.attachedSessionIds[0], + environmentId: entry.environmentId, + workerBundleHash: index.toString(16).padStart(64, "0"), + }), + ); + const { coordinator, invoke, warn } = createHarness({ environments, placements }); + + await coordinator.start(); + + const input = invoke.mock.calls[0]?.[0].params as Record; + expect(input.bundleHashes).toHaveLength(NODE_WORKER_BUNDLE_RETAIN_MAX_HASHES); + expect(input).not.toHaveProperty("bundleStatusHash"); + expect(Buffer.byteLength(JSON.stringify(input), "utf8")).toBeLessThanOrEqual( + NODE_WORKER_RETAIN_REQUEST_MAX_BYTES, + ); + expect( + Buffer.byteLength( + JSON.stringify({ + ...input, + bundleStatusHash: (NODE_WORKER_BUNDLE_RETAIN_MAX_HASHES - 1) + .toString(16) + .padStart(64, "0"), + }), + "utf8", + ), + ).toBeGreaterThan(NODE_WORKER_RETAIN_REQUEST_MAX_BYTES); + expect(warn).not.toHaveBeenCalled(); + await coordinator.stop(); + }); + it("omits bundle hashes when the combined maintenance request exceeds one MiB", async () => { const environments = Array.from( { length: NODE_WORKER_BUNDLE_RETAIN_MAX_HASHES }, diff --git a/src/gateway/worker-environments/node-workspace-retain-coordinator.ts b/src/gateway/worker-environments/node-workspace-retain-coordinator.ts index 6f4148272f6c..776d53c9166d 100644 --- a/src/gateway/worker-environments/node-workspace-retain-coordinator.ts +++ b/src/gateway/worker-environments/node-workspace-retain-coordinator.ts @@ -1,7 +1,10 @@ import { randomUUID } from "node:crypto"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { NODE_WORKER_WORKSPACE_RETAIN_COMMAND } from "../../infra/node-commands.js"; -import { NODE_WORKER_BUNDLE_RETENTION_VERSION } from "../../infra/node-runner-inventory.js"; +import { + NODE_WORKER_BUNDLE_RETENTION_VERSION, + NODE_WORKER_BUNDLE_STATUS_VERSION, +} from "../../infra/node-runner-inventory.js"; import { NODE_WORKER_BUNDLE_RETAIN_MAX_HASHES, NODE_WORKER_RETAIN_REQUEST_MAX_BYTES, @@ -46,6 +49,20 @@ function nodeEnvironments(options: NodeWorkspaceRetainCoordinatorOptions, nodeId ); } +function bundleStatusTargetForNode(options: NodeWorkspaceRetainCoordinatorOptions, nodeId: string) { + return nodeEnvironments(options, nodeId) + .filter( + (environment) => + environment.bootstrapReceipt !== null && + !TERMINAL_ENVIRONMENT_STATES.has(environment.state), + ) + .toSorted( + (left, right) => + right.createdAtMs - left.createdAtMs || + left.environmentId.localeCompare(right.environmentId), + )[0]?.bootstrapReceipt; +} + function snapshotBundleHashesForNode( options: NodeWorkspaceRetainCoordinatorOptions, nodeId: string, @@ -135,6 +152,8 @@ export function createNodeWorkspaceRetainCoordinator( const retainedBundleHashes = snapshotBundleHashesForNode(options, node.nodeId); const bundleRetentionSupported = node.workerHost.bundleRetention === NODE_WORKER_BUNDLE_RETENTION_VERSION; + const bundleStatusSupported = + node.workerHost.bundleStatus === NODE_WORKER_BUNDLE_STATUS_VERSION; const baseInput: NodeWorkerWorkspaceRetainInput = { version: 1, gatewayNamespace: options.gatewayNamespace, @@ -145,16 +164,39 @@ export function createNodeWorkspaceRetainCoordinator( const priorGeneration = acknowledgedBundleGenerationByNode.get(node.nodeId); const acknowledgedBundleGeneration = priorGeneration?.connId === node.connId ? priorGeneration.generation : undefined; - const candidateInput: NodeWorkerWorkspaceRetainInput = { + const retentionInput: NodeWorkerWorkspaceRetainInput = { ...baseInput, bundleHashes: retainedBundleHashes, ...(acknowledgedBundleGeneration !== undefined ? { acknowledgedBundleGeneration } : {}), }; const bundleHashesFit = retainedBundleHashes.length <= NODE_WORKER_BUNDLE_RETAIN_MAX_HASHES && - Buffer.byteLength(JSON.stringify(candidateInput), "utf8") <= + Buffer.byteLength(JSON.stringify(retentionInput), "utf8") <= NODE_WORKER_RETAIN_REQUEST_MAX_BYTES; - const input = bundleRetentionSupported && bundleHashesFit ? candidateInput : baseInput; + const bundleStatusTarget = bundleStatusSupported + ? bundleStatusTargetForNode(options, node.nodeId) + : undefined; + const statusInput = + bundleStatusTarget && retainedBundleHashes.includes(bundleStatusTarget.bundleHash) + ? { ...retentionInput, bundleStatusHash: bundleStatusTarget.bundleHash } + : undefined; + const statusInputFits = + statusInput !== undefined && + Buffer.byteLength(JSON.stringify(statusInput), "utf8") <= + NODE_WORKER_RETAIN_REQUEST_MAX_BYTES; + const input = + bundleRetentionSupported && bundleHashesFit + ? statusInput && statusInputFits + ? statusInput + : retentionInput + : baseInput; + const previousBundleStatus = currentTransport.getBundleStatus?.(node.nodeId); + if ( + !input.bundleStatusHash || + (previousBundleStatus && previousBundleStatus.bundleHash !== input.bundleStatusHash) + ) { + currentTransport.acceptBundleStatus?.(node, undefined); + } if (bundleRetentionSupported && !bundleHashesFit) { options.warn( `Node bundle retention skipped (${node.nodeId}): ${retainedBundleHashes.length} retained hashes exceed the bounded maintenance request`, @@ -192,6 +234,30 @@ export function createNodeWorkspaceRetainCoordinator( }); } if (!retained.applied || !retained.hasMore) { + const bundleStatus = retained.bundleStatus; + const requestedBundleHash = input.bundleStatusHash; + const currentStatusTarget = requestedBundleHash + ? bundleStatusTargetForNode(options, node.nodeId) + : undefined; + const statusTargetMatches = + currentStatusTarget != null && + requestedBundleHash !== undefined && + currentStatusTarget.bundleHash === requestedBundleHash; + const statusMatches = + retained.applied && + statusTargetMatches && + bundleStatus?.bundleHash === requestedBundleHash; + if (statusMatches && currentStatusTarget && bundleStatus) { + currentTransport.acceptBundleStatus?.(node, { + bundleHash: currentStatusTarget.bundleHash, + status: + bundleStatus.status === "installed" + ? { status: "installed", version: currentStatusTarget.openclawVersion } + : { status: "missing" }, + }); + } else if (input.bundleStatusHash) { + currentTransport.acceptBundleStatus?.(node, undefined); + } return; } } diff --git a/src/infra/node-runner-inventory.ts b/src/infra/node-runner-inventory.ts index 0de640c41d50..ff24fa1344ae 100644 --- a/src/infra/node-runner-inventory.ts +++ b/src/infra/node-runner-inventory.ts @@ -7,6 +7,7 @@ export const NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE = "node-worker-supervisor-v export const NODE_WORKER_SUPERVISOR_BUILD_PROTOCOL_FEATURE = "node-worker-supervisor-v2"; export const NODE_WORKER_SUPERVISOR_LEGACY_PROTOCOL_FEATURE = "node-worker-supervisor-v1"; export const NODE_WORKER_BUNDLE_RETENTION_VERSION = 1; +export const NODE_WORKER_BUNDLE_STATUS_VERSION = 1; export const NODE_RUNNER_UPDATE_REQUIRED_ISSUE = { code: "update-required", @@ -24,6 +25,7 @@ export type NodeWorkerHostDeclaration = capacity: "available" | "full"; bundlePrewarm?: typeof WORKER_BUNDLE_PREWARM_VERSION; bundleRetention?: typeof NODE_WORKER_BUNDLE_RETENTION_VERSION; + bundleStatus?: typeof NODE_WORKER_BUNDLE_STATUS_VERSION; }; export type NodeRunnerInventoryDeclaration = @@ -49,7 +51,7 @@ function parseWorkerHostDeclaration(value: unknown): NodeWorkerHostDeclaration | } if ( keys.length < 2 || - keys.length > 4 || + keys.length > 5 || !keys.includes("enabled") || !keys.includes("capacity") || keys.some( @@ -57,12 +59,16 @@ function parseWorkerHostDeclaration(value: unknown): NodeWorkerHostDeclaration | key !== "enabled" && key !== "capacity" && key !== "bundlePrewarm" && - key !== "bundleRetention", + key !== "bundleRetention" && + key !== "bundleStatus", ) || (value.capacity !== "available" && value.capacity !== "full") || (value.bundlePrewarm !== undefined && value.bundlePrewarm !== WORKER_BUNDLE_PREWARM_VERSION) || (value.bundleRetention !== undefined && - value.bundleRetention !== NODE_WORKER_BUNDLE_RETENTION_VERSION) + value.bundleRetention !== NODE_WORKER_BUNDLE_RETENTION_VERSION) || + (value.bundleStatus !== undefined && + value.bundleStatus !== NODE_WORKER_BUNDLE_STATUS_VERSION) || + (value.bundleStatus !== undefined && value.bundleRetention === undefined) ) { return null; } @@ -75,6 +81,9 @@ function parseWorkerHostDeclaration(value: unknown): NodeWorkerHostDeclaration | ...(value.bundleRetention === NODE_WORKER_BUNDLE_RETENTION_VERSION ? { bundleRetention: NODE_WORKER_BUNDLE_RETENTION_VERSION } : {}), + ...(value.bundleStatus === NODE_WORKER_BUNDLE_STATUS_VERSION + ? { bundleStatus: NODE_WORKER_BUNDLE_STATUS_VERSION } + : {}), }; } diff --git a/src/node-host/invoke-worker-supervisor.test.ts b/src/node-host/invoke-worker-supervisor.test.ts index 6a6345ee3d34..63ba7f724cf6 100644 --- a/src/node-host/invoke-worker-supervisor.test.ts +++ b/src/node-host/invoke-worker-supervisor.test.ts @@ -275,9 +275,14 @@ describe("node-host worker supervisor commands", () => { it("combines bounded bundle cleanup with the workspace retain snapshot", async () => { const input = launchInput(); const supervisor = supervisorWith(fullReceipt(input)); - const retainBundles = vi.fn(async () => ({ deleted: 2, hasMore: true, generation: 4 })); + const retainBundles = vi.fn(async () => ({ deleted: 2, hasMore: false, generation: 4 })); + const inspectBundle = vi.fn(async () => ({ + bundleHash: "a".repeat(64), + status: "installed" as const, + })); const bundleInstaller = { ensure: vi.fn(), + inspect: inspectBundle, retain: retainBundles, } as unknown as NodeWorkerBundleInstallerControl; const retain = { @@ -288,6 +293,7 @@ describe("node-host worker supervisor commands", () => { retain: [], bundleHashes: ["a".repeat(64)], acknowledgedBundleGeneration: 3, + bundleStatusHash: "a".repeat(64), } as const; const { result } = await invokePrivate({ @@ -302,6 +308,49 @@ describe("node-host worker supervisor commands", () => { bundleHashes: ["a".repeat(64)], acknowledgedGeneration: 3, }); + expect(inspectBundle).toHaveBeenCalledWith({ + gatewayNamespace: input.gatewayNamespace, + bundleHash: "a".repeat(64), + }); + expect(JSON.parse(result?.payloadJSON ?? "{}")).toEqual({ + applied: true, + deleted: 0, + hasMore: false, + bundleDeleted: 2, + bundleGeneration: 4, + bundleStatus: { bundleHash: "a".repeat(64), status: "installed" }, + }); + }); + + it("defers full bundle status validation until the cleanup snapshot is terminal", async () => { + const input = launchInput(); + const supervisor = supervisorWith(fullReceipt(input)); + const inspectBundle = vi.fn(async () => ({ + bundleHash: "a".repeat(64), + status: "installed" as const, + })); + const bundleInstaller = { + ensure: vi.fn(), + inspect: inspectBundle, + retain: vi.fn(async () => ({ deleted: 2, hasMore: true, generation: 4 })), + } as unknown as NodeWorkerBundleInstallerControl; + + const { result } = await invokePrivate({ + command: NODE_WORKER_WORKSPACE_RETAIN_COMMAND, + paramsJSON: JSON.stringify({ + version: 1, + gatewayNamespace: input.gatewayNamespace, + controllerId: "controller-1", + sequence: 1, + retain: [], + bundleHashes: ["a".repeat(64)], + bundleStatusHash: "a".repeat(64), + }), + supervisor, + bundleInstaller, + }); + + expect(inspectBundle).not.toHaveBeenCalled(); expect(JSON.parse(result?.payloadJSON ?? "{}")).toEqual({ applied: true, deleted: 0, diff --git a/src/node-host/node-worker-bundle-installer.test.ts b/src/node-host/node-worker-bundle-installer.test.ts index 050ba0be2519..50524c0c5fb0 100644 --- a/src/node-host/node-worker-bundle-installer.test.ts +++ b/src/node-host/node-worker-bundle-installer.test.ts @@ -151,6 +151,40 @@ describe("node worker bundle installer", () => { ).resolves.toContain(fixture.input.build.bundleHash); }); + it("reports installed only after full bundle validation", async () => { + const fixture = await bundleFixture(); + const served = await serve(fixture.archive, fixture.input.archive.token); + const installer = new NodeWorkerBundleInstaller({ root }); + + await expect( + installer.inspect({ + gatewayNamespace: fixture.input.gatewayNamespace, + bundleHash: fixture.input.build.bundleHash, + }), + ).resolves.toEqual({ bundleHash: fixture.input.build.bundleHash, status: "missing" }); + await installer.ensure({ input: fixture.input, gatewayUrl: served.gatewayUrl }); + await expect( + installer.inspect({ + gatewayNamespace: fixture.input.gatewayNamespace, + bundleHash: fixture.input.build.bundleHash, + }), + ).resolves.toEqual({ bundleHash: fixture.input.build.bundleHash, status: "installed" }); + + const bundleDir = path.join( + root, + fixture.input.gatewayNamespace, + "bundles", + fixture.input.build.bundleHash, + ); + await fs.writeFile(path.join(bundleDir, "worker.mjs"), "tampered\n"); + await expect( + installer.inspect({ + gatewayNamespace: fixture.input.gatewayNamespace, + bundleHash: fixture.input.build.bundleHash, + }), + ).resolves.toEqual({ bundleHash: fixture.input.build.bundleHash, status: "missing" }); + }); + it("prunes superseded bundle artifacts in bounded passes while retaining the latest install", async () => { const fixture = await bundleFixture(); const served = await serve(fixture.archive, fixture.input.archive.token); diff --git a/src/node-host/node-worker-bundle-installer.ts b/src/node-host/node-worker-bundle-installer.ts index e1537f8dc0b1..7d978cd5dbf2 100644 --- a/src/node-host/node-worker-bundle-installer.ts +++ b/src/node-host/node-worker-bundle-installer.ts @@ -324,6 +324,25 @@ export class NodeWorkerBundleInstaller { }); } + async inspect(params: { + gatewayNamespace: string; + bundleHash: string; + }): Promise<{ bundleHash: string; status: "installed" | "missing" }> { + return await this.#operations.enqueue(params.gatewayNamespace, async () => { + const bundleDir = path.join( + this.#root, + params.gatewayNamespace, + "bundles", + params.bundleHash, + ); + const receipt = await readReceipt(bundleDir); + const installed = + receipt?.bundleHash === params.bundleHash && + (await validateInstalledBundle(bundleDir, receipt)); + return { bundleHash: params.bundleHash, status: installed ? "installed" : "missing" }; + }); + } + async retain(params: { gatewayNamespace: string; bundleHashes: readonly string[]; @@ -388,4 +407,4 @@ export class NodeWorkerBundleInstaller { } export type NodeWorkerBundleInstallerControl = Pick & - Partial>; + Partial>; diff --git a/src/node-host/node-worker-supervisor-commands.ts b/src/node-host/node-worker-supervisor-commands.ts index 09a738999921..eade988f2367 100644 --- a/src/node-host/node-worker-supervisor-commands.ts +++ b/src/node-host/node-worker-supervisor-commands.ts @@ -187,17 +187,35 @@ export async function invokeNodeWorkerSupervisorCommand(params: { : {}), }); } + const hasMore = workspace.hasMore || bundles?.hasMore === true; + const inspectBundle = params.bundleInstaller?.inspect?.bind(params.bundleInstaller); + if (workspace.applied && input.bundleStatusHash && !hasMore && !inspectBundle) { + throw new Error("node worker bundle status unavailable"); + } + const bundleStatus = + workspace.applied && input.bundleStatusHash && !hasMore && inspectBundle + ? await inspectBundle({ + gatewayNamespace: input.gatewayNamespace, + bundleHash: input.bundleStatusHash, + }) + : undefined; return { handled: true, ok: true, - payload: bundles - ? { - ...workspace, - bundleDeleted: bundles.deleted, - bundleGeneration: bundles.generation, - hasMore: workspace.hasMore || bundles.hasMore, - } - : workspace, + payload: + bundles || bundleStatus + ? { + ...workspace, + ...(bundles + ? { + bundleDeleted: bundles.deleted, + bundleGeneration: bundles.generation, + hasMore, + } + : {}), + ...(bundleStatus ? { bundleStatus } : {}), + } + : workspace, }; } const receipt = diff --git a/src/node-host/runner.test.ts b/src/node-host/runner.test.ts index 06d358099dca..9b43013212cb 100644 --- a/src/node-host/runner.test.ts +++ b/src/node-host/runner.test.ts @@ -741,6 +741,30 @@ describe("runNodeHost", () => { }); }); + options?.onHelloOk?.({ + protocol: 4, + features: { + methods: [], + events: [], + capabilities: [ + GATEWAY_SERVER_CAPS.NODE_WORKER_BUNDLE_RETENTION, + GATEWAY_SERVER_CAPS.NODE_WORKER_BUNDLE_STATUS, + ], + }, + } as unknown as Parameters>[0]); + await vi.waitFor(() => { + expect(client?.request).toHaveBeenCalledWith(NODE_RUNNER_INVENTORY_UPDATE_METHOD, { + protocolFeatures: [NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE], + workerHost: { + enabled: true, + capacity: "available", + bundlePrewarm: 1, + bundleRetention: 1, + bundleStatus: 1, + }, + }); + }); + mocks.runnerAvailabilityChanged?.(false); await vi.waitFor(() => { expect(client?.request).toHaveBeenLastCalledWith(NODE_RUNNER_INVENTORY_UPDATE_METHOD, { @@ -750,6 +774,7 @@ describe("runNodeHost", () => { capacity: "full", bundlePrewarm: 1, bundleRetention: 1, + bundleStatus: 1, }, }); }); @@ -763,6 +788,7 @@ describe("runNodeHost", () => { capacity: "available", bundlePrewarm: 1, bundleRetention: 1, + bundleStatus: 1, }, }); }); diff --git a/src/node-host/runner.ts b/src/node-host/runner.ts index 9b8960351ce1..21633b5b49c2 100644 --- a/src/node-host/runner.ts +++ b/src/node-host/runner.ts @@ -16,6 +16,7 @@ import { getMachineDisplayName } from "../infra/machine-name.js"; import { NODE_RUNNER_INVENTORY_UPDATE_METHOD, NODE_WORKER_BUNDLE_RETENTION_VERSION, + NODE_WORKER_BUNDLE_STATUS_VERSION, NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE, } from "../infra/node-runner-inventory.js"; import { VERSION } from "../version.js"; @@ -260,6 +261,7 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise { let gatewayConnectionGeneration = 0; let connectedGatewayProtocol = 0; let gatewaySupportsBundleRetention = false; + let gatewaySupportsBundleStatus = false; let optionalPublicationStates = new Map< NodeOptionalPublicationMethod, NodeOptionalPublicationState @@ -277,6 +279,7 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise { gatewayHelloReceived = false; connectedGatewayProtocol = 0; gatewaySupportsBundleRetention = false; + gatewaySupportsBundleStatus = false; retireOptionalPublications(); }; @@ -477,6 +480,9 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise { ...(gatewaySupportsBundleRetention ? { bundleRetention: NODE_WORKER_BUNDLE_RETENTION_VERSION } : {}), + ...(gatewaySupportsBundleRetention && gatewaySupportsBundleStatus + ? { bundleStatus: NODE_WORKER_BUNDLE_STATUS_VERSION } + : {}), } : { enabled: false }, }, @@ -553,6 +559,9 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise { gatewaySupportsBundleRetention = hello.features?.capabilities?.includes(GATEWAY_SERVER_CAPS.NODE_WORKER_BUNDLE_RETENTION) === true; + gatewaySupportsBundleStatus = + hello.features?.capabilities?.includes(GATEWAY_SERVER_CAPS.NODE_WORKER_BUNDLE_STATUS) === + true; retireOptionalPublications(); optionalPublicationStates = new Map(); if (opts.stopAfterFirstConnect) { diff --git a/src/shared/node-list-types.ts b/src/shared/node-list-types.ts index c8f23bdb5648..465e8e9667c4 100644 --- a/src/shared/node-list-types.ts +++ b/src/shared/node-list-types.ts @@ -2,6 +2,10 @@ import type { RuntimeTargetIssue } from "../../packages/gateway-protocol/src/sch import type { NodePluginToolDescriptor } from "../../packages/gateway-protocol/src/schema/nodes.js"; import type { ComputerUseCapabilityDescriptor } from "../plugins/computer-use-contract.js"; +export type NodeWorkerBundleStatus = + | { status: "installed"; version: string } + | { status: "missing" }; + /** Node record returned by gateway node-list endpoints. */ export type NodeListNode = { nodeId: string; @@ -23,6 +27,7 @@ export type NodeListNode = { computerUse?: ComputerUseCapabilityDescriptor; /** Connected node currently advertises full worker session hosting. */ sessionHost?: boolean; + workerBundle?: NodeWorkerBundleStatus; issues?: readonly RuntimeTargetIssue[]; nodePluginTools?: NodePluginToolDescriptor[]; permissions?: Record; diff --git a/src/worker/node-workspace-retain-protocol.test.ts b/src/worker/node-workspace-retain-protocol.test.ts index efcd205c532c..10aed91e8ff5 100644 --- a/src/worker/node-workspace-retain-protocol.test.ts +++ b/src/worker/node-workspace-retain-protocol.test.ts @@ -22,6 +22,7 @@ describe("node workspace retain protocol", () => { sequence: 4, bundleHashes: ["b".repeat(64), "a".repeat(64)], acknowledgedBundleGeneration: 3, + bundleStatusHash: "a".repeat(64), retain: [{ ...entry, environmentId: "environment-2", manifestRefs: null }, entry], }), ), @@ -32,6 +33,7 @@ describe("node workspace retain protocol", () => { sequence: 4, bundleHashes: ["a".repeat(64), "b".repeat(64)], acknowledgedBundleGeneration: 3, + bundleStatusHash: "a".repeat(64), retain: [entry, { ...entry, environmentId: "environment-2", manifestRefs: null }], }); }); @@ -54,6 +56,22 @@ describe("node workspace retain protocol", () => { ).toThrow("INVALID_REQUEST"); }); + it("rejects a bundle status hash that is not retained", () => { + expect(() => + parseNodeWorkerWorkspaceRetainInput( + JSON.stringify({ + version: 1, + gatewayNamespace: "gateway-test", + controllerId: "controller-1", + sequence: 1, + retain: [], + bundleHashes: ["a".repeat(64)], + bundleStatusHash: "b".repeat(64), + }), + ), + ).toThrow("must be retained"); + }); + it("rejects a bundle-generation acknowledgement without bundle hashes", () => { expect(() => parseNodeWorkerWorkspaceRetainInput( @@ -91,6 +109,7 @@ describe("node workspace retain protocol", () => { hasMore: false, bundleDeleted: 3, bundleGeneration: 4, + bundleStatus: { bundleHash: "a".repeat(64), status: "installed" }, }), ).toEqual({ applied: true, @@ -98,6 +117,7 @@ describe("node workspace retain protocol", () => { hasMore: false, bundleDeleted: 3, bundleGeneration: 4, + bundleStatus: { bundleHash: "a".repeat(64), status: "installed" }, }); expect( parseNodeWorkerWorkspaceRetainResult({ diff --git a/src/worker/node-workspace-retain-protocol.ts b/src/worker/node-workspace-retain-protocol.ts index c16d6c71d99b..2be492a3db39 100644 --- a/src/worker/node-workspace-retain-protocol.ts +++ b/src/worker/node-workspace-retain-protocol.ts @@ -53,6 +53,7 @@ const RetainInputSchema = z retain: z.array(RetainEntrySchema).max(RETAIN_MAX_ENTRIES), bundleHashes: BundleHashesSchema.optional(), acknowledgedBundleGeneration: z.number().int().min(0).max(Number.MAX_SAFE_INTEGER).optional(), + bundleStatusHash: z.string().regex(BUNDLE_HASH_PATTERN).optional(), }) .strict() .superRefine((input, context) => { @@ -62,8 +63,24 @@ const RetainInputSchema = z message: "acknowledgedBundleGeneration requires bundleHashes", }); } + if ( + input.bundleStatusHash !== undefined && + !input.bundleHashes?.includes(input.bundleStatusHash) + ) { + context.addIssue({ + code: "custom", + message: "bundleStatusHash must be retained by bundleHashes", + }); + } }); +const BundleStatusSchema = z + .object({ + bundleHash: z.string().regex(BUNDLE_HASH_PATTERN), + status: z.enum(["installed", "missing"]), + }) + .strict(); + const RetainResultSchema = z .object({ applied: z.boolean(), @@ -71,6 +88,7 @@ const RetainResultSchema = z hasMore: z.boolean(), bundleDeleted: z.number().int().min(0).max(Number.MAX_SAFE_INTEGER).optional(), bundleGeneration: z.number().int().min(0).max(Number.MAX_SAFE_INTEGER).optional(), + bundleStatus: BundleStatusSchema.optional(), }) .strict(); diff --git a/test/vitest/vitest.ui-isolated-paths.mjs b/test/vitest/vitest.ui-isolated-paths.mjs index 574cd989d0d2..c0e35c913e64 100644 --- a/test/vitest/vitest.ui-isolated-paths.mjs +++ b/test/vitest/vitest.ui-isolated-paths.mjs @@ -12,6 +12,7 @@ export const uiIsolatedTestFiles = [ "ui/src/pages/chat/chat-page-attachment-handoff.test.ts", "ui/src/pages/chat/chat-pane-attachment-handoff.test.ts", "ui/src/pages/chat/chat-pane-board.test.ts", + "ui/src/pages/chat/chat-pane-catalog.test.ts", "ui/src/pages/chat/chat-pane-history.test.ts", "ui/src/pages/chat/chat-pane-identity.test.ts", "ui/src/pages/chat/chat-pane-lifecycle.test.ts", diff --git a/ui/src/e2e/operator-admin.e2e.test.ts b/ui/src/e2e/operator-admin.e2e.test.ts index e8417695474b..b5d381ad559a 100644 --- a/ui/src/e2e/operator-admin.e2e.test.ts +++ b/ui/src/e2e/operator-admin.e2e.test.ts @@ -240,7 +240,7 @@ suite.define(() => { gateway.waitForRequest("exec.approvals.get"), ]); await expect.poll(() => page.getByText("Build Node", { exact: true }).isVisible()).toBe(true); - await expect.poll(() => page.getByText("connected", { exact: true }).isVisible()).toBe(true); + await expect.poll(() => page.getByText("connected", { exact: true }).count()).toBe(0); await page.getByText("Details", { exact: true }).click(); await expect .poll(() => page.getByText(/Capabilities: browser, filesystem/).isVisible()) diff --git a/ui/src/i18n/locales/en.ts b/ui/src/i18n/locales/en.ts index f9e7107c7eee..89224f11043e 100644 --- a/ui/src/i18n/locales/en.ts +++ b/ui/src/i18n/locales/en.ts @@ -589,6 +589,10 @@ export const en: TranslationMap = { versionDrift: "version drift", versionDriftTitle: "Device {nodeVersion}; Gateway {gatewayVersion}. Update the older component to align the fleet.", + workerVersion: "Worker {version}", + workerMissing: "worker missing", + workerMissingTitle: + "The Gateway-managed worker bundle is missing. Start a new session on this device to reinstall it.", manualWake: "manual wake required", manualWakeTitle: "The Gateway cannot wake an offline Windows device. Start the machine or restore its network connection.", diff --git a/ui/src/lib/nodes/inventory.test.ts b/ui/src/lib/nodes/inventory.test.ts index c44d5f4862ad..6a98632a6d36 100644 --- a/ui/src/lib/nodes/inventory.test.ts +++ b/ui/src/lib/nodes/inventory.test.ts @@ -43,6 +43,7 @@ describe("buildDeviceInventory", () => { commands: ["system.run"], version: "2026.6.11", coreVersion: "2026.7.2", + workerBundle: { status: "installed", version: "2026.8.9" }, uiVersion: "19.5", }, ], @@ -57,6 +58,43 @@ describe("buildDeviceInventory", () => { expect(entry.node?.caps).toEqual(["screen"]); expect(entry.node?.coreVersion).toBe("2026.7.2"); expect(entry.node?.uiVersion).toBe("19.5"); + expect(entry.node?.workerBundle).toEqual({ status: "installed", version: "2026.8.9" }); + }); + + it("preserves a valid missing worker bundle status", () => { + const groups = buildDeviceInventory({ + paired: [], + nodes: [ + { + nodeId: "node-1", + connected: true, + paired: true, + workerBundle: { status: "missing" }, + }, + ], + }); + + expect(firstGroup(groups).primary.node?.workerBundle).toEqual({ status: "missing" }); + }); + + it("drops malformed worker bundle status instead of exposing private fields", () => { + const groups = buildDeviceInventory({ + paired: [], + nodes: [ + { + nodeId: "node-1", + connected: true, + paired: true, + workerBundle: { + status: "installed", + version: "2026.8.9", + bundleHash: "a".repeat(64), + }, + }, + ], + }); + + expect(firstGroup(groups).primary.node?.workerBundle).toBeUndefined(); }); it("joins presence case-insensitively and prefers its display metadata", () => { diff --git a/ui/src/lib/nodes/inventory.ts b/ui/src/lib/nodes/inventory.ts index 48529094b7dd..0621cc5b3418 100644 --- a/ui/src/lib/nodes/inventory.ts +++ b/ui/src/lib/nodes/inventory.ts @@ -4,11 +4,13 @@ import { asFiniteNumber as optionalNumber } from "@openclaw/normalization-core/n // records (roles + tokens) and the node catalog (caps + live links). This module // joins them by id and groups duplicate pairings of the same client so the page // renders one row per machine instead of one row per historical keypair. +import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import type { PresenceEntry } from "../../api/types.ts"; import type { PairedDevice } from "./index.ts"; type NodeApprovalState = "approved" | "pending-approval" | "pending-reapproval" | "unapproved"; +type NodeWorkerBundleStatus = { status: "installed"; version: string } | { status: "missing" }; /** Typed projection of one raw `node.list` row. */ type NodeListEntry = { @@ -26,6 +28,7 @@ type NodeListEntry = { commands: string[]; approvalState?: NodeApprovalState; pendingRequestId?: string; + workerBundle?: NodeWorkerBundleStatus; connected: boolean; paired: boolean; connectedAtMs?: number; @@ -78,6 +81,20 @@ function stringList(value: unknown): string[] { .filter((entry): entry is string => entry !== undefined); } +function parseWorkerBundleStatus(value: unknown): NodeWorkerBundleStatus | undefined { + if (!isRecord(value)) { + return undefined; + } + const raw = value; + if (raw.status === "missing" && Object.keys(raw).length === 1) { + return { status: "missing" }; + } + const version = normalizeOptionalString(raw.version); + return raw.status === "installed" && version && Object.keys(raw).length === 2 + ? { status: "installed", version } + : undefined; +} + function parseNodeListEntry(raw: Record): NodeListEntry | null { const nodeId = normalizeOptionalString(raw.nodeId); if (!nodeId) { @@ -102,6 +119,7 @@ function parseNodeListEntry(raw: Record): NodeListEntry | null ? (approvalState as NodeApprovalState) : undefined, pendingRequestId: normalizeOptionalString(raw.pendingRequestId), + workerBundle: parseWorkerBundleStatus(raw.workerBundle), connected: raw.connected === true, paired: raw.paired === true, connectedAtMs: optionalNumber(raw.connectedAtMs), diff --git a/ui/src/pages/chat/chat-gateway.test.ts b/ui/src/pages/chat/chat-gateway.test.ts index 89de61f4eea8..145b0bbfbb55 100644 --- a/ui/src/pages/chat/chat-gateway.test.ts +++ b/ui/src/pages/chat/chat-gateway.test.ts @@ -939,8 +939,9 @@ describe("handleChatGatewayEvent", () => { text: "Use the deployment plan", createdAt: 3, kind: "steered", - pendingRunId: "active-run", + pendingRunId: "steer-request-run", sendRunId: "steer-request-run", + steerTargetRunId: "active-run", sessionKey: "main", }, ], diff --git a/ui/src/pages/chat/chat-send.test.ts b/ui/src/pages/chat/chat-send.test.ts index 7bef78c18d94..494dc7f3d001 100644 --- a/ui/src/pages/chat/chat-send.test.ts +++ b/ui/src/pages/chat/chat-send.test.ts @@ -8328,6 +8328,7 @@ describe("handleSendChat", () => { kind: "steered", pendingRunId: "steer-run", sendRunId: original.sendRunId, + steerTargetRunId: "active-run", text: original.text, }), ]); diff --git a/ui/src/pages/chat/steer-lifecycle.ts b/ui/src/pages/chat/steer-lifecycle.ts index 8c891f29e925..71b80ea43462 100644 --- a/ui/src/pages/chat/steer-lifecycle.ts +++ b/ui/src/pages/chat/steer-lifecycle.ts @@ -267,9 +267,11 @@ export function retireSteeredChipsForRequestRun( for (const item of landed) { // A started active turn can still exist only as an optimistic queue row. // Promote that target before its landed steer so stable transcript history - // cannot render the newer steer ahead of the original prompt. + // cannot render the newer steer ahead of the original prompt. Older persisted + // chips used pendingRunId as both identities, so retain it as the migration fallback. + const targetRunId = item.steerTargetRunId?.trim() || item.pendingRunId; const target = state.chatQueue.find( - (candidate) => candidate.id !== item.id && candidate.sendRunId === item.pendingRunId, + (candidate) => candidate.id !== item.id && candidate.sendRunId === targetRunId, ); if (target) { preserveQueuedUserTurn(state, target); @@ -426,6 +428,7 @@ export async function sendQueuedChatMessageWithQueueMode( sendRunId: claimed.sendRunId, sessionKey: claimed.sessionKey, agentId: claimed.agentId, + ...(claimed.steerTargetRunId ? { steerTargetRunId: claimed.steerTargetRunId } : {}), }; const steeringChip = buildInflightSteerChip(pendingItem, claimed.sendRunId, activeRunId); const pendingIndicator = isSteer diff --git a/ui/src/pages/devices/devices-page.test.ts b/ui/src/pages/devices/devices-page.test.ts index 856869530edf..2794bb4e86f5 100644 --- a/ui/src/pages/devices/devices-page.test.ts +++ b/ui/src/pages/devices/devices-page.test.ts @@ -145,8 +145,11 @@ function gatewaySnapshot( }; } -function gateway(client: GatewayBrowserClient | null): ApplicationContext["gateway"] { - const snapshot: ApplicationGatewaySnapshot = { +function gateway( + client: GatewayBrowserClient | null, + snapshotOverride?: ApplicationGatewaySnapshot, +): ApplicationContext["gateway"] { + const snapshot: ApplicationGatewaySnapshot = snapshotOverride ?? { client, phase: "stopped", offlineStable: false, @@ -233,6 +236,34 @@ describe("DevicesPage gateway lifecycle", () => { expect(page.ensureInitialData).toHaveBeenCalledOnce(); }); + it("reloads node status when runner inventory changes", async () => { + const request = vi.fn(async (method: string) => + method === "node.list" ? { nodes: [] } : { paired: [], pending: [] }, + ); + const client = { request } as unknown as GatewayBrowserClient; + let onEvent: ((event: { event: string; payload?: unknown }) => void) | undefined; + const currentGateway = gateway(client, gatewaySnapshot(client, true)); + currentGateway.subscribeEvents = vi.fn((listener) => { + onEvent = listener as typeof onEvent; + return () => undefined; + }); + const page = document.createElement("openclaw-devices-page") as TestDevicesPage; + page.context = { + gateway: currentGateway, + runtimeConfig: { + state: { configSnapshot: {}, configLoading: false }, + subscribe: vi.fn(() => () => undefined), + }, + } as unknown as ApplicationContext; + document.body.append(page); + await vi.waitFor(() => expect(onEvent).toBeDefined()); + + onEvent?.({ event: "node.runnerInventory.changed", payload: { nodeId: "node-1" } }); + + await vi.waitFor(() => expect(request).toHaveBeenCalledWith("node.list", {})); + page.remove(); + }); + it("retries a node load after a same-client disconnect", async () => { const first = deferred<{ nodes: Array> }>(); const second = deferred<{ nodes: Array> }>(); diff --git a/ui/src/pages/devices/devices-page.ts b/ui/src/pages/devices/devices-page.ts index c9b3fc50703b..739249b25c75 100644 --- a/ui/src/pages/devices/devices-page.ts +++ b/ui/src/pages/devices/devices-page.ts @@ -163,7 +163,11 @@ class DevicesPage extends OpenClawLightDomElement { if (event.event === "device.pair.requested" || event.event === "device.pair.resolved") { void this.runPageTask((pageState) => loadDevices(pageState, { quiet: true })); } - if (event.event === "node.pair.requested" || event.event === "node.pair.resolved") { + if ( + event.event === "node.pair.requested" || + event.event === "node.pair.resolved" || + event.event === "node.runnerInventory.changed" + ) { void this.runPageTask((pageState) => loadNodes(pageState, { quiet: true })); } }), diff --git a/ui/src/pages/devices/view-inventory.ts b/ui/src/pages/devices/view-inventory.ts index fb0ab0cd0d4d..6211a13dc0e3 100644 --- a/ui/src/pages/devices/view-inventory.ts +++ b/ui/src/pages/devices/view-inventory.ts @@ -192,6 +192,13 @@ function entryWarnStatuses( `, ); } + if (entry.node?.workerBundle?.status === "missing") { + statuses.push( + html` + ${renderSettingsStatus({ kind: "warn", label: t("devices.inventory.workerMissing") })} + `, + ); + } if (isApprovedNode && !entry.connected && isWindowsPlatform(entry.platform)) { statuses.push( html` @@ -225,6 +232,9 @@ function entryMetaLine(entry: DeviceInventoryEntry): string { if (entry.version) { parts.push(entry.version); } + if (entry.node?.workerBundle?.status === "installed") { + parts.push(t("devices.inventory.workerVersion", { version: entry.node.workerBundle.version })); + } if (entry.connected && entry.presence?.lastInputSeconds != null) { parts.push(formatInputRecency(entry.presence.lastInputSeconds)); } else if (!entry.connected && entry.lastSeenAtMs) { @@ -295,7 +305,7 @@ function renderInventoryEntry(entry: DeviceInventoryEntry, props: DevicesProps) ? entry.node.pendingRequestId : undefined; const connectionStatus = entry.connected - ? renderSettingsStatus({ kind: "ok", label: t("devices.inventory.connected") }) + ? nothing : renderSettingsStatus({ kind: "muted", label: t("devices.inventory.offline") }); return html`
@@ -372,7 +382,6 @@ function renderPresenceRow( : nothing}
- ${renderSettingsStatus({ kind: "ok", label: t("devices.inventory.connected") })} ${gateway ? renderSettingsStatus({ kind: "accent", label: t("devices.inventory.gateway") }) : renderSettingsStatus({ kind: "muted", label: t("devices.inventory.unpaired") })} diff --git a/ui/src/pages/devices/view.devices.test.ts b/ui/src/pages/devices/view.devices.test.ts index 19f84c6608e9..63d7e461b73f 100644 --- a/ui/src/pages/devices/view.devices.test.ts +++ b/ui/src/pages/devices/view.devices.test.ts @@ -248,7 +248,7 @@ describe("devices inventory rendering", () => { const gatewayEntry = expectDefined(entries[0], "gateway inventory entry"); expect(statusesByText(gatewayEntry, "gateway")).toHaveLength(1); - expect(statusesByText(gatewayEntry, "connected")).toHaveLength(1); + expect(statusesByText(gatewayEntry, "connected")).toHaveLength(0); expect(gatewayEntry.textContent).toContain("gateway-host"); expect(gatewayEntry.textContent).toContain("Linux · 2026.7.11 · input 5s ago"); expect(gatewayEntry.querySelector("button")).toBeNull(); @@ -360,6 +360,45 @@ describe("devices inventory rendering", () => { expect(approvals).toEqual(["node-req-1"]); }); + it("keeps installed workers quiet and warns when the retained bundle is missing", () => { + const container = renderDevicesContainer({ + nodes: [ + { + nodeId: "node-installed", + displayName: "Installed Mac", + connected: true, + paired: true, + workerBundle: { status: "installed", version: "2026.8.9" }, + }, + { + nodeId: "node-missing", + displayName: "Missing Mac", + connected: true, + paired: true, + workerBundle: { status: "missing" }, + }, + ], + }); + const section = getInventorySection(container); + const rows = Array.from(section.querySelectorAll(".device-entry")); + const installed = rows.find((row) => row.textContent?.includes("Installed Mac")); + const missing = rows.find((row) => row.textContent?.includes("Missing Mac")); + + expect(installed?.querySelector(".settings-row__desc")?.textContent).toContain( + "Worker 2026.8.9", + ); + expect(installed ? statusesByText(installed, "connected") : []).toHaveLength(0); + expect(installed ? statusesByText(installed, "worker missing") : []).toHaveLength(0); + expect(missing ? statusesByText(missing, "worker missing") : []).toHaveLength(1); + expect( + Array.from(missing?.querySelectorAll("[title]") ?? []) + .find((element) => element.textContent?.trim() === "worker missing") + ?.getAttribute("title"), + ).toBe( + "The Gateway-managed worker bundle is missing. Start a new session on this device to reinstall it.", + ); + }); + it("shows device and Gateway version drift", () => { const container = renderDevicesContainer({ gatewayVersion: "2026.7.2", @@ -561,7 +600,7 @@ describe("devices inventory rendering", () => { ); expect(entry?.textContent).toContain("unpaired"); expect(entry?.textContent).toContain("macOS 26.5.2"); - expect(entry ? statusesByText(entry, "connected") : []).toHaveLength(1); + expect(entry ? statusesByText(entry, "connected") : []).toHaveLength(0); expect(entry?.querySelector("button")).toBeNull(); });