diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift index c0070f549a3c..4c29afe42f4d 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift @@ -7314,22 +7314,26 @@ public struct FailedSessionPlacement: Codable, Sendable { public struct SessionsDispatchParams: Codable, Sendable { public let key: String public let agentid: String? - public let profileid: String + public let profileid: String? + public let deviceid: String? public init( key: String, agentid: String? = nil, - profileid: String) + profileid: String? = nil, + deviceid: String? = nil) { self.key = key self.agentid = agentid self.profileid = profileid + self.deviceid = deviceid } private enum CodingKeys: String, CodingKey { case key case agentid = "agentId" case profileid = "profileId" + case deviceid = "deviceId" } } diff --git a/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json b/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json index 4f1b2767f6d4..3d2a400d6e49 100644 --- a/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json @@ -1 +1 @@ -{"contentHash":"bcb2d5e0214c73a0352443baf8cb707bb90c701853cefe61eefc1fdfa3e7fe92","entrypoint":"agent-harness-runtime","importSpecifier":"openclaw/plugin-sdk/agent-harness-runtime"} +{"contentHash":"db186d810ff456142338c36af49cf272d4d58063d341a9d2f93885acc1c0a168","entrypoint":"agent-harness-runtime","importSpecifier":"openclaw/plugin-sdk/agent-harness-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/agent-harness.json b/docs/.generated/plugin-sdk-api-baseline/agent-harness.json index 7a370bd526f9..849ceca4c967 100644 --- a/docs/.generated/plugin-sdk-api-baseline/agent-harness.json +++ b/docs/.generated/plugin-sdk-api-baseline/agent-harness.json @@ -1 +1 @@ -{"contentHash":"12d73f365af2349d0b2318ad303af1c24db36a6ddf68bc141b3713696a6daa43","entrypoint":"agent-harness","importSpecifier":"openclaw/plugin-sdk/agent-harness"} +{"contentHash":"30bff907731a9c485e5ebf857d2bccb22f64733d6db696f65630ed24ffb0ab9d","entrypoint":"agent-harness","importSpecifier":"openclaw/plugin-sdk/agent-harness"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-core.json b/docs/.generated/plugin-sdk-api-baseline/channel-core.json index 60a8fec7d9c0..605ed60d34d3 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-core.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-core.json @@ -1 +1 @@ -{"contentHash":"2dcd8c79a124e110e625371372bbd7410c66f256934f9003cd0c49826a785459","entrypoint":"channel-core","importSpecifier":"openclaw/plugin-sdk/channel-core"} +{"contentHash":"354bdde4f8741dc12786458920d1c1f251d6771f86e1b35455388fcffad7412e","entrypoint":"channel-core","importSpecifier":"openclaw/plugin-sdk/channel-core"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json b/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json index 6d61f5e72341..b404b5cb2caa 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json @@ -1 +1 @@ -{"contentHash":"d382ede401d9c5f6a3121630fa13ac34b1c2aa1425d7ae28659d47f29e5f6290","entrypoint":"channel-entry-contract","importSpecifier":"openclaw/plugin-sdk/channel-entry-contract"} +{"contentHash":"e758c281a2468ed7b6692d7aac41532aa392199205d34c21425885650fe5002a","entrypoint":"channel-entry-contract","importSpecifier":"openclaw/plugin-sdk/channel-entry-contract"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-message.json b/docs/.generated/plugin-sdk-api-baseline/channel-message.json index b3f0fd04af56..1d836262f76f 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-message.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-message.json @@ -1 +1 @@ -{"contentHash":"9cb2f743250d5b0d1469ea6344d7d30de4d2aa0b8c40881e9d0415e3d9912961","entrypoint":"channel-message","importSpecifier":"openclaw/plugin-sdk/channel-message"} +{"contentHash":"ffdbd5eee1c1d4c3e3a906e9ef11cac1b67594c3f28698fbb91c9b4e9cefef01","entrypoint":"channel-message","importSpecifier":"openclaw/plugin-sdk/channel-message"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json b/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json index e4b6a921560f..b13ce727b9f0 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json @@ -1 +1 @@ -{"contentHash":"e1f879c60c79ffee904e89c29c1a4f12f0c3282e69afc1bcb46b8b9d309f92a7","entrypoint":"channel-outbound","importSpecifier":"openclaw/plugin-sdk/channel-outbound"} +{"contentHash":"e32f1483f2008c9ba937a6190e632ed3d47c59f9403770cad5ec4079c66f6e67","entrypoint":"channel-outbound","importSpecifier":"openclaw/plugin-sdk/channel-outbound"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json b/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json index 9b465b102f2b..ae5e7c472a20 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json @@ -1 +1 @@ -{"contentHash":"7c674d21cfd9f48f10234d7ced1c3ae4589c9750e0f402d049865bb0459c03ba","entrypoint":"channel-plugin-common","importSpecifier":"openclaw/plugin-sdk/channel-plugin-common"} +{"contentHash":"4b024e2a19bf93fbe9c910d84d9960751b89c287e69f7d56fc85ccae4e39e3ba","entrypoint":"channel-plugin-common","importSpecifier":"openclaw/plugin-sdk/channel-plugin-common"} diff --git a/docs/.generated/plugin-sdk-api-baseline/core.json b/docs/.generated/plugin-sdk-api-baseline/core.json index db1c4b59849d..2123f1354917 100644 --- a/docs/.generated/plugin-sdk-api-baseline/core.json +++ b/docs/.generated/plugin-sdk-api-baseline/core.json @@ -1 +1 @@ -{"contentHash":"ecd20ca3c6c947bbf1d6867f9ac7ddc8da0ee6c0d51ed55f5e946d186484077f","entrypoint":"core","importSpecifier":"openclaw/plugin-sdk/core"} +{"contentHash":"a12c5a0cb965b393e90521aeb1ce23d9939d9468935a33d8c7c3e717f822d10a","entrypoint":"core","importSpecifier":"openclaw/plugin-sdk/core"} diff --git a/docs/.generated/plugin-sdk-api-baseline/discord.json b/docs/.generated/plugin-sdk-api-baseline/discord.json index c5b209c6feac..70a7d939b0ef 100644 --- a/docs/.generated/plugin-sdk-api-baseline/discord.json +++ b/docs/.generated/plugin-sdk-api-baseline/discord.json @@ -1 +1 @@ -{"contentHash":"cdf57e7f08eefd8ede2162297d4db0f1e30094dd9d9e5f0176bc8585d5372084","entrypoint":"discord","importSpecifier":"openclaw/plugin-sdk/discord"} +{"contentHash":"18eac1e06d9ac32918f30178967692e13a47d7272b30c4386d12b74366192a47","entrypoint":"discord","importSpecifier":"openclaw/plugin-sdk/discord"} diff --git a/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json b/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json index d0784bc4cfec..ad270f9e06da 100644 --- a/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json @@ -1 +1 @@ -{"contentHash":"9be585be7caffa3686f64e9d84ae3c51c2fe4cccbfaf598f33f3082e3ceba40e","entrypoint":"gateway-runtime","importSpecifier":"openclaw/plugin-sdk/gateway-runtime"} +{"contentHash":"d47afcf53c6c5d9ad1cc99772d09ae60f447824583ab66e0a521f8230fec7bf3","entrypoint":"gateway-runtime","importSpecifier":"openclaw/plugin-sdk/gateway-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json b/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json index c7cd524f3d99..aec0c0d5c2c7 100644 --- a/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json +++ b/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json @@ -1 +1 @@ -{"contentHash":"644de05b59eae4368a5aa846d88137e7e77e61c7c45ee1439770e544f33ce534","entrypoint":"inbound-reply-dispatch","importSpecifier":"openclaw/plugin-sdk/inbound-reply-dispatch"} +{"contentHash":"6949fac4c91bb514789686bf0808d4255c1c9af27e0b880c65be2f25e20a4ec1","entrypoint":"inbound-reply-dispatch","importSpecifier":"openclaw/plugin-sdk/inbound-reply-dispatch"} diff --git a/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json b/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json index 8b80d90a95f9..023d21b4ba1d 100644 --- a/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json @@ -1 +1 @@ -{"contentHash":"b13afbf796748191ddab99cbae60d40ec5cddc50b08b09ab8f9e683ebf7f3c34","entrypoint":"meeting-runtime","importSpecifier":"openclaw/plugin-sdk/meeting-runtime"} +{"contentHash":"57a1b7494090393e3c88294bf717a91ce894a059af4ef940cffe110ed4567a47","entrypoint":"meeting-runtime","importSpecifier":"openclaw/plugin-sdk/meeting-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json b/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json index fcdc91c49bc7..035c3d81ad07 100644 --- a/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json +++ b/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json @@ -1 +1 @@ -{"contentHash":"bb957b8916022522a8f9b8309e8dcea2f6cc03941ece8b1f6fcff569973f677d","entrypoint":"plugin-entry","importSpecifier":"openclaw/plugin-sdk/plugin-entry"} +{"contentHash":"606c0eefefe6c30667c337de3a7222bbd1f03ab2c87703445407aa48e651895a","entrypoint":"plugin-entry","importSpecifier":"openclaw/plugin-sdk/plugin-entry"} diff --git a/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json b/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json index d374870d01f7..8e2dd444e761 100644 --- a/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json @@ -1 +1 @@ -{"contentHash":"d7cf3b452540f716d0de28d8cfd39721577a69bac00f6df1ddc0e8223d4ffca3","entrypoint":"plugin-runtime","importSpecifier":"openclaw/plugin-sdk/plugin-runtime"} +{"contentHash":"ef25098b086e8959a75cd35962a000ab2e564aa4c57d0987cb991532e623b509","entrypoint":"plugin-runtime","importSpecifier":"openclaw/plugin-sdk/plugin-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json b/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json index aa57f14dc4cb..208436ebafc0 100644 --- a/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json @@ -1 +1 @@ -{"contentHash":"4ebeddf97cd8fee4eb9edeab6b6f05d3540bfa160a842c4c83993836deae6ba4","entrypoint":"provider-catalog-runtime","importSpecifier":"openclaw/plugin-sdk/provider-catalog-runtime"} +{"contentHash":"d2642457b2c68f9a0c7b4b46da3ad1f960b1358f235e07da83ec2b121c4897f3","entrypoint":"provider-catalog-runtime","importSpecifier":"openclaw/plugin-sdk/provider-catalog-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json b/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json index 5128a3cfd088..7e6ba611df2d 100644 --- a/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json +++ b/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json @@ -1 +1 @@ -{"contentHash":"cd431f3c3ab8d51ae78b12c5c008b819a8059f36003ead70e0dd660abe1461d1","entrypoint":"tool-plugin","importSpecifier":"openclaw/plugin-sdk/tool-plugin"} +{"contentHash":"56b152bbedcc3c35b9f8ab1e2d44d999931aa24f88b1b902948803f08c8899fe","entrypoint":"tool-plugin","importSpecifier":"openclaw/plugin-sdk/tool-plugin"} diff --git a/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json b/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json index 1ad68f4bdde7..0d529a68f5f2 100644 --- a/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json +++ b/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json @@ -1 +1 @@ -{"contentHash":"b94f65403518cb98f9d3ac7cd885cd5366c8621dff261196fe499d428bade3b4","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"} +{"contentHash":"1e5dfd14f428c5181b3f59a6d793d3688cefa04ed4fbde992fdfd7ee9c2bb84c","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"} diff --git a/docs/plan/runners.md b/docs/plan/runners.md index e1bd5270b417..d691aa65d9b8 100644 --- a/docs/plan/runners.md +++ b/docs/plan/runners.md @@ -24,7 +24,7 @@ advances a milestone. | 4 | Picker: grouping, placement, liveness, enrichment | in progress | #120804, #122531, #122635, #122774 | | F | Real-wire session boundary harness | landed | #121212 | | 5 | Public worker ingress path | landed | #122578, #122643 | -| 6 | Node worker provider (device runners) | in progress | #122683, #122829 | +| 6 | Node worker provider (device runners) | in progress | #122683, #122769, #122829 | | 7 | Bundle push consent + runner updates | not started | — | | 8 | Stop-and-continue moves | not started | — | | 9 | Deletions (ssh sandbox, openshell, exec-host clones, …) | not started | — | diff --git a/packages/gateway-protocol/src/schema/session-placement.test.ts b/packages/gateway-protocol/src/schema/session-placement.test.ts index 41b483617aea..0dc682d47d8e 100644 --- a/packages/gateway-protocol/src/schema/session-placement.test.ts +++ b/packages/gateway-protocol/src/schema/session-placement.test.ts @@ -42,7 +42,7 @@ const workerOwnedFields = { }; describe("session dispatch protocol schemas", () => { - it("accepts only the dedicated dispatch selector and configured profile", () => { + it("accepts exactly one profile or device dispatch target", () => { expect( validateSessionsDispatchParams({ key: "agent:main:dispatch", @@ -50,7 +50,20 @@ describe("session dispatch protocol schemas", () => { profileId: "development", }), ).toBe(true); + expect( + validateSessionsDispatchParams({ + key: "agent:main:dispatch", + deviceId: "device-1", + }), + ).toBe(true); expect(validateSessionsDispatchParams({ key: "agent:main:dispatch" })).toBe(false); + expect( + validateSessionsDispatchParams({ + key: "agent:main:dispatch", + profileId: "development", + deviceId: "device-1", + }), + ).toBe(false); expect( validateSessionsDispatchParams({ key: "agent:main:dispatch", diff --git a/packages/gateway-protocol/src/schema/session-placement.ts b/packages/gateway-protocol/src/schema/session-placement.ts index a0a027af6ed2..520b52810bc7 100644 --- a/packages/gateway-protocol/src/schema/session-placement.ts +++ b/packages/gateway-protocol/src/schema/session-placement.ts @@ -163,12 +163,22 @@ export const SessionPlacementSchema = Type.Union([ FailedSessionPlacementSchema, ]); -/** Requests one-way dispatch of an existing local session to a configured worker profile. */ -export const SessionsDispatchParamsSchema = closedObject({ - key: NonEmptyString, - agentId: Type.Optional(NonEmptyString), - profileId: NonEmptyString, -}); +/** Requests one-way dispatch of an existing local session to exactly one worker target. */ +export const SessionsDispatchParamsSchema = Type.Object( + { + key: NonEmptyString, + agentId: Type.Optional(NonEmptyString), + profileId: Type.Optional(NonEmptyString), + deviceId: Type.Optional(NonEmptyString), + }, + { + additionalProperties: false, + oneOf: [ + { required: ["profileId"], not: { required: ["deviceId"] } }, + { required: ["deviceId"], not: { required: ["profileId"] } }, + ], + }, +); /** Result returned once session dispatch reaches durable worker ownership. */ export const SessionsDispatchResultSchema = closedObject({ diff --git a/src/gateway/server-lifecycle.ts b/src/gateway/server-lifecycle.ts index bf9ad3a09690..eccd1fe758b3 100644 --- a/src/gateway/server-lifecycle.ts +++ b/src/gateway/server-lifecycle.ts @@ -102,6 +102,7 @@ export async function prepareGatewayLifecycle(params: { residentRegistry, desktopSessionRegistry, nodeDesktopStreamBroker, + bindDeviceNodeRegistry, } = runtime; const completeControlUiDeviceAuthMigrationForEffectiveOperator = ( device: EffectiveOperatorDeviceIdentity, @@ -199,6 +200,7 @@ export async function prepareGatewayLifecycle(params: { }) : undefined; nodeDesktopServiceRef.current = nodeDesktopService; + bindDeviceNodeRegistry?.(nodeRegistry); const { createWatchNodeHttpRuntime } = await import("./watch-node-http.js"); const watchNodeHttpRuntime = createWatchNodeHttpRuntime({ nodeRegistry, diff --git a/src/gateway/server-methods/sessions-dispatch.ts b/src/gateway/server-methods/sessions-dispatch.ts index 05a6bea9bdae..ba6796b1ef7a 100644 --- a/src/gateway/server-methods/sessions-dispatch.ts +++ b/src/gateway/server-methods/sessions-dispatch.ts @@ -10,6 +10,7 @@ import { managedWorktrees } from "../../agents/worktrees/service.js"; import { formatErrorMessage } from "../../infra/errors.js"; import { resolveRequestedSessionAgentId as resolveRequestedGlobalAgentId } from "../session-request-agent.js"; import { SessionMutationAuthorizationChangedError } from "../session-sharing.js"; +import { DEVICE_WORKER_PROVIDER_ID } from "../worker-environments/device-provider.js"; import { projectWorkerSessionPlacement } from "../worker-environments/placement-projector.js"; import type { WorkerSessionPlacementRecord } from "../worker-environments/placement-record.js"; import { @@ -38,10 +39,11 @@ function respondInvalidWorkerSession(respond: RespondFn, message: string): void respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, message)); } -function resolveWorkerSessionTarget(params: { +async function resolveWorkerSessionTarget(params: { key: string; agentId?: string; profileId?: string; + deviceId?: string; context: GatewayRequestContext; respond: RespondFn; }) { @@ -51,16 +53,52 @@ function resolveWorkerSessionTarget(params: { params.respond(false, undefined, requestedAgent.error); return undefined; } - if ( - params.profileId !== undefined && - !Object.hasOwn(cfg.cloudWorkers?.profiles ?? {}, params.profileId) - ) { + const profileId = normalizeOptionalString(params.profileId); + const deviceId = normalizeOptionalString(params.deviceId); + let dispatchTarget: + | { + profileId: string; + deviceId?: undefined; + inheritedProfile?: undefined; + } + | { + profileId: string; + deviceId: string; + inheritedProfile: { + providerId: typeof DEVICE_WORKER_PROVIDER_ID; + profileSnapshot: { install: "bundle"; settings: { device: string } }; + }; + } + | undefined; + if (profileId && !Object.hasOwn(cfg.cloudWorkers?.profiles ?? {}, profileId)) { respondInvalidWorkerSession( params.respond, - `cloud worker profile is not configured: ${params.profileId}`, + `cloud worker profile is not configured: ${profileId}`, ); return undefined; } + if (profileId) { + dispatchTarget = { profileId }; + } else if (deviceId) { + const node = (await params.context.nodeRegistry.listCurrentConnected()).find( + (candidate) => candidate.nodeId === deviceId && candidate.commands.includes("system.run"), + ); + if (!node) { + respondInvalidWorkerSession( + params.respond, + `device is not a connected session-capable paired node: ${deviceId}`, + ); + return undefined; + } + dispatchTarget = { + profileId: `device:${deviceId}`, + deviceId, + inheritedProfile: { + providerId: DEVICE_WORKER_PROVIDER_ID, + profileSnapshot: { install: "bundle", settings: { device: deviceId } }, + }, + }; + } const target = loadAccessorSessionEntryForGatewayTarget({ key: params.key, cfg, @@ -72,7 +110,7 @@ function resolveWorkerSessionTarget(params: { respondInvalidWorkerSession(params.respond, `session not found: ${params.key}`); return undefined; } - return { cfg, target, entry, sessionId }; + return { cfg, target, entry, sessionId, dispatchTarget }; } function hasManagedSessionWorktree(params: { @@ -145,17 +183,22 @@ export const sessionDispatchHandlers: GatewayRequestHandlers = { respondInvalidWorkerSession(respond, "cloud worker dispatch is not configured"); return; } - const resolved = resolveWorkerSessionTarget({ + const resolved = await resolveWorkerSessionTarget({ key, agentId: params.agentId, profileId: params.profileId, + deviceId: params.deviceId, context, respond, }); if (!resolved) { return; } - const { cfg, target, entry, sessionId } = resolved; + const { cfg, target, entry, sessionId, dispatchTarget } = resolved; + if (!dispatchTarget) { + respondInvalidWorkerSession(respond, "worker dispatch target is missing"); + return; + } if (entry.archivedAt !== undefined) { respondInvalidWorkerSession(respond, "cannot dispatch an archived session"); return; @@ -218,7 +261,7 @@ export const sessionDispatchHandlers: GatewayRequestHandlers = { sessionId, sessionKey: target.canonicalKey, agentId: target.target.agentId, - profileId: params.profileId, + ...dispatchTarget, }, () => emitSessionsChanged(context, { @@ -245,7 +288,7 @@ export const sessionDispatchHandlers: GatewayRequestHandlers = { respondInvalidWorkerSession(respond, "cloud worker stop is not configured"); return; } - const resolved = resolveWorkerSessionTarget({ + const resolved = await resolveWorkerSessionTarget({ key, agentId: params.agentId, context, diff --git a/src/gateway/server-methods/sessions.dispatch.test.ts b/src/gateway/server-methods/sessions.dispatch.test.ts index af76f31e8e6a..7e0ae3ede438 100644 --- a/src/gateway/server-methods/sessions.dispatch.test.ts +++ b/src/gateway/server-methods/sessions.dispatch.test.ts @@ -101,14 +101,17 @@ function makeContext(overrides: Partial = {}): GatewayReq } as unknown as GatewayRequestContext; } -async function invoke(context: GatewayRequestContext) { +async function invoke( + context: GatewayRequestContext, + target: { profileId: string } | { deviceId: string } = { profileId: "test" }, +) { const respond = vi.fn() as unknown as RespondFn; await expectDefined( sessionDispatchHandlers["sessions.dispatch"], 'sessionDispatchHandlers["sessions.dispatch"] test invariant', )({ req: { id: "dispatch-request" } as never, - params: { key: sessionKey, profileId: "test" }, + params: { key: sessionKey, ...target }, respond, context, client: null, @@ -149,6 +152,81 @@ describe("sessions.dispatch", () => { ); }); + it("synthesizes the core device-provider target for a connected session-capable node", async () => { + mocks.resolveTarget.mockReturnValue( + targetWithEntry({ + sessionId, + worktree: { id: "worktree-1", branch: "openclaw/device-test", repoRoot: "/repo" }, + }), + ); + mocks.findLiveByOwner.mockReturnValue({ + id: "worktree-1", + ownerKind: "session", + ownerId: sessionKey, + }); + const dispatch = vi.fn().mockRejectedValue( + Object.assign(new Error("device-runner-transport-unimplemented: launch is pending"), { + code: "device-runner-transport-unimplemented", + }), + ); + const respond = await invoke( + makeContext({ + nodeRegistry: { + listCurrentConnected: vi.fn(async () => [ + { nodeId: "device-1", commands: ["system.run"] }, + ]), + } as never, + workerPlacementDispatchService: { dispatch }, + workerSessionPlacementService: { getMany: () => new Map() }, + }), + { deviceId: "device-1" }, + ); + + expect(dispatch).toHaveBeenCalledWith( + expect.objectContaining({ + profileId: "device:device-1", + deviceId: "device-1", + inheritedProfile: { + providerId: "device", + profileSnapshot: { install: "bundle", settings: { device: "device-1" } }, + }, + }), + expect.any(Function), + ); + expect(respond).toHaveBeenCalledWith( + false, + undefined, + expect.objectContaining({ + code: ErrorCodes.UNAVAILABLE, + message: expect.stringContaining("device-runner-transport-unimplemented"), + }), + ); + }); + + it("rejects a device target without a connected session-capable pairing", async () => { + const dispatch = vi.fn(); + const respond = await invoke( + makeContext({ + nodeRegistry: { + listCurrentConnected: vi.fn(async () => [{ nodeId: "device-1", commands: ["camera"] }]), + } as never, + workerPlacementDispatchService: { dispatch }, + workerSessionPlacementService: { getMany: () => new Map() }, + }), + { deviceId: "device-1" }, + ); + + expect(dispatch).not.toHaveBeenCalled(); + expect(respond).toHaveBeenCalledWith( + false, + undefined, + expect.objectContaining({ + code: ErrorCodes.INVALID_REQUEST, + message: expect.stringContaining("connected session-capable paired node"), + }), + ); + }); + it("rejects a missing session before dispatch", async () => { const dispatch = vi.fn(); const respond = await invoke( diff --git a/src/gateway/server-runtime-state-prepare.ts b/src/gateway/server-runtime-state-prepare.ts index 5d6283c1d463..838215660200 100644 --- a/src/gateway/server-runtime-state-prepare.ts +++ b/src/gateway/server-runtime-state-prepare.ts @@ -109,13 +109,9 @@ export async function prepareGatewayKernelState(params: { registry: pluginBootstrap.pluginRegistry, baseGatewayMethods: pluginBootstrap.baseGatewayMethods, }; - // Unconfigured clean installs get no service; durable rows still need list/status projection. - const hasConfiguredWorkerProfiles = - Object.keys(gatewayPluginConfigAtStart.cloudWorkers?.profiles ?? {}).length > 0; - const shouldStartWorkerEnvironmentService = - hasConfiguredWorkerProfiles || - Boolean(workerEnvironmentStartup?.records.length) || - Boolean(workerEnvironmentStartup?.hasNonlocalPlacementRecords); + // The core device provider is configuration-free, so every full Gateway owns the + // worker service even when no plugin-backed cloud profile has been configured. + const shouldStartWorkerEnvironmentService = Boolean(workerEnvironmentStartup); const hostDesktopConfig = gatewayPluginConfigAtStart.desktop?.host; const hostDesktopEnabled = hostDesktopConfig?.enabled === true; const nodeCommandConfig = gatewayPluginConfigAtStart.gateway?.nodes?.commands; @@ -166,7 +162,8 @@ export async function prepareGatewayKernelState(params: { }); }) : {}; - const { workerEnvironmentService, workerLiveEvents } = workerEnvironmentRuntime; + const { workerEnvironmentService, workerLiveEvents, bindDeviceNodeRegistry } = + workerEnvironmentRuntime; // Assigned once approval managers exist; placement dispatch must not run before then. const workerDispatchAuthority = { revoke: (_params: { sessionId: string; sessionKeys: readonly string[] }): void => { @@ -180,7 +177,7 @@ export async function prepareGatewayKernelState(params: { return placementModule.createGatewayWorkerPlacementRuntime({ placements: workerEnvironmentStartup.placementStore, environments: workerEnvironmentService, - admitNewPlacements: hasConfiguredWorkerProfiles, + admitNewPlacements: true, revokeSessionAuthority: (request) => workerDispatchAuthority.revoke(request), warn: (message) => log.warn(message), }); @@ -191,11 +188,8 @@ export async function prepareGatewayKernelState(params: { workerPlacementRuntime.dispatchService.dispatch, ); } - // Without configured profiles, existing placements still reconcile but new dispatches stay off. const workerPlacementControlAvailable = workerPlacementRuntime?.dispatchService; - const workerPlacementDispatchAvailable = hasConfiguredWorkerProfiles - ? workerPlacementControlAvailable - : undefined; + const workerPlacementDispatchAvailable = workerPlacementControlAvailable; const workerDesktopObserveAvailable = Boolean(workerEnvironmentService) && gatewayPluginConfigAtStart.cloudWorkers?.desktop === true; const desktopObserveAvailable = @@ -482,9 +476,9 @@ export async function prepareGatewayKernelState(params: { return { ...bootstrap, pluginRuntime, - hasConfiguredWorkerProfiles, workerEnvironmentService, workerLiveEvents, + bindDeviceNodeRegistry, workerDispatchAuthority, workerPlacementRuntime, workerPlacementControlAvailable, diff --git a/src/gateway/server-worker-environment-startup.ts b/src/gateway/server-worker-environment-startup.ts index b3033216c7ed..98481aba3de0 100644 --- a/src/gateway/server-worker-environment-startup.ts +++ b/src/gateway/server-worker-environment-startup.ts @@ -1,5 +1,6 @@ import { uniqueStrings } from "@openclaw/normalization-core/string-normalization"; import { getRuntimeConfig } from "../config/config.js"; +import { getPairedDevice } from "../infra/device-pairing.js"; import type { PluginRegistry } from "../plugins/registry-types.js"; import { getActiveSecretsRuntimeConfigSnapshot, @@ -7,7 +8,12 @@ import { } from "../secrets/runtime-state.js"; import { createLazyRuntimeModule } from "../shared/lazy-runtime.js"; import type { DesktopSessionRegistry } from "./desktop/session-registry.js"; +import type { NodeRegistry } from "./node-registry.js"; import type { WorkerBundleProducer, WorkerNpmArtifact } from "./worker-environments/bundle.js"; +import { + createDeviceWorkerProvider, + DEVICE_WORKER_PROVIDER_ID, +} from "./worker-environments/device-provider.js"; import type { WorkerLiveEventReceiver } from "./worker-environments/live-events.js"; import type { WorkerSessionPlacementStore } from "./worker-environments/placement-store.js"; import type { WorkerPlacementDispatchContract } from "./worker-environments/service-contract.js"; @@ -37,6 +43,7 @@ export type GatewayWorkerEnvironmentRuntime = { workerLiveEvents?: WorkerLiveEventReceiver; workerTunnelManager?: WorkerTunnelManager; bindWorkerSessionDispatch?: (dispatch: WorkerPlacementDispatchContract["dispatch"]) => void; + bindDeviceNodeRegistry?: (nodeRegistry: Pick) => void; }; const loadWorkerEnvironmentRuntimeModule = createLazyRuntimeModule( @@ -59,11 +66,18 @@ export async function loadGatewayWorkerEnvironmentStartupState(): Promise record.state === "destroyed" || record.state === "failed" || record.state === "orphaned" ? [] - : [record.providerId], + : record.providerId === DEVICE_WORKER_PROVIDER_ID + ? [] + : [record.providerId], ), ); const listDurableProviderIds = () => - uniqueStrings(store.listForReconcile().map((record) => record.providerId)); + uniqueStrings( + store + .listForReconcile() + .filter((record) => record.providerId !== DEVICE_WORKER_PROVIDER_ID) + .map((record) => record.providerId), + ); return { durableProviderIds, listDurableProviderIds, @@ -82,6 +96,11 @@ export async function createGatewayWorkerEnvironmentRuntime(params: { startup: GatewayWorkerEnvironmentStartupState; log: WorkerEnvironmentLogger; }): Promise { + let deviceNodeRegistry: Pick | undefined; + const deviceProvider = createDeviceWorkerProvider({ + getPairedDevice, + listConnectedNodes: async () => (await deviceNodeRegistry?.listCurrentConnected()) ?? [], + }); const [ { createWorkerEnvironmentService }, { createWorkerLiveEventReceiver }, @@ -159,7 +178,10 @@ export async function createGatewayWorkerEnvironmentRuntime(params: { store: params.startup.store, getConfig: getRuntimeConfig, // Plugin reload replaces the registry object; resolve against the live binding. - resolveProvider: (providerId) => resolveWorkerProvider(params.getPluginRegistry(), providerId), + resolveProvider: (providerId) => + providerId === DEVICE_WORKER_PROVIDER_ID + ? deviceProvider + : resolveWorkerProvider(params.getPluginRegistry(), providerId), prepareInstallation, tunnelManager: workerTunnelManager, resolveWorkerGateway: params.resolveWorkerGateway, @@ -221,5 +243,8 @@ export async function createGatewayWorkerEnvironmentRuntime(params: { bindWorkerSessionDispatch: (dispatch) => { dispatchChild = dispatch; }, + bindDeviceNodeRegistry: (nodeRegistry) => { + deviceNodeRegistry = nodeRegistry; + }, }; } diff --git a/src/gateway/server-worker-placement-startup.ts b/src/gateway/server-worker-placement-startup.ts index 215061d7e658..ae369c5e0407 100644 --- a/src/gateway/server-worker-placement-startup.ts +++ b/src/gateway/server-worker-placement-startup.ts @@ -199,6 +199,7 @@ export function coordinateWorkerPlacementDispatch( inFlight.request.sessionKey !== request.sessionKey || inFlight.request.agentId !== request.agentId || inFlight.request.profileId !== request.profileId || + inFlight.request.deviceId !== request.deviceId || !isDeepStrictEqual(inFlight.request.inheritedProfile, request.inheritedProfile) ) { throw new Error(`Session ${request.sessionKey} is already dispatching another request`); diff --git a/src/gateway/server.worker-placement-startup-context.test.ts b/src/gateway/server.worker-placement-startup-context.test.ts index dfeb083abb78..2985a554b179 100644 --- a/src/gateway/server.worker-placement-startup-context.test.ts +++ b/src/gateway/server.worker-placement-startup-context.test.ts @@ -15,15 +15,16 @@ afterEach(async () => { }); test( - "profiles-disabled startup publishes lightweight placement ownership to real session RPCs", + "profiles-disabled startup publishes core worker placement ownership to real session RPCs", { timeout: 30_000 }, async () => { // The shared server harness defaults to its minimal mode, which deliberately skips all - // worker stores. Exercise the production startup path while keeping profiles unconfigured. + // worker stores. Exercise the production startup path while keeping plugin profiles unconfigured; + // the core device provider still owns the worker service. process.env.OPENCLAW_TEST_MINIMAL_GATEWAY = "0"; harness = await startGatewayServerHarness(); const context = getFallbackGatewayContext(); - expect(context?.workerEnvironmentService).toBeUndefined(); + expect(context?.workerEnvironmentService).toBeDefined(); const placements = context?.workerSessionPlacementService as | WorkerSessionPlacementStore | undefined; @@ -99,7 +100,7 @@ test( lifecycleRevision: resetLifecycleRevision, }); expect(placements.get(resetSessionId)).toBeUndefined(); - expect(getFallbackGatewayContext()?.workerEnvironmentService).toBeUndefined(); + expect(getFallbackGatewayContext()?.workerEnvironmentService).toBeDefined(); ws.close(); }, ); diff --git a/src/gateway/worker-environments/device-provider.test.ts b/src/gateway/worker-environments/device-provider.test.ts new file mode 100644 index 000000000000..bb628f05a58d --- /dev/null +++ b/src/gateway/worker-environments/device-provider.test.ts @@ -0,0 +1,88 @@ +import { describe, expect, it } from "vitest"; +import type { PairedDevice } from "../../infra/device-pairing.types.js"; +import { WorkerProviderError } from "../../plugins/types.js"; +import { createDeviceWorkerProvider } from "./device-provider.js"; + +const DEVICE_ID = "device-session-host"; + +function pairedDevice(deviceId = DEVICE_ID): PairedDevice { + return { + deviceId, + publicKey: `public-key-${deviceId}`, + role: "node", + roles: ["node"], + tokens: { + node: { + token: "fixture-token", + role: "node", + scopes: [], + createdAtMs: 1, + }, + }, + createdAtMs: 1, + approvedAtMs: 1, + }; +} + +describe("device worker provider", () => { + it("provisions deterministic node leases only for connected paired session hosts", async () => { + const provider = createDeviceWorkerProvider({ + getPairedDevice: async (deviceId) => pairedDevice(deviceId), + listConnectedNodes: async () => [{ nodeId: DEVICE_ID, commands: ["system.run"] }], + }); + + const first = await provider.provision({ device: DEVICE_ID }, "operation-1"); + const repeated = await provider.provision({ device: DEVICE_ID }, "operation-1"); + const next = await provider.provision({ device: DEVICE_ID }, "operation-2"); + + expect(first).toEqual({ + leaseId: expect.stringMatching(/^device:[a-f0-9]{64}:[a-f0-9]{32}$/u), + node: { deviceId: DEVICE_ID }, + sharedHost: true, + }); + expect(repeated.leaseId).toBe(first.leaseId); + expect(next.leaseId).not.toBe(first.leaseId); + }); + + it.each([ + { + name: "missing pairing", + getPairedDevice: async () => null, + listConnectedNodes: async () => [{ nodeId: DEVICE_ID, commands: ["system.run"] }], + }, + { + name: "offline device", + getPairedDevice: async () => pairedDevice(), + listConnectedNodes: async () => [], + }, + { + name: "connected node without session execution", + getPairedDevice: async () => pairedDevice(), + listConnectedNodes: async () => [{ nodeId: DEVICE_ID, commands: [] }], + }, + ])("rejects $name during provision", async ({ getPairedDevice, listConnectedNodes }) => { + const provider = createDeviceWorkerProvider({ getPairedDevice, listConnectedNodes }); + + await expect(provider.provision({ device: DEVICE_ID }, "operation")).rejects.toBeInstanceOf( + WorkerProviderError, + ); + }); + + it("reports active, dormant, and unknown from pairing plus live presence", async () => { + let paired: PairedDevice | null = pairedDevice(); + let connected = true; + const provider = createDeviceWorkerProvider({ + getPairedDevice: async () => paired, + listConnectedNodes: async () => + connected ? [{ nodeId: DEVICE_ID, commands: ["system.run"] }] : [], + }); + const lease = { leaseId: "device-lease", profile: { device: DEVICE_ID } }; + + await expect(provider.inspect(lease)).resolves.toEqual({ status: "active", sharedHost: true }); + connected = false; + await expect(provider.inspect(lease)).resolves.toEqual({ status: "dormant" }); + paired = null; + await expect(provider.inspect(lease)).resolves.toEqual({ status: "unknown" }); + await expect(provider.destroy(lease)).resolves.toBeUndefined(); + }); +}); diff --git a/src/gateway/worker-environments/device-provider.ts b/src/gateway/worker-environments/device-provider.ts new file mode 100644 index 000000000000..25a7de6db88b --- /dev/null +++ b/src/gateway/worker-environments/device-provider.ts @@ -0,0 +1,82 @@ +import { createHash } from "node:crypto"; +import { hasEffectivePairedDeviceRole } from "../../infra/device-pairing.js"; +import type { PairedDevice } from "../../infra/device-pairing.types.js"; +import { + WorkerProviderError, + type WorkerProfile, + type WorkerProvider, +} from "../../plugins/types.js"; + +export const DEVICE_WORKER_PROVIDER_ID = "device"; + +type DeviceWorkerNode = { + nodeId: string; + commands: readonly string[]; +}; + +type DeviceWorkerProviderOptions = { + getPairedDevice: (deviceId: string) => Promise; + listConnectedNodes: () => Promise; +}; + +function requireDeviceId(profile: WorkerProfile): string { + const deviceId = profile.device; + if (typeof deviceId !== "string" || !deviceId.trim()) { + throw new WorkerProviderError("device worker profile requires a device setting"); + } + return deviceId.trim(); +} + +function isSessionCapableNode(node: DeviceWorkerNode): boolean { + return node.commands.includes("system.run"); +} + +function hasPairedNodeRole(device: PairedDevice | null): device is PairedDevice { + return Boolean(device && hasEffectivePairedDeviceRole(device, "node")); +} + +function deviceLeaseId(deviceId: string, operationId: string): string { + const deviceHash = createHash("sha256").update(deviceId).digest("hex"); + const operationHash = createHash("sha256").update(operationId).digest("hex"); + return `device:${deviceHash}:${operationHash.slice(0, 32)}`; +} + +/** Core provider for already-paired node hosts; pairing remains the durable trust owner. */ +export function createDeviceWorkerProvider(options: DeviceWorkerProviderOptions): WorkerProvider { + const findConnectedNode = async (deviceId: string) => + (await options.listConnectedNodes()).find( + (node) => node.nodeId === deviceId && isSessionCapableNode(node), + ); + + return { + id: DEVICE_WORKER_PROVIDER_ID, + provisionBeforeInstallation: true, + provision: async (profile, operationId) => { + const deviceId = requireDeviceId(profile); + const [paired, connected] = await Promise.all([ + options.getPairedDevice(deviceId), + findConnectedNode(deviceId), + ]); + if (!hasPairedNodeRole(paired) || !connected) { + throw new WorkerProviderError( + `device worker is not a connected session-capable paired node: ${deviceId}`, + ); + } + return { + leaseId: deviceLeaseId(deviceId, operationId), + node: { deviceId }, + sharedHost: true, + }; + }, + inspect: async ({ profile }) => { + const deviceId = requireDeviceId(profile); + const paired = await options.getPairedDevice(deviceId); + if (!hasPairedNodeRole(paired)) { + return { status: "unknown" }; + } + const connected = await findConnectedNode(deviceId); + return connected ? { status: "active", sharedHost: true } : { status: "dormant" }; + }, + destroy: async () => {}, + }; +} diff --git a/src/gateway/worker-environments/environment-access.test.ts b/src/gateway/worker-environments/environment-access.test.ts index 5c6a187f8611..c19715f67bf3 100644 --- a/src/gateway/worker-environments/environment-access.test.ts +++ b/src/gateway/worker-environments/environment-access.test.ts @@ -69,6 +69,33 @@ describe("worker environment service", () => { }); }); + it("rejects node tunnel startup with the typed milestone gate before SSH", async () => { + const tunnelManager = { + status: () => "stopped" as const, + start: vi.fn(), + stop: vi.fn(async () => {}), + stopAll: vi.fn(async () => {}), + } as unknown as WorkerTunnelManager; + const workerService = support.createService( + support.createProvider({ + provision: async () => ({ leaseId: "device-lease", node: { deviceId: "device-1" } }), + }), + { tunnelManager }, + ); + const environment = await workerService.create("development", "device-tunnel-gate"); + + await expect( + workerService.startTunnel({ + environmentId: environment.environmentId, + ownerEpoch: environment.ownerEpoch, + }), + ).rejects.toMatchObject({ + code: "device-runner-transport-unimplemented", + message: expect.stringContaining("device-runner-transport-unimplemented"), + } satisfies Partial); + expect(tunnelManager.start).not.toHaveBeenCalled(); + }); + it("reconciles shared-host isolation for a persisted lease before tunnel startup", async () => { support.seedReady("worker-legacy-shared"); support.testState.stateDb.db diff --git a/src/gateway/worker-environments/environment-access.ts b/src/gateway/worker-environments/environment-access.ts index e317b3b68f5f..7c5c5c81c13b 100644 --- a/src/gateway/worker-environments/environment-access.ts +++ b/src/gateway/worker-environments/environment-access.ts @@ -27,6 +27,7 @@ type WorkerEnvironmentAccessOptions = { serviceError: ( code: | "desktop_app_not_found" + | "device-runner-transport-unimplemented" | "environment_not_found" | "invalid_state" | "launcher_failure" @@ -87,12 +88,19 @@ export function createWorkerEnvironmentAccess(options: WorkerEnvironmentAccessOp if ( !inState(record, "ready", "idle", "attached") || record.destroyRequestedAtMs !== null || - !record.leaseId || - !record.sshEndpoint || - !record.bootstrapReceipt + !record.leaseId ) { throw serviceError("invalid_state", `Cannot start tunnel in state: ${record.state}`); } + if (!record.sshEndpoint) { + throw serviceError( + "device-runner-transport-unimplemented", + "device-runner-transport-unimplemented: device runner launch is not available in this build", + ); + } + if (!record.bootstrapReceipt) { + throw serviceError("invalid_state", `Cannot start tunnel in state: ${record.state}`); + } if (record.sharedHost === null) { throw serviceError( "provider_failure", @@ -269,19 +277,19 @@ export function createWorkerEnvironmentAccess(options: WorkerEnvironmentAccessOp `environment does not advertise desktop app: ${request.app}`, ); } - return { app, record }; + return { app, record, sshEndpoint: record.sshEndpoint }; }; let startup: Promise | undefined; let launchEpoch: number | undefined; await withLock(request.environmentId, async () => { - const { app, record } = requireLaunchable(); + const { app, record, sshEndpoint } = requireLaunchable(); const provider = providerFor(record.providerId); launchEpoch = record.ownerEpoch; startup = tunnels.desktop.launchApp({ environmentId: record.environmentId, ownerEpoch: record.ownerEpoch, - ssh: record.sshEndpoint, + ssh: sshEndpoint, app, resolveIdentity: identityResolverFor(record, provider, record.leaseId), }); diff --git a/src/gateway/worker-environments/placement-dispatch-device.test.ts b/src/gateway/worker-environments/placement-dispatch-device.test.ts new file mode 100644 index 000000000000..6fedee75a67e --- /dev/null +++ b/src/gateway/worker-environments/placement-dispatch-device.test.ts @@ -0,0 +1,76 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; +import { + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, + type OpenClawStateDatabase, +} from "../../state/openclaw-state-db.js"; +import { REQUEST, type PlacementStore } from "./placement-dispatch-test-fixtures.js"; +import { createHarness } from "./placement-dispatch-test-harness.js"; +import { createWorkerSessionPlacementStore } from "./placement-store.js"; + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + +describe("device worker placement dispatch", () => { + let root: string; + let database: OpenClawStateDatabase; + let placementStore: PlacementStore; + + beforeEach(() => { + root = tempDirs.make("openclaw-device-dispatch-"); + database = openOpenClawStateDatabase({ env: { OPENCLAW_STATE_DIR: root } }); + placementStore = createWorkerSessionPlacementStore({ database, now: () => 1_000 }); + }); + + afterEach(() => { + closeOpenClawStateDatabaseForTest(); + }); + + it("provisions the environment and surfaces the honest transport gate", async () => { + const harness = createHarness(placementStore); + const transportError = Object.assign( + new Error("device-runner-transport-unimplemented: launch is pending"), + { code: "device-runner-transport-unimplemented" }, + ); + vi.mocked(harness.environments.createFromProfileSnapshot).mockResolvedValue({ + ...harness.ready, + providerId: "device", + profileId: "device:device-1", + profileSnapshot: { install: "bundle", settings: { device: "device-1" } }, + leaseId: "device-lease-1", + sshEndpoint: null, + bootstrapReceipt: null, + sharedHost: true, + tunnelStatus: "stopped", + }); + vi.mocked(harness.environments.startTunnel).mockRejectedValue(transportError); + const request = { + ...REQUEST, + profileId: "device:device-1", + deviceId: "device-1", + inheritedProfile: { + providerId: "device", + profileSnapshot: { install: "bundle" as const, settings: { device: "device-1" } }, + }, + }; + + await expect(harness.service.dispatch(request)).rejects.toMatchObject({ + code: "device-runner-transport-unimplemented", + }); + + expect(harness.environments.createFromProfileSnapshot).toHaveBeenCalledWith( + { profileId: request.profileId, ...request.inheritedProfile }, + expect.stringMatching(/^session-dispatch:/u), + ); + expect(harness.environments.startTunnel).toHaveBeenCalledWith({ + environmentId: harness.ready.environmentId, + ownerEpoch: harness.ready.ownerEpoch, + }); + expect(harness.environments.attachSession).not.toHaveBeenCalled(); + expect(harness.environments.destroy).toHaveBeenCalledWith(harness.ready.environmentId); + expect(harness.placements.current()).toMatchObject({ + state: "failed", + recoveryError: expect.stringContaining("device-runner-transport-unimplemented"), + }); + }); +}); diff --git a/src/gateway/worker-environments/placement-dispatch.ts b/src/gateway/worker-environments/placement-dispatch.ts index ad5206924205..7a53606bd4ae 100644 --- a/src/gateway/worker-environments/placement-dispatch.ts +++ b/src/gateway/worker-environments/placement-dispatch.ts @@ -1,5 +1,6 @@ import { randomUUID } from "node:crypto"; import { supportsWorkerExecutionContextLaunch } from "./admission.js"; +import { DEVICE_WORKER_PROVIDER_ID } from "./device-provider.js"; import { createPlacementFailureActions, isUnavailableEnvironment, @@ -77,11 +78,30 @@ type WorkerPlacementDispatchOptions = { function requireProvisionedEnvironment( environment: Awaited>, expectedEnvironmentId: string, -): { environmentId: string; ownerEpoch: number; bundleHash: string } { +): + | { transport: "node"; environmentId: string; ownerEpoch: number } + | { transport: "ssh"; environmentId: string; ownerEpoch: number; bundleHash: string } { if ( (environment.state !== "ready" && environment.state !== "idle") || + environment.environmentId !== expectedEnvironmentId + ) { + throw new Error( + `Worker environment is not dispatchable with the current execution-context contract: ${environment.state}`, + ); + } + if ( + environment.providerId === DEVICE_WORKER_PROVIDER_ID && + !environment.sshEndpoint && + !environment.bootstrapReceipt + ) { + return { + transport: "node", + environmentId: environment.environmentId, + ownerEpoch: environment.ownerEpoch, + }; + } + if ( !environment.bootstrapReceipt || - environment.environmentId !== expectedEnvironmentId || !supportsWorkerExecutionContextLaunch(environment.bootstrapReceipt) ) { throw new Error( @@ -89,6 +109,7 @@ function requireProvisionedEnvironment( ); } return { + transport: "ssh", environmentId: environment.environmentId, ownerEpoch: environment.ownerEpoch, bundleHash: environment.bootstrapReceipt.bundleHash, @@ -179,6 +200,10 @@ export function createWorkerPlacementDispatchService(options: WorkerPlacementDis const provisioned = requireProvisionedEnvironment(environment, expectedEnvironmentId); environmentId = provisioned.environmentId; ownerEpoch = provisioned.ownerEpoch; + if (provisioned.transport === "node") { + await environments.startTunnel({ environmentId, ownerEpoch }); + throw new Error("Device worker transport unexpectedly started before launch support"); + } placement = placements.transition({ sessionId: request.sessionId, from: "provisioning", diff --git a/src/gateway/worker-environments/provider-lifecycle.ts b/src/gateway/worker-environments/provider-lifecycle.ts index 47d1b22d0121..5367e7bbf1fa 100644 --- a/src/gateway/worker-environments/provider-lifecycle.ts +++ b/src/gateway/worker-environments/provider-lifecycle.ts @@ -220,14 +220,16 @@ export function createWorkerProviderLifecycle(options: WorkerProviderLifecycleOp if (record.state !== "bootstrapping" || !record.leaseId || !record.sshEndpoint) { throw serviceError("invalid_state", "Worker bootstrap requires a provisioned SSH lease"); } + const leaseId = record.leaseId; + const sshEndpoint = record.sshEndpoint; let receipt: WorkerAdmissionHandshake; try { receipt = await callBootstrap((signal) => options.bootstrapWorker({ operationId: record.provisionOperationId, - sshEndpoint: record.sshEndpoint, + sshEndpoint, installation, - resolveIdentity: identityResolverFor(record, provider, record.leaseId), + resolveIdentity: identityResolverFor(record, provider, leaseId), signal, }), ); @@ -235,7 +237,7 @@ export function createWorkerProviderLifecycle(options: WorkerProviderLifecycleOp throw new Error("Worker bootstrap receipt does not match the expected build identity"); } } catch (error) { - return await failBootstrap(record, record.leaseId, provider, error); + return await failBootstrap(record, leaseId, provider, error); } const material = credentialMaterial(); // Receipt, owner epoch, and credential hash commit together. A failed write leaves the @@ -291,11 +293,16 @@ export function createWorkerProviderLifecycle(options: WorkerProviderLifecycleOp // A timeout can happen after allocation; retain the same operation id for safe replay. const patch = { leaseId: lease.leaseId, - sshEndpoint: lease.ssh, sharedHost: lease.sharedHost === true, desktop: lease.desktop ?? null, }; - const bootstrapping = move(record, "bootstrapping", patch); + if (lease.node) { + return move(record, "ready", { ...patch, sshEndpoint: null }); + } + const bootstrapping = move(record, "bootstrapping", { + ...patch, + sshEndpoint: lease.ssh, + }); if (record.destroyRequestedAtMs !== null) { return bootstrapping; } @@ -317,7 +324,11 @@ export function createWorkerProviderLifecycle(options: WorkerProviderLifecycleOp provider = providerFor(record.providerId), ) => { let installation: WorkerInstallationArtifact | undefined; - if (record.state === "requested" && record.destroyRequestedAtMs === null) { + if ( + record.state === "requested" && + record.destroyRequestedAtMs === null && + provider.provisionBeforeInstallation !== true + ) { try { // Fresh requests package before allocation. Once provisioning is durable, provider replay // must happen first because the previous response may have been lost after allocation. @@ -410,7 +421,7 @@ export function createWorkerProviderLifecycle(options: WorkerProviderLifecycleOp const leaseId = record.leaseId; if (!leaseId) { const provisioned = await resumeProvision(record, provider).catch(() => undefined); - if (provisioned?.state === "bootstrapping") { + if (provisioned?.leaseId && provisioned.destroyRequestedAtMs !== null) { await finishDestroy(provisioned, provider).catch(() => undefined); } return; @@ -456,6 +467,14 @@ export function createWorkerProviderLifecycle(options: WorkerProviderLifecycleOp move(draining, "orphaned", { lastError: ORPHANED_LEASE_ERROR }); return; } + if (status === "dormant") { + if (teardownExpected) { + await finishDestroy(record, provider).catch(() => undefined); + } + // A paired device may be offline without losing its lease. Keep that authoritative + // holding state out of the unknown/orphan path until pairing itself is removed. + return; + } const inspectedSharedHost = inspection.sharedHost === true; if (record.sharedHost !== null && record.sharedHost !== inspectedSharedHost) { // Workspace actions capture isolation at tunnel creation. Fence the old actions before @@ -472,6 +491,11 @@ export function createWorkerProviderLifecycle(options: WorkerProviderLifecycleOp await finishDestroy(record, provider).catch(() => undefined); return; } + if (!record.sshEndpoint) { + // Node leases deliberately have no SSH bootstrap path; their transport owner advances + // this lifecycle once supervised node launch is available. + return; + } if (record.state === "attached") { if ( currentBundle && diff --git a/src/gateway/worker-environments/provider-provisioning.test.ts b/src/gateway/worker-environments/provider-provisioning.test.ts index 64623d563170..02464d3e1c31 100644 --- a/src/gateway/worker-environments/provider-provisioning.test.ts +++ b/src/gateway/worker-environments/provider-provisioning.test.ts @@ -76,6 +76,36 @@ describe("worker environment service", () => { expect(workerService.takeMintedCredential(binding)).toBeUndefined(); }); + it("holds a node lease ready without entering SSH bootstrap", async () => { + support.testState.prepareInstallation = vi.fn(async () => { + throw new Error("node leases must not prepare an SSH installation"); + }); + const workerService = support.createService( + support.createProvider({ + provisionBeforeInstallation: true, + provision: async () => ({ + leaseId: "device-lease-1", + node: { deviceId: "device-1" }, + sharedHost: true, + }), + }), + ); + + const result = await workerService.create("development", "request-device"); + + expect(result).toMatchObject({ + state: "ready", + leaseId: "device-lease-1", + sshEndpoint: null, + bootstrapReceipt: null, + sharedHost: true, + ownerEpoch: 1, + }); + expect(support.testState.prepareInstallation).not.toHaveBeenCalled(); + expect(support.testState.bootstrapWorker).not.toHaveBeenCalled(); + expect(support.testState.store.getCredential(result.environmentId)).toBeUndefined(); + }); + it("creates a nested environment from its parent's snapshot after config drift", async () => { const provisionedProfiles: WorkerProfile[] = []; let lease = 0; @@ -699,6 +729,17 @@ describe("worker environment service", () => { it.each([ ["missing result", null, "invalid provision result"], + ["missing transport", { leaseId: "lease-invalid" }, "invalid provision result"], + [ + "ambiguous transport", + { leaseId: "lease-invalid", ssh: support.SSH_ENDPOINT, node: { deviceId: "device-1" } }, + "invalid provision result", + ], + [ + "blank node device id", + { leaseId: "lease-invalid", node: { deviceId: " " } }, + "invalid node device id", + ], [ "malformed SSH endpoint", { leaseId: "lease-invalid", ssh: { ...support.SSH_ENDPOINT, keyRef: "not-a-secret-ref" } }, diff --git a/src/gateway/worker-environments/provider-reconciliation.test.ts b/src/gateway/worker-environments/provider-reconciliation.test.ts index f546edbea6ab..e011b25c03bd 100644 --- a/src/gateway/worker-environments/provider-reconciliation.test.ts +++ b/src/gateway/worker-environments/provider-reconciliation.test.ts @@ -395,10 +395,33 @@ describe("worker environment service", () => { }); }); + it("keeps a dormant paired-device lease in its nonterminal holding state", async () => { + support.seedReady("worker-dormant"); + const destroy = vi.fn(async () => {}); + const tunnelManager = { + start: vi.fn(), + stop: vi.fn(async () => {}), + stopAll: vi.fn(async () => {}), + status: () => "stopped" as const, + } as unknown as WorkerTunnelManager; + const workerService = support.createService( + support.createProvider({ inspect: async () => ({ status: "dormant" }), destroy }), + { tunnelManager }, + ); + + await workerService.reconcileOnce(); + await workerService.reconcileOnce(); + + expect(support.testState.store.get("worker-dormant")).toMatchObject({ state: "ready" }); + expect(tunnelManager.stop).not.toHaveBeenCalled(); + expect(destroy).not.toHaveBeenCalled(); + }); + it.each([ null, { status: "future" }, { status: "active", sharedHost: "yes" }, + { status: "dormant", sharedHost: true }, { status: "unknown", sharedHost: true }, ])("retains retryable state for malformed inspection result %#", async (inspection) => { support.seedReady("worker-malformed"); diff --git a/src/gateway/worker-environments/service-contract.ts b/src/gateway/worker-environments/service-contract.ts index c7a4db4c17f0..a27f2494887b 100644 --- a/src/gateway/worker-environments/service-contract.ts +++ b/src/gateway/worker-environments/service-contract.ts @@ -73,6 +73,7 @@ export type WorkerPlacementDispatchRequest = { sessionKey: string; agentId: string; profileId: string; + deviceId?: string; inheritedProfile?: { providerId: string; profileSnapshot: WorkerProfile; diff --git a/src/gateway/worker-environments/service-validation.ts b/src/gateway/worker-environments/service-validation.ts index 1602b2d6b09b..bea89dc40458 100644 --- a/src/gateway/worker-environments/service-validation.ts +++ b/src/gateway/worker-environments/service-validation.ts @@ -12,7 +12,12 @@ export function requireWorkerLeaseStatus(value: unknown): WorkerLeaseStatus { throw new Error("Worker provider returned an invalid inspection result"); } const status = value.status; - if (status !== "active" && status !== "destroyed" && status !== "unknown") { + if ( + status !== "active" && + status !== "dormant" && + status !== "destroyed" && + status !== "unknown" + ) { throw new Error("Worker provider returned an invalid inspection status"); } if (status === "active") { @@ -28,21 +33,35 @@ export function requireWorkerLeaseStatus(value: unknown): WorkerLeaseStatus { } export function requireWorkerLease(value: unknown): WorkerLease { + const hasSsh = isRecord(value) && Object.hasOwn(value, "ssh"); + const hasNode = isRecord(value) && Object.hasOwn(value, "node"); if ( !isRecord(value) || typeof value.leaseId !== "string" || !value.leaseId.trim() || - !isRecord(value.ssh) || + hasSsh === hasNode || + (hasSsh && !isRecord(value.ssh)) || + (hasNode && !isRecord(value.node)) || (value.sharedHost !== undefined && typeof value.sharedHost !== "boolean") ) { throw new Error("Worker provider returned an invalid provision result"); } - return { + const common = { leaseId: value.leaseId.trim(), - ssh: normalizeWorkerSshEndpoint(value.ssh as WorkerSshEndpoint), ...(value.sharedHost === true ? { sharedHost: true } : {}), ...(value.desktop === undefined ? {} : { desktop: normalizeWorkerDesktopEndpoint(value.desktop as WorkerDesktopEndpoint) }), }; + if (hasSsh) { + return { + ...common, + ssh: normalizeWorkerSshEndpoint(value.ssh as WorkerSshEndpoint), + }; + } + const deviceId = (value.node as { deviceId?: unknown }).deviceId; + if (typeof deviceId !== "string" || !deviceId.trim()) { + throw new Error("Worker provider returned an invalid node device id"); + } + return { ...common, node: { deviceId: deviceId.trim() } }; } diff --git a/src/gateway/worker-environments/service.ts b/src/gateway/worker-environments/service.ts index 9bdcf7f0bdd1..15e0dbfc3e3f 100644 --- a/src/gateway/worker-environments/service.ts +++ b/src/gateway/worker-environments/service.ts @@ -49,6 +49,7 @@ type WorkerEnvironmentServiceErrorCode = | "invalid_profile" | "invalid_state" | "desktop_app_not_found" + | "device-runner-transport-unimplemented" | "unsupported_platform" | "launcher_failure" | "provider_failure" diff --git a/src/gateway/worker-environments/state.test.ts b/src/gateway/worker-environments/state.test.ts index 2929b4b1a133..87ae6654c255 100644 --- a/src/gateway/worker-environments/state.test.ts +++ b/src/gateway/worker-environments/state.test.ts @@ -7,7 +7,7 @@ import { const EXPECTED_TRANSITIONS: Record = { requested: ["provisioning", "failed"], - provisioning: ["bootstrapping", "failed"], + provisioning: ["bootstrapping", "ready", "failed"], bootstrapping: ["ready", "draining", "orphaned"], ready: ["bootstrapping", "attached", "idle", "draining", "orphaned"], attached: ["idle", "draining", "orphaned"], diff --git a/src/gateway/worker-environments/state.ts b/src/gateway/worker-environments/state.ts index 4ffaa58d5e16..56a6a208c5ee 100644 --- a/src/gateway/worker-environments/state.ts +++ b/src/gateway/worker-environments/state.ts @@ -9,7 +9,7 @@ export type WorkerEnvironmentLeasedState = Exclude< const TRANSITIONS = { requested: ["provisioning", "failed"], - provisioning: ["bootstrapping", "failed"], + provisioning: ["bootstrapping", "ready", "failed"], bootstrapping: ["ready", "draining", "orphaned"], ready: ["bootstrapping", "attached", "idle", "draining", "orphaned"], attached: ["idle", "draining", "orphaned"], diff --git a/src/gateway/worker-environments/store.test.ts b/src/gateway/worker-environments/store.test.ts index 2e3e9b7395d3..05d5998fdb51 100644 --- a/src/gateway/worker-environments/store.test.ts +++ b/src/gateway/worker-environments/store.test.ts @@ -638,6 +638,14 @@ describe("worker environment store", () => { patch: { leaseId: "lease-1" }, }), ).toThrow("requires an SSH endpoint reference"); + expect(() => + store.transition({ + environmentId: "worker-1", + from: "provisioning", + to: "ready", + patch: { leaseId: "lease-1", sshEndpoint: SSH_ENDPOINT }, + }), + ).toThrow("requires bootstrap proof or a node lease"); store.transition({ environmentId: "worker-1", @@ -662,6 +670,33 @@ describe("worker environment store", () => { ).toThrow("lease id is immutable"); }); + it("persists a ready node lease without validating SSH metadata", () => { + createIntent("worker-node", { settings: { device: "device-1" } }); + store.transition({ environmentId: "worker-node", from: "requested", to: "provisioning" }); + + const ready = store.transition({ + environmentId: "worker-node", + from: "provisioning", + to: "ready", + patch: { leaseId: "device-lease-1", sshEndpoint: null, sharedHost: true }, + }); + + expect(ready).toMatchObject({ + state: "ready", + leaseId: "device-lease-1", + sshEndpoint: null, + bootstrapReceipt: null, + sharedHost: true, + ownerEpoch: 1, + }); + expect(store.get("worker-node")).toEqual(ready); + expect( + database.db + .prepare("SELECT ssh_host, ssh_host_key FROM worker_environments WHERE environment_id = ?") + .get("worker-node"), + ).toEqual({ ssh_host: null, ssh_host_key: null }); + }); + it("enforces one credential-bound session and teardown fencing", () => { const bootstrapping = seedBootstrapping("worker-multi-session", "lease-multi-session"); const ready = readyPatch(); diff --git a/src/gateway/worker-environments/store.ts b/src/gateway/worker-environments/store.ts index 5c947d46e957..66b97428926b 100644 --- a/src/gateway/worker-environments/store.ts +++ b/src/gateway/worker-environments/store.ts @@ -63,7 +63,11 @@ type RecordBase = RecordIdentity & { }; type Ssh = WorkerEnvironmentSshEndpoint; type UnleasedRecord = { state: WorkerEnvironmentUnleasedState; leaseId: null; sshEndpoint: null }; -type LeasedRecord = { state: WorkerEnvironmentLeasedState; leaseId: string; sshEndpoint: Ssh }; +type LeasedRecord = { + state: WorkerEnvironmentLeasedState; + leaseId: string; + sshEndpoint: Ssh | null; +}; export type WorkerEnvironmentRecord = RecordBase & (UnleasedRecord | LeasedRecord); export class WorkerSessionAlreadyAttachedError extends Error { constructor( @@ -406,8 +410,8 @@ function assertShape( if (!leaseId) { throw new Error(`Worker environment state ${state} requires a provider lease`); } - if (!sshEndpoint) { - throw new Error("Worker environment provider lease requires an SSH endpoint reference"); + if (state === "bootstrapping" && !sshEndpoint) { + throw new Error("Worker environment bootstrap requires an SSH endpoint reference"); } } else if (leaseId || sshEndpoint || desktop) { throw new Error(`Worker environment state ${state} cannot retain a provider lease`); @@ -954,6 +958,11 @@ export function createWorkerEnvironmentStore( ? null : normalizeWorkerDesktopEndpoint(patch.desktop); const acceptsBootstrapReceipt = from === "bootstrapping" && to === "ready"; + const acceptsDeferredNodeReady = + from === "provisioning" && to === "ready" && sshEndpoint === null; + if (to === "ready" && !acceptsBootstrapReceipt && !acceptsDeferredNodeReady) { + throw new Error("Ready worker transition requires bootstrap proof or a node lease"); + } if (patch.bootstrapReceipt !== undefined && !acceptsBootstrapReceipt) { throw new Error("Bootstrap receipt can only be recorded when a worker becomes ready"); } @@ -1023,11 +1032,12 @@ export function createWorkerEnvironmentStore( to === "destroyed" || to === "failed" || to === "orphaned"); - const ownerEpoch = acceptsBootstrapReceipt - ? Math.max(1, current.ownerEpoch) - : acceptsAttachedCredential || ownerEndingTransition - ? nextGlobalOwnerEpoch(db) - : current.ownerEpoch; + const ownerEpoch = + acceptsBootstrapReceipt || acceptsDeferredNodeReady + ? Math.max(1, current.ownerEpoch) + : acceptsAttachedCredential || ownerEndingTransition + ? nextGlobalOwnerEpoch(db) + : current.ownerEpoch; updateRow(db, environmentId, from, { lease_id: leaseId, shared_host: sharedHost === null ? null : sharedHost ? 1 : 0, diff --git a/src/gateway/worker-environments/worker-turn-launcher.test.ts b/src/gateway/worker-environments/worker-turn-launcher.test.ts index bf45913cf622..36970e1aadf0 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.test.ts @@ -1,6 +1,5 @@ import { createHash } from "node:crypto"; import fs from "node:fs/promises"; -import os from "node:os"; import path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { @@ -48,6 +47,10 @@ import { openOpenClawStateDatabase, type OpenClawStateDatabase, } from "../../state/openclaw-state-db.js"; +import { + createOpenClawTestState, + type OpenClawTestState, +} from "../../test-utils/openclaw-test-state.js"; import { parseWorkerLaunchDescriptor, type WorkerLaunchDescriptor, @@ -135,6 +138,7 @@ describe("worker turn launcher", () => { }); let root: string; + let testState: OpenClawTestState; let database: OpenClawStateDatabase; let placements: WorkerSessionPlacementStore; let sessionFile: string; @@ -146,8 +150,12 @@ describe("worker turn launcher", () => { }; beforeEach(async () => { - root = await fs.mkdtemp(path.join(await fs.realpath(os.tmpdir()), "openclaw-worker-turn-")); - database = openOpenClawStateDatabase({ env: { OPENCLAW_STATE_DIR: root } }); + testState = await createOpenClawTestState({ + label: "worker-turn", + layout: "state-only", + }); + root = testState.root; + database = openOpenClawStateDatabase({ env: testState.env }); placements = createWorkerSessionPlacementStore({ database }); sessionTarget = { agentId: "main", @@ -168,7 +176,7 @@ describe("worker turn launcher", () => { cleanupAdmissionSink = undefined; closeOpenClawStateDatabaseForTest(); resetAgentEventsForTest(); - await fs.rm(root, { recursive: true, force: true }); + await testState.cleanup(); }); function createWorkerSessionTurnPlacementProvider( @@ -1369,7 +1377,9 @@ describe("worker turn launcher", () => { acquireTurnCredential: vi.fn(async () => credential()), acknowledgeCredentialDelivery, startTunnel: vi.fn(async () => { - throw new Error("tunnel unavailable"); + throw Object.assign(new Error("device-runner-transport-unimplemented: launch is pending"), { + code: "device-runner-transport-unimplemented", + }); }), stopTunnel, destroy, @@ -1388,7 +1398,7 @@ describe("worker turn launcher", () => { turn("run-tunnel-unavailable"), runLocal, ), - ).rejects.toThrow("tunnel unavailable"); + ).rejects.toMatchObject({ code: "device-runner-transport-unimplemented" }); expect(runLocal).not.toHaveBeenCalled(); expect(acknowledgeCredentialDelivery).not.toHaveBeenCalled(); diff --git a/src/plugins/capability-provider.types.ts b/src/plugins/capability-provider.types.ts index e246d8afa7b1..08d38e05caa5 100644 --- a/src/plugins/capability-provider.types.ts +++ b/src/plugins/capability-provider.types.ts @@ -102,11 +102,10 @@ export type WorkerDesktopEndpoint = { /** Durable lease identity and endpoint returned by a successful provision operation. */ export type WorkerLease = { leaseId: string; - ssh: WorkerSshEndpoint; /** The SSH account also owns processes unrelated to this worker lease. */ sharedHost?: boolean; desktop?: WorkerDesktopEndpoint; -}; +} & ({ ssh: WorkerSshEndpoint; node?: never } | { node: { deviceId: string }; ssh?: never }); /** Authoritative inspection result for an already-known worker lease. */ export type WorkerLeaseStatus = @@ -115,6 +114,7 @@ export type WorkerLeaseStatus = /** Explicit provider fact used to reconcile leases persisted before this metadata existed. */ sharedHost?: boolean; } + | { status: "dormant" } | { status: "destroyed" } | { status: "unknown" }; @@ -131,6 +131,11 @@ export class WorkerProviderError extends Error { /** Cloud-worker lifecycle capability registered by a plugin. */ export type WorkerProvider = { id: string; + /** + * Provision before preparing an installation when the lease transport decides whether an + * installation is needed. Defaults to false so SSH providers retain prepare-before-allocation. + */ + provisionBeforeInstallation?: boolean; /** * Provision or adopt the lease for this operation id. * Repeating the same operation id must be idempotent across gateway restarts. diff --git a/src/plugins/worker-provider-registry.test.ts b/src/plugins/worker-provider-registry.test.ts index badbfdec603d..647840202758 100644 --- a/src/plugins/worker-provider-registry.test.ts +++ b/src/plugins/worker-provider-registry.test.ts @@ -88,6 +88,23 @@ describe("worker provider registry", () => { ); }); + it("rejects a non-boolean provision-before-installation declaration", () => { + const pluginRegistry = createTestRegistry(); + const provider = { + ...createWorkerProvider("static-ssh"), + provisionBeforeInstallation: "sometimes", + } as unknown as WorkerProvider; + + pluginRegistry.registerWorkerProvider(createOwner("owner", ["static-ssh"]), provider); + + expect(pluginRegistry.registry.workerProviders.size).toBe(0); + expect(pluginRegistry.registry.diagnostics).toContainEqual( + expect.objectContaining({ + message: "worker provider registration provisionBeforeInstallation must be a boolean", + }), + ); + }); + it("rejects a non-function optional SSH identity resolver", () => { const pluginRegistry = createTestRegistry(); const provider = { diff --git a/src/plugins/worker-provider-registry.ts b/src/plugins/worker-provider-registry.ts index cba6e019956d..58cd7d11fe35 100644 --- a/src/plugins/worker-provider-registry.ts +++ b/src/plugins/worker-provider-registry.ts @@ -20,6 +20,15 @@ export function validateWorkerProviderContract( if (provider.renew !== undefined && typeof provider.renew !== "function") { return { ok: false, message: "worker provider registration renew must be a function" }; } + if ( + provider.provisionBeforeInstallation !== undefined && + typeof provider.provisionBeforeInstallation !== "boolean" + ) { + return { + ok: false, + message: "worker provider registration provisionBeforeInstallation must be a boolean", + }; + } if ( provider.resolveSshIdentity !== undefined && typeof provider.resolveSshIdentity !== "function"