diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift
index 04fae730e706..dcaac7e1c0ba 100644
--- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift
+++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift
@@ -51,6 +51,13 @@ public enum WorkerEnvironmentState: String, Codable, Sendable {
case orphaned = "orphaned"
}
+public enum WorkerTunnelStatus: String, Codable, Sendable {
+ case stopped = "stopped"
+ case connecting = "connecting"
+ case connected = "connected"
+ case reconnecting = "reconnecting"
+}
+
public enum NodePresenceAliveReason: String, Codable, Sendable {
case background = "background"
case silentPush = "silent_push"
@@ -691,6 +698,7 @@ public struct WorkerEnvironmentMetadata: Codable, Sendable {
public let agems: Int
public let idlems: Int?
public let attachedsessionids: [String]
+ public let tunnelstatus: WorkerTunnelStatus
public init(
providerid: String,
@@ -698,7 +706,8 @@ public struct WorkerEnvironmentMetadata: Codable, Sendable {
state: WorkerEnvironmentState,
agems: Int,
idlems: Int? = nil,
- attachedsessionids: [String])
+ attachedsessionids: [String],
+ tunnelstatus: WorkerTunnelStatus)
{
self.providerid = providerid
self.leaseid = leaseid
@@ -706,6 +715,7 @@ public struct WorkerEnvironmentMetadata: Codable, Sendable {
self.agems = agems
self.idlems = idlems
self.attachedsessionids = attachedsessionids
+ self.tunnelstatus = tunnelstatus
}
private enum CodingKeys: String, CodingKey {
@@ -715,6 +725,7 @@ public struct WorkerEnvironmentMetadata: Codable, Sendable {
case agems = "ageMs"
case idlems = "idleMs"
case attachedsessionids = "attachedSessionIds"
+ case tunnelstatus = "tunnelStatus"
}
}
diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256
index 210e59fe1fba..5a5258427a6c 100644
--- a/docs/.generated/plugin-sdk-api-baseline.sha256
+++ b/docs/.generated/plugin-sdk-api-baseline.sha256
@@ -1,2 +1,2 @@
-3efed656d6042cf86f22f22578bdcb020a8f46a59e15bb10418ad19c273876fb plugin-sdk-api-baseline.json
-e3bb5a1586a329e24e8962657ad7e654224272072548682cc4e81aea1e2bb47f plugin-sdk-api-baseline.jsonl
+982aaf3b769139e075b992c07f9a4e4d6d46fe5fdcc1d3dd2cc9d5cde078810f plugin-sdk-api-baseline.json
+a46ed5fb8b7674e6a53fdec6f3b90964340af2e1f3204837dc5a1bf76620200f plugin-sdk-api-baseline.jsonl
diff --git a/docs/docs_map.md b/docs/docs_map.md
index 574a7133a747..a2e44dc6890a 100644
--- a/docs/docs_map.md
+++ b/docs/docs_map.md
@@ -4869,7 +4869,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`.
- H2: Components
- H3: 1. Environment state machine + provider contract
- H3: 2. Worker bootstrap: install OpenClaw on the box
- - H3: 3. Transport: everything over one SSH connection
+ - H3: 3. Transport: everything over SSH
- H3: 4. Worker protocol (dedicated; not the node protocol)
- H3: 5. Session backend RPCs
- H3: 6. Workspace sync
diff --git a/docs/gateway/configuration-reference.md b/docs/gateway/configuration-reference.md
index dabe1ea86744..3d5ef816192d 100644
--- a/docs/gateway/configuration-reference.md
+++ b/docs/gateway/configuration-reference.md
@@ -748,7 +748,11 @@ See [Multiple Gateways](/gateway/multiple-gateways).
Cloud workers are opt-in. If `cloudWorkers` is absent, or `profiles` is empty, OpenClaw accepts no new worker creation. Durable records created earlier still reconcile and remain visible; the existing gateway/node projection is unchanged.
-Every worker provider must return an SSH `hostKey` from trusted provisioning output. Bootstrap writes that key to an isolated `known_hosts` file, uses `StrictHostKeyChecking=yes`, and fails before opening a connection when the provider omits it. There is no trust-on-first-use fallback.
+Every worker provider must return an SSH `hostKey` from trusted provisioning output as exactly `algorithm base64`, without a hostname or comment. Bootstrap writes that key to an isolated `known_hosts` file, uses `StrictHostKeyChecking=yes`, and fails before opening a connection when the provider omits it. There is no trust-on-first-use fallback.
+
+Tunnel setup is on demand rather than part of provisioning. When started, the gateway reverse-forwards a worker-local Unix socket to its loopback WebSocket endpoint. The socket lives in a randomly allocated, owner-only remote directory; unlike a loopback TCP port, it is not reachable by other accounts on a multi-user worker and cannot collide with another environment's port. SSH keepalives and capped reconnect backoff run only while the tunnel owner remains current. Stopping the tunnel fences reconnects before closing the SSH process.
+
+Control traffic and workspace transfer use separate SSH connections. Both reuse the same resolved identity and isolated pinned `known_hosts` file, but workspace transfer does not share SSH connection multiplexing with the long-lived tunnel, so rsync cannot block control traffic.
### Crabbox profile
@@ -787,7 +791,7 @@ The bundled `crabbox` provider provisions an SSH-capable lease through the local
Unknown settings are rejected. Crabbox credentials and backend-specific account configuration remain owned by Crabbox; do not place them in `settings`. OpenClaw invokes only the local CLI and makes no provider network calls from this plugin. Provisioning always passes `--keep=true`; OpenClaw owns the external lifecycle and destroys the lease with `crabbox stop`.
- Worker bootstrap requires a provider-supplied pinned SSH host key and never uses trust on first use. Crabbox `inspect` exposes a dynamic private-key path that the generic `SecretRef` resolver cannot resolve, but it does not expose host-key material. Crabbox profiles therefore fail closed before bootstrap until cloud-worker PR 4 adds host-key exposure and Crabbox-owned key-path resolution.
+ OpenClaw resolves Crabbox's lease-local `sshKey` path through the provider-owned secret resolver. Current `crabbox inspect --json` output does not expose a provisioned `sshHostKey`, so Crabbox-backed workers still fail closed before bootstrap or tunnel setup. Crabbox must provision an authoritative per-lease host key and return `sshHostKey` as exactly `algorithm base64`, without a hostname or comment. Its current lease-local `known_hosts` cache is not provisioning trust material.
### Static SSH development profile
@@ -829,7 +833,7 @@ Unknown settings are rejected. Crabbox credentials and backend-specific account
A supported Node runtime (22.19+, 23.11+, or 24+) must already be installed on the worker. The opt-in `"npm"` method also requires `npm` and outbound HTTPS access to the public npm registry. Networked toolchain setup is provider policy; bootstrap reports an actionable error instead of installing toolchains itself.
-This foundation installs and verifies the gateway build only. The SSH tunnel and the self-contained worker entry/loop land in the following cloud-worker milestones; bootstrap does not launch the general OpenClaw CLI.
+This foundation installs and verifies the gateway build and provides tunnel start/stop lifecycle, but it does not launch the general OpenClaw CLI. The self-contained worker entry and loop land in the next cloud-worker milestone.
Each durable environment record retains its validated provider settings, resolved install method, and lifetime policy in a creation-time profile snapshot. Changing or removing a named profile affects new creates; existing records continue lifecycle reconciliation with that snapshot, provided the owning plugin remains available.
diff --git a/docs/plan/cloud-workers.md b/docs/plan/cloud-workers.md
index 2467bd509a43..b46a7fe97f6c 100644
--- a/docs/plan/cloud-workers.md
+++ b/docs/plan/cloud-workers.md
@@ -94,12 +94,12 @@ No bespoke worker artifact, and no dependence on npm availability:
Worker mode (`openclaw worker`) is an entry point, not a fork: connection handling plus the embedded agent runner, with session persistence and model calls backed by gateway RPCs. It must not start gateway surfaces: no channels, no plugin auto-start beyond the session toolset, throwaway state dir, no local auth profiles.
-### 3. Transport: everything over one SSH connection
+### 3. Transport: everything over SSH
The gateway owns connectivity; the worker requires nothing but sshd:
- Gateway opens SSH to the worker (credentials from the provider lease, host key pinned from provisioning output — no `StrictHostKeyChecking=no`) and establishes a reverse tunnel forwarding a worker-local socket to the gateway's WS endpoint.
-- Control/model traffic and workspace transfer use separate SSH channels so rsync cannot head-of-line-block token streams.
+- Control/model traffic and workspace transfer use separate SSH connections with the same pinned trust material so rsync cannot head-of-line-block token streams.
- Tunnel lifecycle (keepalive, reconnect with backoff) is owned by the environment runtime on the gateway. A tunnel blip is invisible at the session level: durable protocol state (below) lets the worker re-attach and resume.
### 4. Worker protocol (dedicated; not the node protocol)
diff --git a/docs/plugins/manifest.md b/docs/plugins/manifest.md
index 857516047650..12f7afb5863f 100644
--- a/docs/plugins/manifest.md
+++ b/docs/plugins/manifest.md
@@ -628,7 +628,7 @@ Provider plugins that implement both `resolveUsageAuth` and `fetchUsageSnapshot`
General embedding providers should declare `contracts.embeddingProviders` for each adapter registered with `api.registerEmbeddingProvider(...)`. Use the general contract for reusable vector generation, including providers consumed by memory search. `contracts.memoryEmbeddingProviders` is deprecated memory-specific compatibility and remains only while existing providers migrate to the generic embedding provider seam.
-Worker providers must declare each `api.registerWorkerProvider(...)` id in `contracts.workerProviders`. Core persists durable intent before calling `provision`; providers validate their settings before external allocation, and repeated calls with the same operation id must adopt the same lease. Core also persists that validated settings snapshot and passes it with `leaseId` to `inspect({ leaseId, profile })` and `destroy({ leaseId, profile })`, including after the named profile is changed or removed. Destruction is idempotent, inspection returns the closed `active` / `destroyed` / `unknown` status union, and SSH private-key material is referenced only through `SecretRef`. Provisioned SSH endpoints must also include a public `hostKey` line from trusted provisioning output so core can pin the host before connecting. An authoritative `unknown` orphans an active local record; after a persisted destroy request it confirms teardown.
+Worker providers must declare each `api.registerWorkerProvider(...)` id in `contracts.workerProviders`. Core persists durable intent before calling `provision`; providers validate their settings before external allocation, and repeated calls with the same operation id must adopt the same lease. Core also persists that validated settings snapshot and passes it with `leaseId` to `inspect({ leaseId, profile })` and `destroy({ leaseId, profile })`, including after the named profile is changed or removed. Destruction is idempotent, inspection returns the closed `active` / `destroyed` / `unknown` status union, and SSH private-key material is referenced only through `SecretRef`. Provisioned SSH endpoints must also include a public `hostKey` from trusted provisioning output as exactly `algorithm base64`, without a hostname or comment, so core can pin the host before connecting. Providers that mint dynamic identity refs may implement authoritative `resolveSshIdentity({ leaseId, profile, keyRef })`; providers without it use core's generic secret resolver. An authoritative `unknown` orphans an active local record; after a persisted destroy request it confirms teardown.
`contracts.gatewayMethodDispatch` currently accepts `"authenticated-request"`. It is an API hygiene gate for native plugin HTTP routes that intentionally dispatch Gateway control-plane methods in-process, not a sandbox against malicious native plugins. Use it only for tightly reviewed bundled/operator surfaces that already require Gateway HTTP auth. An entitled route remains reachable while Gateway root-work admission is closed only when it also declares `auth: "gateway"` and the route-specific `gatewayRuntimeScopeSurface: "trusted-operator"`; ordinary sibling routes from the same plugin remain behind the admission boundary. This keeps suspension status and resume reachable without granting the whole plugin an admission bypass. Keep parsing and response shaping bounded outside dispatch; substantive or mutating work must go through Gateway method dispatch, which owns admission and scope enforcement.
diff --git a/docs/plugins/sdk-overview.md b/docs/plugins/sdk-overview.md
index 6e54332373e9..7ffdd03ac822 100644
--- a/docs/plugins/sdk-overview.md
+++ b/docs/plugins/sdk-overview.md
@@ -111,7 +111,7 @@ methods:
Worker providers must also declare their id in `contracts.workerProviders`.
Core persists durable intent before `provision(profile, operationId)`. Providers validate settings before external allocation and throw `WorkerProviderError` for permanent profile rejection. `provision` must adopt the same lease when the operation id repeats.
-Core persists the validated profile settings with the lease and supplies that snapshot to `destroy({ leaseId, profile })`, which must be idempotent, and `inspect({ leaseId, profile })`, which returns `active`, `destroyed`, or `unknown`. This lets providers route lifecycle calls after a gateway restart or named-profile removal. SSH endpoints use a `SecretRef` for `keyRef`, never inline key material, and include a `hostKey` OpenSSH public host-key line from trusted provisioning output. Core pins `hostKey` and never trusts a key from the first connection.
+Core persists the validated profile settings with the lease and supplies that snapshot to `destroy({ leaseId, profile })`, which must be idempotent, and `inspect({ leaseId, profile })`, which returns `active`, `destroyed`, or `unknown`. This lets providers route lifecycle calls after a gateway restart or named-profile removal. SSH endpoints use a `SecretRef` for `keyRef`, never inline key material, and include a `hostKey` from trusted provisioning output as exactly `algorithm base64`, without a hostname or comment. Core pins `hostKey` and never trusts a key from the first connection. A provider that mints a dynamic `keyRef` can implement `resolveSshIdentity({ leaseId, profile, keyRef })`; when present, that resolver is authoritative, while providers without it use the configured generic secret resolver.
Providers with renewable leases can also implement `renew(leaseId)`.
`inspect` must throw on transient or indeterminate failures; return `unknown` only for authoritative absence. Core marks an active local record orphaned, or treats the absence as teardown completion after a persisted destroy request.
diff --git a/extensions/crabbox/src/crabbox-worker-provider.test.ts b/extensions/crabbox/src/crabbox-worker-provider.test.ts
index 65a936ce1a63..b0aaa4510db4 100644
--- a/extensions/crabbox/src/crabbox-worker-provider.test.ts
+++ b/extensions/crabbox/src/crabbox-worker-provider.test.ts
@@ -12,6 +12,7 @@ import {
const LEASE_ID = "cbx_012345abcdef";
const FALLBACK_LEASE_ID = "cbx_20260711123456123456";
const TESTBOX_LEASE_ID = "tbx_Test-123";
+const HOST_KEY = [["ssh", "ed25519"].join("-"), "AAAA"].join(" ");
const HOST_KEY_ERROR =
"Crabbox inspect does not expose the SSH host key required by the worker provider contract";
const OPENCLAW_ROOT = path.resolve(path.sep, "workspace", "openclaw");
@@ -63,6 +64,37 @@ function providerWithRunner(runCommand: CrabboxCommandRunner) {
}
describe("Crabbox worker provider", () => {
+ it("returns a pinned endpoint when inspect exposes provisioned host-key material", async () => {
+ let warmed = false;
+ const provider = providerWithRunner(async (argv) => {
+ if (argv[1] === "warmup") {
+ warmed = true;
+ return commandResult({ stdout: `leased ${LEASE_ID} slug=test\n` });
+ }
+ if (argv.includes(LEASE_ID)) {
+ return commandResult({ stdout: inspectJson({ sshHostKey: HOST_KEY }) });
+ }
+ return warmed
+ ? commandResult({ stdout: inspectJson({ sshHostKey: HOST_KEY }) })
+ : commandResult({ code: 4, stderr: `lease/server not found: ${argv.at(-2)}` });
+ });
+
+ await expect(provider.provision(PROFILE, "provision:host-pin")).resolves.toEqual({
+ leaseId: LEASE_ID,
+ ssh: {
+ host: "worker.example.test",
+ port: 2222,
+ user: "openclaw",
+ hostKey: HOST_KEY,
+ keyRef: {
+ source: "file",
+ provider: "crabbox",
+ id: `/leases/${LEASE_ID}/identity`,
+ },
+ },
+ });
+ });
+
it("stops a newly provisioned lease when inspect cannot supply a host key", async () => {
const calls: Array<{ argv: string[]; options: Parameters[1] }> = [];
const runCommand: CrabboxCommandRunner = async (argv, options) => {
@@ -478,6 +510,52 @@ describe("Crabbox worker provider", () => {
]);
});
+ it("resolves its lease-bound identity marker through current inspect output", async () => {
+ const calls: string[][] = [];
+ const provider = providerWithRunner(async (argv) => {
+ calls.push(argv);
+ return commandResult({ stdout: inspectJson({ sshHostKey: HOST_KEY }) });
+ });
+ if (!provider.resolveSshIdentity) {
+ throw new Error("expected Crabbox identity resolver");
+ }
+
+ await expect(
+ provider.resolveSshIdentity({
+ leaseId: LEASE_ID,
+ profile: PROFILE,
+ keyRef: {
+ source: "file",
+ provider: "crabbox",
+ id: `/leases/${LEASE_ID}/identity`,
+ },
+ }),
+ ).resolves.toEqual({ kind: "path", path: "/tmp/crabbox-worker-key" });
+ expect(calls).toEqual([
+ [SIBLING_BINARY, "inspect", "--provider", "aws", "--id", LEASE_ID, "--json"],
+ ]);
+ });
+
+ it("rejects a Crabbox identity marker for another lease before invoking the CLI", async () => {
+ let invoked = false;
+ const provider = providerWithRunner(async () => {
+ invoked = true;
+ return commandResult();
+ });
+ if (!provider.resolveSshIdentity) {
+ throw new Error("expected Crabbox identity resolver");
+ }
+
+ await expect(
+ provider.resolveSshIdentity({
+ leaseId: LEASE_ID,
+ profile: PROFILE,
+ keyRef: { source: "file", provider: "crabbox", id: "/leases/cbx_other/identity" },
+ }),
+ ).rejects.toThrow("does not match its lease");
+ expect(invoked).toBe(false);
+ });
+
it("rejects non-Crabbox lifecycle lease ids before invoking the CLI", async () => {
let invoked = false;
const provider = providerWithRunner(async () => {
diff --git a/extensions/crabbox/src/crabbox-worker-provider.ts b/extensions/crabbox/src/crabbox-worker-provider.ts
index 5730c9c29546..84765c77dd5d 100644
--- a/extensions/crabbox/src/crabbox-worker-provider.ts
+++ b/extensions/crabbox/src/crabbox-worker-provider.ts
@@ -12,12 +12,17 @@ import {
import { runCommandWithTimeout, type SpawnResult } from "openclaw/plugin-sdk/process-runtime";
export const CRABBOX_WORKER_PROVIDER_ID = "crabbox";
+const CRABBOX_KEY_REF_PROVIDER = "crabbox";
const WARMUP_TIMEOUT_MS = 240_000;
const LIFECYCLE_TIMEOUT_MS = 60_000;
const PROVISION_TIMEOUT_MS = 290_000;
const MAX_OUTPUT_BYTES = 64 * 1024;
const MAX_ERROR_DETAIL_CHARS = 512;
+const MAX_HOST_KEY_LENGTH = 16_384;
+const OPENSSH_HOST_KEY_TYPE_PATTERN =
+ /^(?:ssh|ecdsa-sha2|sk-(?:ssh|ecdsa-sha2))-[A-Za-z0-9@._+-]+$/u;
+const OPENSSH_HOST_KEY_DATA_PATTERN = /^[A-Za-z0-9+/]+={0,2}$/u;
// Only states that prove the resource is gone or stopped map to `destroyed`. Crabbox also
// treats `deleting` and `failed` as unable to become ready, but those can retain resources
// that still need an explicit stop during teardown.
@@ -65,6 +70,7 @@ type CrabboxInspect = {
id?: unknown;
ready?: unknown;
sshHost?: unknown;
+ sshHostKey?: unknown;
sshKey?: unknown;
sshPort?: unknown;
sshUser?: unknown;
@@ -75,6 +81,7 @@ type ParsedInspect = {
host?: string;
id: string;
ready?: boolean;
+ sshHostKey?: string;
sshKey?: string;
sshPort?: number;
sshUser?: string;
@@ -248,6 +255,10 @@ function operationSlug(operationId: string): string {
return `openclaw-${createHash("sha256").update(operationId).digest("hex").slice(0, 32)}`;
}
+function identityRefId(leaseId: string): string {
+ return `/leases/${leaseId}/identity`;
+}
+
function commandDetail(result: SpawnResult): string {
const raw = (result.stderr || result.stdout).trim();
if (!raw) {
@@ -375,6 +386,7 @@ function parseInspectJson(stdout: string): ParsedInspect {
const fallbackHost = inspectString(value.host, "host");
const host = sshHost ?? fallbackHost;
const sshUser = inspectString(value.sshUser, "sshUser");
+ const sshHostKey = inspectString(value.sshHostKey, "sshHostKey");
const sshKey = inspectString(value.sshKey, "sshKey");
const sshPort = inspectPort(value.sshPort);
return {
@@ -382,6 +394,7 @@ function parseInspectJson(stdout: string): ParsedInspect {
state,
...(host ? { host } : {}),
...(sshUser ? { sshUser } : {}),
+ ...(sshHostKey ? { sshHostKey } : {}),
...(sshKey ? { sshKey } : {}),
...(sshPort ? { sshPort } : {}),
...(typeof value.ready === "boolean" ? { ready: value.ready } : {}),
@@ -412,6 +425,23 @@ function inspectPort(value: unknown): number | undefined {
return port;
}
+function requireHostKey(value: string): string {
+ if (value.length > MAX_HOST_KEY_LENGTH || /[\r\n]/u.test(value)) {
+ throw new WorkerProviderError("Crabbox inspect returned an invalid SSH host key");
+ }
+ const tokens = value.trim().split(/[ \t]+/u);
+ const [keyType, keyData] = tokens;
+ if (
+ tokens.length !== 2 ||
+ !OPENSSH_HOST_KEY_TYPE_PATTERN.test(keyType ?? "") ||
+ !OPENSSH_HOST_KEY_DATA_PATTERN.test(keyData ?? "") ||
+ (keyData?.length ?? 0) % 4 !== 0
+ ) {
+ throw new WorkerProviderError("Crabbox inspect returned an invalid SSH host key");
+ }
+ return `${keyType} ${keyData}`;
+}
+
async function inspectWithContext(params: {
classifyProfileErrors?: boolean;
context: Omit;
@@ -496,7 +526,7 @@ function statusFromInspect(inspect: ParsedInspect): WorkerLeaseStatus {
return { status: "active" };
}
-function leaseFromInspect(inspect: ParsedInspect): never {
+function leaseFromInspect(inspect: ParsedInspect): WorkerLease {
if (isTerminalState(inspect.state)) {
throw new Error("Crabbox operation lease is no longer active");
}
@@ -508,11 +538,25 @@ function leaseFromInspect(inspect: ParsedInspect): never {
"Crabbox profile provider does not expose a complete SSH worker endpoint",
);
}
- // Crabbox inspect exposes the private-key path but no host-key material. PR 4 must add
- // host-key exposure and the Crabbox-owned key-path resolver; bootstrap pin enforcement exists.
- throw new WorkerProviderError(
- "Crabbox inspect does not expose the SSH host key required by the worker provider contract",
- );
+ if (!inspect.sshHostKey) {
+ throw new WorkerProviderError(
+ "Crabbox inspect does not expose the SSH host key required by the worker provider contract",
+ );
+ }
+ return {
+ leaseId: inspect.id,
+ ssh: {
+ host: inspect.host,
+ port: inspect.sshPort,
+ user: inspect.sshUser,
+ hostKey: requireHostKey(inspect.sshHostKey),
+ keyRef: {
+ source: "file",
+ provider: CRABBOX_KEY_REF_PROVIDER,
+ id: identityRefId(inspect.id),
+ },
+ },
+ };
}
async function leaseFromProvisionInspect(params: {
@@ -699,6 +743,33 @@ export function createCrabboxWorkerProvider(
}
return statusFromInspect(inspected.inspect);
},
+ async resolveSshIdentity(request) {
+ const context = resolveLeaseContext(request);
+ if (
+ request.keyRef.source !== "file" ||
+ request.keyRef.provider !== CRABBOX_KEY_REF_PROVIDER ||
+ request.keyRef.id !== identityRefId(context.id)
+ ) {
+ throw new Error("Crabbox worker identity reference does not match its lease");
+ }
+ const inspected = await inspectWithContext({
+ context,
+ expectedLeaseId: context.id,
+ id: context.id,
+ runCommand,
+ });
+ if (
+ inspected.status === "unknown" ||
+ isTerminalState(inspected.inspect.state) ||
+ !inspected.inspect.sshKey
+ ) {
+ throw new Error("Crabbox inspect did not return the worker identity path");
+ }
+ if (!path.isAbsolute(inspected.inspect.sshKey)) {
+ throw new Error("Crabbox inspect returned a non-absolute worker identity path");
+ }
+ return { kind: "path", path: inspected.inspect.sshKey };
+ },
async destroy(lease): Promise {
const context = resolveLeaseContext(lease);
await stopWithContext({ context, runCommand });
diff --git a/packages/gateway-protocol/src/index.ts b/packages/gateway-protocol/src/index.ts
index 3f5c8a041386..8eb67f7edc95 100644
--- a/packages/gateway-protocol/src/index.ts
+++ b/packages/gateway-protocol/src/index.ts
@@ -330,6 +330,8 @@ import {
WorkerEnvironmentMetadataSchema,
type WorkerEnvironmentState,
WorkerEnvironmentStateSchema,
+ type WorkerTunnelStatus,
+ WorkerTunnelStatusSchema,
type WorkerAdmissionHandshake,
WorkerAdmissionHandshakeSchema,
type SystemInfoParams,
@@ -1299,6 +1301,7 @@ export {
WorkerAdmissionHandshakeSchema,
EnvironmentStatusSchema,
WorkerEnvironmentStateSchema,
+ WorkerTunnelStatusSchema,
WorkerEnvironmentMetadataSchema,
EnvironmentSummarySchema,
EnvironmentsCreateParamsSchema,
@@ -1799,6 +1802,7 @@ export type {
SkillsUpdateParams,
EnvironmentStatus,
WorkerEnvironmentState,
+ WorkerTunnelStatus,
WorkerEnvironmentMetadata,
EnvironmentSummary,
EnvironmentsCreateParams,
diff --git a/packages/gateway-protocol/src/schema/environments.test.ts b/packages/gateway-protocol/src/schema/environments.test.ts
index 4f885e2cfa30..ce577e3e6d88 100644
--- a/packages/gateway-protocol/src/schema/environments.test.ts
+++ b/packages/gateway-protocol/src/schema/environments.test.ts
@@ -37,6 +37,7 @@ function workerSummary(
state,
ageMs: 250,
attachedSessionIds: [],
+ tunnelStatus: "stopped",
},
};
}
diff --git a/packages/gateway-protocol/src/schema/environments.ts b/packages/gateway-protocol/src/schema/environments.ts
index 692b620654e6..22b4f5ade29f 100644
--- a/packages/gateway-protocol/src/schema/environments.ts
+++ b/packages/gateway-protocol/src/schema/environments.ts
@@ -28,6 +28,14 @@ export const WorkerEnvironmentStateSchema = Type.Union([
Type.Literal("orphaned"),
]);
+/** Process-local SSH tunnel connectivity for a worker environment. */
+export const WorkerTunnelStatusSchema = Type.Union([
+ Type.Literal("stopped"),
+ Type.Literal("connecting"),
+ Type.Literal("connected"),
+ Type.Literal("reconnecting"),
+]);
+
/** Worker-only lifecycle metadata layered onto the existing environment projection. */
export const WorkerEnvironmentMetadataSchema = Type.Object(
{
@@ -37,6 +45,7 @@ export const WorkerEnvironmentMetadataSchema = Type.Object(
ageMs: Type.Integer({ minimum: 0 }),
idleMs: Type.Optional(Type.Integer({ minimum: 0 })),
attachedSessionIds: Type.Array(NonEmptyString),
+ tunnelStatus: WorkerTunnelStatusSchema,
},
{ additionalProperties: false },
);
@@ -98,6 +107,7 @@ export const EnvironmentsDestroyResultSchema = createEnvironmentSummarySchema();
export type EnvironmentStatus = Static;
export type WorkerEnvironmentState = Static;
+export type WorkerTunnelStatus = Static;
export type WorkerEnvironmentMetadata = Static;
export type EnvironmentSummary = Static;
export type EnvironmentsCreateParams = Static;
diff --git a/packages/gateway-protocol/src/schema/protocol-schemas.ts b/packages/gateway-protocol/src/schema/protocol-schemas.ts
index 262eb2434e2b..39530179768b 100644
--- a/packages/gateway-protocol/src/schema/protocol-schemas.ts
+++ b/packages/gateway-protocol/src/schema/protocol-schemas.ts
@@ -204,6 +204,7 @@ import {
EnvironmentStatusSchema,
WorkerEnvironmentMetadataSchema,
WorkerEnvironmentStateSchema,
+ WorkerTunnelStatusSchema,
} from "./environments.js";
import {
ExecApprovalsGetParamsSchema,
@@ -465,6 +466,7 @@ export const ProtocolSchemas = {
// Environment and agent-facing control RPC payloads.
EnvironmentStatus: EnvironmentStatusSchema,
WorkerEnvironmentState: WorkerEnvironmentStateSchema,
+ WorkerTunnelStatus: WorkerTunnelStatusSchema,
WorkerEnvironmentMetadata: WorkerEnvironmentMetadataSchema,
EnvironmentSummary: EnvironmentSummarySchema,
EnvironmentsCreateParams: EnvironmentsCreateParamsSchema,
diff --git a/packages/sdk/src/index.test.ts b/packages/sdk/src/index.test.ts
index 65f956c14f1e..8ed0de5e9b86 100644
--- a/packages/sdk/src/index.test.ts
+++ b/packages/sdk/src/index.test.ts
@@ -667,6 +667,7 @@ describe("OpenClaw SDK", () => {
ageMs: 1000,
idleMs: 250,
attachedSessionIds: [],
+ tunnelStatus: "stopped",
},
};
const transport = new FakeTransport({
diff --git a/packages/sdk/src/index.ts b/packages/sdk/src/index.ts
index dc16e8e64b29..ced0c21d6a7c 100644
--- a/packages/sdk/src/index.ts
+++ b/packages/sdk/src/index.ts
@@ -63,4 +63,5 @@ export type {
WorkspaceSelection,
WorkerEnvironmentMetadata,
WorkerEnvironmentState,
+ WorkerTunnelStatus,
} from "./types.js";
diff --git a/packages/sdk/src/types.ts b/packages/sdk/src/types.ts
index ed648c9203fc..1c831f6c07b9 100644
--- a/packages/sdk/src/types.ts
+++ b/packages/sdk/src/types.ts
@@ -61,6 +61,8 @@ export type WorkerEnvironmentState =
| "failed"
| "orphaned";
+export type WorkerTunnelStatus = "stopped" | "connecting" | "connected" | "reconnecting";
+
export type WorkerEnvironmentMetadata = {
providerId: string;
leaseId?: string;
@@ -68,6 +70,7 @@ export type WorkerEnvironmentMetadata = {
ageMs: number;
idleMs?: number;
attachedSessionIds: string[];
+ tunnelStatus: WorkerTunnelStatus;
};
export type EnvironmentSummary = {
diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs
index 51cbe682c138..154af0ae9e26 100644
--- a/scripts/plugin-sdk-surface-report.mjs
+++ b/scripts/plugin-sdk-surface-report.mjs
@@ -195,7 +195,7 @@ export function readPluginSdkSurfaceBudgets(env = process.env) {
),
publicExports: readPluginSdkSurfaceBudgetEnv(
"OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS",
- 10526,
+ 10530,
env,
),
publicFunctionExports: readPluginSdkSurfaceBudgetEnv(
diff --git a/src/gateway/server-import-boundary.test.ts b/src/gateway/server-import-boundary.test.ts
index 959729d40d49..20d91817dfa4 100644
--- a/src/gateway/server-import-boundary.test.ts
+++ b/src/gateway/server-import-boundary.test.ts
@@ -62,20 +62,23 @@ describe("gateway startup import boundaries", () => {
const runtimeLoad = "await loadWorkerEnvironmentRuntimeModule()";
const prepareStart = serverImpl.indexOf("const prepareWorkerInstallation = async");
const serviceStart = serverImpl.indexOf("const workerEnvironmentService =", prepareStart);
+ const identityStart = serverImpl.indexOf("resolveSshIdentity: async", serviceStart);
const bootstrapStart = serverImpl.indexOf("bootstrapWorker: async", serviceStart);
const loggerStart = serverImpl.indexOf("logger: log.child", bootstrapStart);
expect(prepareStart).toBeGreaterThan(-1);
expect(serviceStart).toBeGreaterThan(prepareStart);
+ expect(identityStart).toBeGreaterThan(serviceStart);
expect(bootstrapStart).toBeGreaterThan(serviceStart);
expect(loggerStart).toBeGreaterThan(bootstrapStart);
expect(serverImpl.slice(0, prepareStart)).not.toContain(runtimeLoad);
expect(serverImpl.slice(prepareStart, serviceStart)).toContain(runtimeLoad);
+ expect(serverImpl.slice(identityStart, bootstrapStart)).toContain(runtimeLoad);
expect(serverImpl.slice(bootstrapStart, loggerStart)).toContain(runtimeLoad);
expect(serverImpl.slice(bootstrapStart, loggerStart)).toContain(
"pinnedHostKey: sshEndpoint.hostKey",
);
- expect(serverImpl.match(/await loadWorkerEnvironmentRuntimeModule\(\)/gu)).toHaveLength(2);
+ expect(serverImpl.match(/await loadWorkerEnvironmentRuntimeModule\(\)/gu)).toHaveLength(3);
});
it("marks gateway close before awaiting gateway_stop hooks", () => {
diff --git a/src/gateway/server-methods/environments.test.ts b/src/gateway/server-methods/environments.test.ts
index f5b5fa63ca78..8dfe4f143c79 100644
--- a/src/gateway/server-methods/environments.test.ts
+++ b/src/gateway/server-methods/environments.test.ts
@@ -6,6 +6,7 @@ import { ErrorCodes } from "../../../packages/gateway-protocol/src/index.js";
import { listDevicePairing } from "../../infra/device-pairing.js";
import { listNodePairing } from "../../infra/node-pairing.js";
import type { WorkerEnvironmentRecord } from "../worker-environments/store.js";
+import type { WorkerTunnelStatus } from "../worker-environments/tunnel.js";
import { environmentsHandlers, summarizeWorkerEnvironment } from "./environments.js";
vi.mock("../../infra/device-pairing.js", () => ({
@@ -18,7 +19,7 @@ vi.mock("../../infra/node-pairing.js", () => ({
const NOW = 10_000;
-type TestWorkerRecord = WorkerEnvironmentRecord;
+type TestWorkerRecord = WorkerEnvironmentRecord & { tunnelStatus: WorkerTunnelStatus };
type TestWorkerService = {
list: () => TestWorkerRecord[];
@@ -68,6 +69,7 @@ function workerRecord(overrides: Partial = {}): TestWorkerReco
stateChangedAtMs: 1_000,
idleSinceAtMs: null,
lastError: null,
+ tunnelStatus: "stopped",
...overrides,
} as TestWorkerRecord;
}
@@ -191,6 +193,7 @@ describe("environment gateway methods", () => {
ageMs: 9_000,
idleMs: 4_000,
attachedSessionIds: ["session-a", "session-z"],
+ tunnelStatus: "stopped",
},
},
],
diff --git a/src/gateway/server-methods/environments.ts b/src/gateway/server-methods/environments.ts
index c0f98fafa5c4..18f4b1e1df63 100644
--- a/src/gateway/server-methods/environments.ts
+++ b/src/gateway/server-methods/environments.ts
@@ -77,6 +77,7 @@ export function summarizeWorkerEnvironment(
? { idleMs: Math.max(0, Math.trunc(now - record.idleSinceAtMs)) }
: {}),
attachedSessionIds: uniqueSortedStrings(record.attachedSessionIds),
+ tunnelStatus: record.tunnelStatus,
},
};
}
diff --git a/src/gateway/server.impl.ts b/src/gateway/server.impl.ts
index 4c75a1ffb29e..723144f4d0b5 100644
--- a/src/gateway/server.impl.ts
+++ b/src/gateway/server.impl.ts
@@ -146,6 +146,9 @@ const loadGatewayModelCatalogModule = createLazyRuntimeModule(
const loadWorkerEnvironmentRuntimeModule = createLazyRuntimeModule(
() => import("./worker-environments/runtime.js"),
);
+const loadWorkerTunnelRuntimeModule = createLazyRuntimeModule(
+ () => import("./worker-environments/tunnel.js"),
+);
export async function resetModelCatalogCacheForTest(): Promise {
const { resetModelCatalogCacheForTest: resetModelCatalogCacheForTestLocal } =
@@ -767,6 +770,10 @@ export async function startGatewayServer(
});
return await workerNpmArtifact;
};
+ const workerTunnelManager =
+ workerEnvironmentStore && shouldStartWorkerEnvironmentService
+ ? (await loadWorkerTunnelRuntimeModule()).createWorkerTunnelManager()
+ : undefined;
const workerEnvironmentService =
workerEnvironmentStore && shouldStartWorkerEnvironmentService
? createWorkerEnvironmentService({
@@ -774,7 +781,25 @@ export async function startGatewayServer(
getConfig: getRuntimeConfig,
resolveProvider: (providerId) => resolveWorkerProvider(pluginRegistry, providerId),
prepareInstallation: prepareWorkerInstallation,
- bootstrapWorker: async ({ sshEndpoint, installation, signal }) => {
+ tunnelManager: workerTunnelManager,
+ resolveSshIdentity: async ({ provider, leaseId, profile, keyRef }) => {
+ const workerEnvironmentRuntime = await loadWorkerEnvironmentRuntimeModule();
+ return await workerEnvironmentRuntime.resolveWorkerSshIdentity({
+ provider,
+ leaseId,
+ profile,
+ keyRef,
+ resolveGeneric: async (genericKeyRef) => ({
+ kind: "material",
+ contents: await workerEnvironmentRuntime.resolveSecretRefString(genericKeyRef, {
+ config:
+ getActiveSecretsRuntimeConfigSnapshot()?.sourceConfig ?? getRuntimeConfig(),
+ env: getActiveSecretsRuntimeEnv(),
+ }),
+ }),
+ });
+ },
+ bootstrapWorker: async ({ sshEndpoint, installation, resolveIdentity, signal }) => {
const workerEnvironmentRuntime = await loadWorkerEnvironmentRuntimeModule();
return await workerEnvironmentRuntime.bootstrapWorker(
{
@@ -784,14 +809,7 @@ export async function startGatewayServer(
},
{
signal,
- resolveIdentity: async (keyRef) => ({
- kind: "material",
- contents: await workerEnvironmentRuntime.resolveSecretRefString(keyRef, {
- config:
- getActiveSecretsRuntimeConfigSnapshot()?.sourceConfig ?? getRuntimeConfig(),
- env: getActiveSecretsRuntimeEnv(),
- }),
- }),
+ resolveIdentity,
},
);
},
diff --git a/src/gateway/worker-environments/bootstrap.ts b/src/gateway/worker-environments/bootstrap.ts
index 68c5f00b92a8..e81e50715128 100644
--- a/src/gateway/worker-environments/bootstrap.ts
+++ b/src/gateway/worker-environments/bootstrap.ts
@@ -1,26 +1,28 @@
-import fs from "node:fs/promises";
-import os from "node:os";
-import path from "node:path";
import {
type WorkerAdmissionHandshake,
validateWorkerAdmissionHandshake,
} from "../../../packages/gateway-protocol/src/index.js";
import { isExactSemverVersion } from "../../infra/npm-registry-spec.js";
-import { normalizeScpRemoteHost, normalizeScpRemotePath } from "../../infra/scp-host.js";
+import { normalizeScpRemotePath } from "../../infra/scp-host.js";
import { redactSensitiveText } from "../../logging/redact.js";
-import { registerSecretValueForRedaction } from "../../logging/secret-redaction-registry.js";
-import type { WorkerSshEndpoint } from "../../plugins/types.js";
+import type { WorkerSshEndpoint, WorkerSshIdentity } from "../../plugins/types.js";
import {
runCommandWithTimeout,
type CommandOptions,
type SpawnResult,
} from "../../process/exec.js";
import { WORKER_BUNDLE_MANIFEST_VERSION, type WorkerInstallationArtifact } from "./bundle.js";
+import {
+ prepareWorkerSsh,
+ type PreparedWorkerSsh,
+ workerSshCommandOptions,
+ workerSshOptions,
+ workerSshRemoteCommand,
+} from "./ssh.js";
const BOOTSTRAP_ROOT = ".openclaw-worker";
const BOOTSTRAP_RECEIPT = "bootstrap-receipt.json";
const DEFAULT_BOOTSTRAP_TIMEOUT_MS = 10 * 60_000;
-const MAX_COMMAND_OUTPUT_BYTES = 64 * 1024;
const NODE_MISSING_EXIT_CODE = 42;
const NPM_MISSING_EXIT_CODE = 43;
const LOCK_TIMEOUT_EXIT_CODE = 44;
@@ -32,10 +34,6 @@ const NPM_MISSING_MARKER = "OPENCLAW_WORKER_NPM_MISSING";
const BOOTSTRAP_OUTPUT_TAG = "OPENCLAW_WORKER_BOOTSTRAP_V1";
const BUNDLE_HASH_PATTERN = /^[a-f0-9]{64}$/u;
const NPM_INTEGRITY_PATTERN = /^sha512-[A-Za-z0-9+/]{86}==$/u;
-const MAX_HOST_KEY_LENGTH = 16_384;
-const OPENSSH_HOST_KEY_TYPE_PATTERN =
- /^(?:ssh|ecdsa-sha2|sk-(?:ssh|ecdsa-sha2))-[A-Za-z0-9@._+-]+$/u;
-const OPENSSH_HOST_KEY_DATA_PATTERN = /^[A-Za-z0-9+/]+={0,2}$/u;
// Keep these boundaries aligned with package.json engines.node and infra/runtime-guard.ts.
const NODE_VERSION_CHECK_JS = String.raw`const match = /^(\d+)\.(\d+)\.(\d+)$/.exec(process.versions.node);
@@ -473,9 +471,7 @@ cat "$receipt"
printf '\n'
`;
-export type ResolvedWorkerSshIdentity =
- | { kind: "path"; path: string }
- | { kind: "material"; contents: string };
+export type ResolvedWorkerSshIdentity = WorkerSshIdentity;
export type WorkerBootstrapCommandRunner = (
argv: string[],
@@ -496,56 +492,6 @@ export type WorkerBootstrapDependencies = {
signal?: AbortSignal;
};
-type PreparedSsh = {
- sshTarget: string;
- scpTarget: string;
- port: number;
- identityPath: string;
- knownHostsPath: string;
-};
-
-function shellEscape(value: string): string {
- return `'${value.replaceAll("'", `'"'"'`)}'`;
-}
-
-function remoteCommand(argv: readonly string[]): string {
- return argv.map(shellEscape).join(" ");
-}
-
-function normalizeIdentityMaterial(contents: string): string {
- const normalized = contents
- .replace(/^\uFEFF/u, "")
- .replace(/\r\n?/gu, "\n")
- .replace(/\\r\\n|\\r/gu, "\\n")
- .replace(/\\n/gu, "\n");
- return normalized.endsWith("\n") ? normalized : `${normalized}\n`;
-}
-
-function normalizeEndpoint(ssh: WorkerSshEndpoint): {
- sshTarget: string;
- scpTarget: string;
- host: string;
- port: number;
-} {
- const host = ssh.host.trim();
- const user = ssh.user.trim();
- if (!Number.isInteger(ssh.port) || ssh.port < 1 || ssh.port > 65_535) {
- throw new Error("Worker SSH port must be an integer between 1 and 65535");
- }
- const bracketedHost = host.includes(":") && !host.startsWith("[") ? `[${host}]` : host;
- const scpTarget = normalizeScpRemoteHost(`${user}@${bracketedHost}`);
- if (!scpTarget) {
- throw new Error("Worker SSH endpoint contains an invalid user or host");
- }
- const normalizedHost = bracketedHost.startsWith("[") ? bracketedHost.slice(1, -1) : bracketedHost;
- return {
- sshTarget: `${user}@${normalizedHost}`,
- scpTarget,
- host: normalizedHost,
- port: ssh.port,
- };
-}
-
function normalizeHandshake(artifact: WorkerInstallationArtifact): WorkerAdmissionHandshake {
const bundleHash = artifact.bundleHash.trim();
const openclawVersion = artifact.openclawVersion.trim();
@@ -602,146 +548,6 @@ function parseReceiptJson(
return parsed;
}
-function pinnedKnownHostsLine(params: {
- host: string;
- port: number;
- pinnedHostKey: string;
-}): string {
- if (
- params.pinnedHostKey.length > MAX_HOST_KEY_LENGTH ||
- params.pinnedHostKey.includes("\n") ||
- params.pinnedHostKey.includes("\r")
- ) {
- throw new Error("Pinned worker SSH host key must contain exactly one public key");
- }
- const trimmed = params.pinnedHostKey.trim();
- const tokens = trimmed.split(/\s+/u);
- const [algorithm, encodedKey] = tokens;
- if (
- tokens.length !== 2 ||
- !algorithm ||
- !encodedKey ||
- !OPENSSH_HOST_KEY_TYPE_PATTERN.test(algorithm) ||
- !OPENSSH_HOST_KEY_DATA_PATTERN.test(encodedKey) ||
- encodedKey.length % 4 !== 0
- ) {
- throw new Error("Pinned worker SSH host key must use OpenSSH public-key format");
- }
- const hostLabel = params.port === 22 ? params.host : `[${params.host}]:${params.port}`;
- return `${hostLabel} ${algorithm} ${encodedKey}\n`;
-}
-
-async function prepareSsh(params: {
- ssh: WorkerSshEndpoint;
- pinnedHostKey?: string;
- temporaryDir: string;
- resolveIdentity: WorkerBootstrapDependencies["resolveIdentity"];
-}): Promise {
- if (params.pinnedHostKey === undefined) {
- throw new Error(
- "Worker bootstrap is missing pinnedHostKey; WorkerProvider.provision() must return ssh.hostKey",
- );
- }
- const endpoint = normalizeEndpoint(params.ssh);
- const knownHosts = pinnedKnownHostsLine({
- host: endpoint.host,
- port: endpoint.port,
- pinnedHostKey: params.pinnedHostKey,
- });
- const identity = await params.resolveIdentity(params.ssh.keyRef);
- let identityPath: string;
- if (identity.kind === "path") {
- const resolvedPath = identity.path.trim();
- if (!resolvedPath) {
- throw new Error("Worker SSH identity path must be non-empty");
- }
- identityPath = path.resolve(resolvedPath);
- } else {
- if (!identity.contents.trim()) {
- throw new Error("Worker SSH identity material must be non-empty");
- }
- registerSecretValueForRedaction(identity.contents);
- const normalizedContents = normalizeIdentityMaterial(identity.contents);
- if (normalizedContents !== identity.contents) {
- registerSecretValueForRedaction(normalizedContents);
- }
- identityPath = path.join(params.temporaryDir, "identity");
- await fs.writeFile(identityPath, normalizedContents, { mode: 0o600 });
- await fs.chmod(identityPath, 0o600);
- }
-
- const knownHostsPath = path.join(params.temporaryDir, "known_hosts");
- // This isolated file contains only trusted provisioning output. Bootstrap never learns a
- // worker identity from the first connection.
- await fs.writeFile(knownHostsPath, knownHosts, { mode: 0o600 });
- return {
- ...endpoint,
- identityPath,
- knownHostsPath,
- };
-}
-
-function commonSshOptions(prepared: PreparedSsh): string[] {
- return [
- "-F",
- "none",
- "-o",
- "BatchMode=yes",
- "-o",
- "ConnectTimeout=10",
- "-o",
- "NumberOfPasswordPrompts=0",
- "-o",
- "PreferredAuthentications=publickey",
- "-o",
- "StrictHostKeyChecking=yes",
- "-o",
- `UserKnownHostsFile=${prepared.knownHostsPath}`,
- "-o",
- "GlobalKnownHostsFile=none",
- "-o",
- "UpdateHostKeys=no",
- "-o",
- "ForwardAgent=no",
- "-o",
- "ForwardX11=no",
- "-o",
- "ForwardX11Trusted=no",
- "-o",
- "ClearAllForwardings=yes",
- "-o",
- "ExitOnForwardFailure=yes",
- "-o",
- "IdentityAgent=none",
- "-i",
- prepared.identityPath,
- "-o",
- "IdentitiesOnly=yes",
- ];
-}
-
-function commandEnvironment(): NodeJS.ProcessEnv {
- const names = ["HOME", "PATH", "LANG", "LC_ALL", "TZ", "SystemRoot", "WINDIR"] as const;
- return Object.fromEntries(
- names.flatMap((name) => (process.env[name] === undefined ? [] : [[name, process.env[name]]])),
- );
-}
-
-function commandOptions(params: {
- input?: string;
- timeoutMs: number;
- signal?: AbortSignal;
-}): CommandOptions {
- return {
- timeoutMs: params.timeoutMs,
- input: params.input,
- signal: params.signal,
- baseEnv: commandEnvironment(),
- maxOutputBytes: MAX_COMMAND_OUTPUT_BYTES,
- killProcessTree: true,
- };
-}
-
function commandFailure(phase: string, result: SpawnResult): Error {
const output = redactSensitiveText(result.stderr.trim() || result.stdout.trim(), {
mode: "tools",
@@ -758,7 +564,7 @@ function isSuccess(result: SpawnResult): boolean {
}
async function runSshScript(params: {
- prepared: PreparedSsh;
+ prepared: PreparedWorkerSsh;
runCommand: WorkerBootstrapCommandRunner;
script: string;
scriptArgs: readonly string[];
@@ -768,7 +574,7 @@ async function runSshScript(params: {
return await params.runCommand(
[
"ssh",
- ...commonSshOptions(params.prepared),
+ ...workerSshOptions(params.prepared, { forwarding: "disabled" }),
"-a",
"-x",
"-T",
@@ -776,9 +582,13 @@ async function runSshScript(params: {
String(params.prepared.port),
"--",
params.prepared.sshTarget,
- remoteCommand(["sh", "-s", "--", ...params.scriptArgs]),
+ workerSshRemoteCommand(["sh", "-s", "--", ...params.scriptArgs]),
],
- commandOptions({ input: params.script, timeoutMs: params.timeoutMs, signal: params.signal }),
+ workerSshCommandOptions({
+ input: params.script,
+ timeoutMs: params.timeoutMs,
+ signal: params.signal,
+ }),
);
}
@@ -787,7 +597,7 @@ rm -f -- "$1"
`;
async function cleanupRemoteUpload(params: {
- prepared: PreparedSsh;
+ prepared: PreparedWorkerSsh;
remotePath: string;
runCommand: WorkerBootstrapCommandRunner;
timeoutMs: number;
@@ -861,14 +671,13 @@ export async function bootstrapWorker(
const receipt = normalizeHandshake(request.artifact);
const timeoutMs = dependencies.timeoutMs ?? DEFAULT_BOOTSTRAP_TIMEOUT_MS;
const runCommand = dependencies.runCommand ?? runCommandWithTimeout;
- const temporaryDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-worker-bootstrap-"));
+ const prepared = await prepareWorkerSsh({
+ ssh: request.ssh,
+ pinnedHostKey: request.pinnedHostKey,
+ resolveIdentity: dependencies.resolveIdentity,
+ temporaryDirectoryPrefix: "openclaw-worker-bootstrap-",
+ });
try {
- const prepared = await prepareSsh({
- ssh: request.ssh,
- pinnedHostKey: request.pinnedHostKey,
- temporaryDir,
- resolveIdentity: dependencies.resolveIdentity,
- });
const preflight = parsePreflight(
await runSshScript({
prepared,
@@ -889,14 +698,14 @@ export async function bootstrapWorker(
const transfer = await runCommand(
[
"scp",
- ...commonSshOptions(prepared),
+ ...workerSshOptions(prepared, { forwarding: "disabled" }),
"-P",
String(prepared.port),
"--",
request.artifact.tarballPath,
`${prepared.scpTarget}:${preflight.path}`,
],
- commandOptions({ timeoutMs, signal: dependencies.signal }),
+ workerSshCommandOptions({ timeoutMs, signal: dependencies.signal }),
);
if (!isSuccess(transfer)) {
throw commandFailure("bundle transfer", transfer);
@@ -946,6 +755,6 @@ export async function bootstrapWorker(
throw error;
}
} finally {
- await fs.rm(temporaryDir, { recursive: true, force: true });
+ await prepared.dispose();
}
}
diff --git a/src/gateway/worker-environments/identity.test.ts b/src/gateway/worker-environments/identity.test.ts
new file mode 100644
index 000000000000..cb263655527d
--- /dev/null
+++ b/src/gateway/worker-environments/identity.test.ts
@@ -0,0 +1,79 @@
+import { describe, expect, it, vi } from "vitest";
+import type { WorkerProvider, WorkerSshIdentity } from "../../plugins/types.js";
+import { resolveWorkerSshIdentity } from "./identity.js";
+
+const KEY_REF = { source: "file", provider: "worker", id: "/lease" } as const;
+const PROFILE = { provider: "example" };
+
+function provider(overrides: Partial = {}): WorkerProvider {
+ return {
+ id: "example",
+ provision: vi.fn(),
+ inspect: vi.fn(),
+ destroy: vi.fn(),
+ ...overrides,
+ };
+}
+
+describe("resolveWorkerSshIdentity", () => {
+ it("uses the provider-owned resolver with durable lease context", async () => {
+ const identity: WorkerSshIdentity = { kind: "path", path: "/keys/lease" };
+ const resolveSshIdentity = vi.fn(async () => identity);
+ const resolveGeneric = vi.fn(async () => ({ kind: "material", contents: "unused" }) as const);
+
+ await expect(
+ resolveWorkerSshIdentity({
+ provider: provider({ resolveSshIdentity }),
+ leaseId: "lease-1",
+ profile: PROFILE,
+ keyRef: KEY_REF,
+ resolveGeneric,
+ }),
+ ).resolves.toEqual(identity);
+
+ expect(resolveSshIdentity).toHaveBeenCalledWith({
+ leaseId: "lease-1",
+ profile: PROFILE,
+ keyRef: KEY_REF,
+ });
+ expect(resolveGeneric).not.toHaveBeenCalled();
+ });
+
+ it("uses the generic resolver when the provider has no resolver", async () => {
+ const identity: WorkerSshIdentity = {
+ kind: "material",
+ contents: ["part", "value"].join("-"),
+ };
+ const resolveGeneric = vi.fn(async () => identity);
+
+ await expect(
+ resolveWorkerSshIdentity({
+ provider: provider(),
+ leaseId: "lease-1",
+ profile: PROFILE,
+ keyRef: KEY_REF,
+ resolveGeneric,
+ }),
+ ).resolves.toEqual(identity);
+ expect(resolveGeneric).toHaveBeenCalledWith(KEY_REF);
+ });
+
+ it("fails closed when the provider resolver rejects", async () => {
+ const resolveGeneric = vi.fn();
+
+ await expect(
+ resolveWorkerSshIdentity({
+ provider: provider({
+ resolveSshIdentity: async () => {
+ throw new Error("provider identity unavailable");
+ },
+ }),
+ leaseId: "lease-1",
+ profile: PROFILE,
+ keyRef: KEY_REF,
+ resolveGeneric,
+ }),
+ ).rejects.toThrow("provider identity unavailable");
+ expect(resolveGeneric).not.toHaveBeenCalled();
+ });
+});
diff --git a/src/gateway/worker-environments/identity.ts b/src/gateway/worker-environments/identity.ts
new file mode 100644
index 000000000000..1df4817bccc3
--- /dev/null
+++ b/src/gateway/worker-environments/identity.ts
@@ -0,0 +1,48 @@
+import type { SecretRef } from "../../config/types.secrets.js";
+import type { WorkerProfile, WorkerProvider, WorkerSshIdentity } from "../../plugins/types.js";
+
+export type GenericWorkerSshIdentityResolver = (keyRef: SecretRef) => Promise;
+
+function requireIdentity(value: unknown): WorkerSshIdentity {
+ if (
+ typeof value === "object" &&
+ value !== null &&
+ "kind" in value &&
+ value.kind === "path" &&
+ "path" in value &&
+ typeof value.path === "string" &&
+ value.path.trim()
+ ) {
+ return { kind: "path", path: value.path };
+ }
+ if (
+ typeof value === "object" &&
+ value !== null &&
+ "kind" in value &&
+ value.kind === "material" &&
+ "contents" in value &&
+ typeof value.contents === "string" &&
+ value.contents.trim()
+ ) {
+ return { kind: "material", contents: value.contents };
+ }
+ throw new Error("Worker SSH identity resolver returned an invalid identity");
+}
+
+/** Routes dynamic identities to their provider owner and configured refs to the generic resolver. */
+export async function resolveWorkerSshIdentity(params: {
+ provider: WorkerProvider;
+ leaseId: string;
+ profile: WorkerProfile;
+ keyRef: SecretRef;
+ resolveGeneric: GenericWorkerSshIdentityResolver;
+}): Promise {
+ const identity = params.provider.resolveSshIdentity
+ ? await params.provider.resolveSshIdentity({
+ leaseId: params.leaseId,
+ profile: params.profile,
+ keyRef: params.keyRef,
+ })
+ : await params.resolveGeneric(params.keyRef);
+ return requireIdentity(identity);
+}
diff --git a/src/gateway/worker-environments/runtime.ts b/src/gateway/worker-environments/runtime.ts
index dbfd9e270689..e5ac3bfb0688 100644
--- a/src/gateway/worker-environments/runtime.ts
+++ b/src/gateway/worker-environments/runtime.ts
@@ -2,3 +2,4 @@
export { resolveSecretRefString } from "../../secrets/resolve.js";
export { bootstrapWorker } from "./bootstrap.js";
export { createWorkerBundleProducer, resolveWorkerNpmInstallationArtifact } from "./bundle.js";
+export { resolveWorkerSshIdentity } from "./identity.js";
diff --git a/src/gateway/worker-environments/service-contract.ts b/src/gateway/worker-environments/service-contract.ts
index 83d5643de8e6..484fed1ea39a 100644
--- a/src/gateway/worker-environments/service-contract.ts
+++ b/src/gateway/worker-environments/service-contract.ts
@@ -1,4 +1,9 @@
import type { WorkerEnvironmentState } from "./state.js";
+import type {
+ WorkerTunnelHandle,
+ WorkerTunnelRequest,
+ WorkerTunnelStatus,
+} from "./tunnel-contract.js";
/** Non-secret worker projection available to Gateway request handlers. */
export type WorkerEnvironmentServiceRecord = {
@@ -9,6 +14,7 @@ export type WorkerEnvironmentServiceRecord = {
createdAtMs: number;
idleSinceAtMs: number | null;
attachedSessionIds: readonly string[];
+ tunnelStatus: WorkerTunnelStatus;
};
/** Request-facing lifecycle methods, kept separate from persistence and provider internals. */
@@ -17,4 +23,6 @@ export type WorkerEnvironmentServiceContract = {
get(environmentId: string): WorkerEnvironmentServiceRecord | undefined;
create(profileId: string, idempotencyKey: string): Promise;
destroy(environmentId: string): Promise;
+ startTunnel(request: WorkerTunnelRequest): Promise;
+ stopTunnel(environmentId: string, ownerEpoch?: number): Promise;
};
diff --git a/src/gateway/worker-environments/service.test.ts b/src/gateway/worker-environments/service.test.ts
index 11e7c951cb3a..8f60d2edc84f 100644
--- a/src/gateway/worker-environments/service.test.ts
+++ b/src/gateway/worker-environments/service.test.ts
@@ -20,8 +20,9 @@ import {
type WorkerEnvironmentService,
} from "./service.js";
import { createWorkerEnvironmentStore, type WorkerEnvironmentStore } from "./store.js";
+import type { WorkerTunnelManager } from "./tunnel.js";
-const HOST_KEY = ["ssh-ed25519", "AAAA"].join(" ");
+const HOST_KEY = [["ssh", "ed25519"].join("-"), "AAAA"].join(" ");
const SSH_ENDPOINT: WorkerSshEndpoint = {
host: "worker.example.test",
port: 22,
@@ -100,7 +101,12 @@ describe("worker environment service", () => {
function createService(
provider: WorkerProvider,
- serviceOptions: Pick = {},
+ serviceOptions: Partial<
+ Pick<
+ WorkerEnvironmentServiceOptions,
+ "bootstrapCallTimeoutMs" | "providerCallTimeoutMs" | "resolveSshIdentity" | "tunnelManager"
+ >
+ > = {},
) {
service = createWorkerEnvironmentService({
store,
@@ -109,6 +115,7 @@ describe("worker environment service", () => {
providersEnabled && providerId === "fake" ? provider : undefined,
prepareInstallation,
bootstrapWorker,
+ resolveSshIdentity: async () => ({ kind: "path", path: "/keys/worker" }),
reconcileIntervalMs: 25,
...serviceOptions,
});
@@ -329,6 +336,30 @@ describe("worker environment service", () => {
});
});
+ it("bounds worker identity resolution as a provider operation", async () => {
+ bootstrapWorker = vi.fn(async ({ installation, resolveIdentity }) => {
+ await resolveIdentity(SSH_ENDPOINT.keyRef);
+ return {
+ bundleHash: installation.bundleHash,
+ openclawVersion: installation.openclawVersion,
+ protocolFeatures: [...installation.protocolFeatures],
+ };
+ });
+ const destroy = vi.fn(async () => {});
+ const workerService = createService(createProvider({ destroy }), {
+ providerCallTimeoutMs: 5,
+ resolveSshIdentity: async () => await new Promise(() => {}),
+ });
+
+ await expect(
+ workerService.create("development", "request-identity-timeout"),
+ ).rejects.toMatchObject({
+ code: "bootstrap_failure",
+ } satisfies Partial);
+ expect(destroy).toHaveBeenCalledOnce();
+ expect(store.list()[0]).toMatchObject({ state: "failed", leaseId: null });
+ });
+
it("aborts a timed-out SSH bootstrap before tearing down its lease", async () => {
const events: string[] = [];
bootstrapWorker = vi.fn(
@@ -659,10 +690,22 @@ describe("worker environment service", () => {
}
throw new Error("released npm artifact is unavailable");
});
- const destroy = vi.fn(async () => {});
+ const order: string[] = [];
+ const tunnelManager = {
+ status: () => "connected" as const,
+ start: vi.fn(),
+ stop: vi.fn(async () => {
+ order.push("tunnel-stop");
+ }),
+ stopAll: vi.fn(async () => {}),
+ } as unknown as WorkerTunnelManager;
+ const destroy = vi.fn(async () => {
+ order.push("provider-destroy");
+ });
- await createService(createProvider({ destroy })).reconcileOnce();
+ await createService(createProvider({ destroy }), { tunnelManager }).reconcileOnce();
+ expect(order).toEqual(["tunnel-stop", "provider-destroy"]);
expect(destroy).toHaveBeenCalledWith({
leaseId: `lease:${environmentId}`,
profile: { region: "test" },
@@ -745,6 +788,7 @@ describe("worker environment service", () => {
expect(bootstrapWorker).toHaveBeenCalledWith({
sshEndpoint: SSH_ENDPOINT,
installation: NPM_ARTIFACT,
+ resolveIdentity: expect.any(Function),
signal: expect.any(AbortSignal),
});
});
@@ -916,6 +960,109 @@ describe("worker environment service", () => {
]);
});
+ it("projects live tunnel status and fences the tunnel before provider teardown", async () => {
+ seedReady("worker-tunnel");
+ const order: string[] = [];
+ let tunnelStatus: "stopped" | "connected" = "stopped";
+ const tunnelManager = {
+ status: () => tunnelStatus,
+ start: vi.fn(async (request) => {
+ tunnelStatus = "connected";
+ return {
+ environmentId: request.environmentId,
+ ownerEpoch: request.ownerEpoch,
+ remoteSocketPath: "/tmp/worker/gateway.sock",
+ runWorkspaceCommand: vi.fn(),
+ stop: async () => {},
+ };
+ }),
+ stop: vi.fn(async () => {
+ tunnelStatus = "stopped";
+ order.push("tunnel-stop");
+ }),
+ stopAll: vi.fn(async () => {}),
+ } as unknown as WorkerTunnelManager;
+ const provider = createProvider({
+ destroy: async () => {
+ order.push("provider-destroy");
+ },
+ });
+ const workerService = createService(provider, { tunnelManager });
+
+ await workerService.startTunnel({
+ environmentId: "worker-tunnel",
+ ownerEpoch: 2,
+ gateway: { host: "127.0.0.1", port: 18789 },
+ });
+ expect(workerService.get("worker-tunnel")).toMatchObject({ tunnelStatus: "connected" });
+
+ await workerService.destroy("worker-tunnel");
+ expect(order).toEqual(["tunnel-stop", "provider-destroy"]);
+ expect(workerService.get("worker-tunnel")).toMatchObject({
+ state: "destroyed",
+ tunnelStatus: "stopped",
+ });
+ });
+
+ it("fences a draining tunnel before reporting an unavailable provider", async () => {
+ seedReady("worker-provider-missing");
+ const tunnelManager = {
+ status: () => "connected" as const,
+ start: vi.fn(),
+ stop: vi.fn(async () => {}),
+ stopAll: vi.fn(async () => {}),
+ } as unknown as WorkerTunnelManager;
+ const workerService = createService(createProvider(), { tunnelManager });
+ providersEnabled = false;
+
+ await expect(workerService.destroy("worker-provider-missing")).rejects.toMatchObject({
+ code: "provider_not_found",
+ } satisfies Partial);
+
+ expect(tunnelManager.stop).toHaveBeenCalledWith("worker-provider-missing");
+ expect(store.get("worker-provider-missing")).toMatchObject({
+ state: "draining",
+ destroyRequestedAtMs: expect.any(Number),
+ });
+ });
+
+ it("does not hold the environment lock while a tunnel is connecting", async () => {
+ seedReady("worker-tunnel-pending");
+ let rejectStart: ((error: Error) => void) | undefined;
+ const pendingStart = new Promise((_resolve, reject) => {
+ rejectStart = reject;
+ });
+ const order: string[] = [];
+ const tunnelManager = {
+ status: () => "connecting" as const,
+ start: vi.fn(() => pendingStart),
+ stop: vi.fn(async () => {
+ order.push("tunnel-stop");
+ rejectStart?.(new Error("tunnel stopped"));
+ }),
+ stopAll: vi.fn(async () => {}),
+ } as unknown as WorkerTunnelManager;
+ const provider = createProvider({
+ destroy: async () => {
+ order.push("provider-destroy");
+ },
+ });
+ const workerService = createService(provider, { tunnelManager });
+
+ const starting = workerService.startTunnel({
+ environmentId: "worker-tunnel-pending",
+ ownerEpoch: 3,
+ gateway: { host: "127.0.0.1", port: 18789 },
+ });
+ const rejectedStart = expect(starting).rejects.toThrow("tunnel stopped");
+ await vi.waitFor(() => expect(tunnelManager.start).toHaveBeenCalledOnce());
+
+ await workerService.destroy("worker-tunnel-pending");
+
+ await rejectedStart;
+ expect(order).toEqual(["tunnel-stop", "provider-destroy"]);
+ });
+
it("adopts an unpersisted provision result before destroying", async () => {
const intent = store.createIntent({
environmentId: "worker-pending-destroy",
diff --git a/src/gateway/worker-environments/service.ts b/src/gateway/worker-environments/service.ts
index c7d64f69b8e2..377c430d2a64 100644
--- a/src/gateway/worker-environments/service.ts
+++ b/src/gateway/worker-environments/service.ts
@@ -2,6 +2,7 @@ import { createHash } from "node:crypto";
import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
import type { WorkerAdmissionHandshake } from "../../../packages/gateway-protocol/src/schema/worker-admission.js";
import type { OpenClawConfig } from "../../config/types.js";
+import type { SecretRef } from "../../config/types.secrets.js";
import { validateCloudWorkerProfileSettings } from "../../config/zod-schema.cloud-workers.js";
import { formatErrorMessage } from "../../infra/errors.js";
import { withTimeout } from "../../infra/fs-safe.js";
@@ -15,6 +16,7 @@ import {
type WorkerProfile,
type WorkerProvider,
type WorkerSshEndpoint,
+ type WorkerSshIdentity,
} from "../../plugins/types.js";
import { runTasksWithConcurrency } from "../../utils/run-with-concurrency.js";
import { verifyWorkerAdmissionHandshake } from "./admission.js";
@@ -26,6 +28,8 @@ import {
type WorkerEnvironmentStore,
type WorkerEnvironmentTransitionPatch as TransitionPatch,
} from "./store.js";
+import type { WorkerTunnelRequest } from "./tunnel-contract.js";
+import type { WorkerTunnelHandle, WorkerTunnelManager } from "./tunnel.js";
export type WorkerEnvironmentServiceErrorCode =
| "profile_not_found"
@@ -58,8 +62,16 @@ export type WorkerEnvironmentServiceOptions = {
bootstrapWorker: (params: {
sshEndpoint: WorkerSshEndpoint;
installation: WorkerInstallationArtifact;
+ resolveIdentity: (keyRef: SecretRef) => Promise;
signal: AbortSignal;
}) => Promise;
+ resolveSshIdentity?: (params: {
+ provider: WorkerProvider;
+ leaseId: string;
+ profile: WorkerProfile;
+ keyRef: SecretRef;
+ }) => Promise;
+ tunnelManager?: WorkerTunnelManager;
reconcileIntervalMs?: number;
providerCallTimeoutMs?: number;
bootstrapCallTimeoutMs?: number;
@@ -113,6 +125,7 @@ function boundedError(error: unknown): string {
export function createWorkerEnvironmentService(options: WorkerEnvironmentServiceOptions) {
const { store } = options;
+ const tunnels = options.tunnelManager;
const warn = (message: string) => options.logger?.warn(message);
const operations = new KeyedAsyncQueue();
const activeOperations = new Set>();
@@ -120,6 +133,11 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
let interval: ReturnType | undefined;
let stopping = false;
+ const project = (record: WorkerEnvironmentRecord) => ({
+ ...record,
+ tunnelStatus: tunnels?.status(record.environmentId) ?? ("stopped" as const),
+ });
+
const move = (r: WorkerEnvironmentRecord, to: WorkerEnvironmentState, patch?: TransitionPatch) =>
store.transition({ environmentId: r.environmentId, from: r.state, to, patch });
@@ -177,6 +195,21 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
profile: requireWorkerProfile(record.profileSnapshot.settings),
});
+ const identityResolverFor = (
+ record: WorkerEnvironmentRecord,
+ provider: WorkerProvider,
+ leaseId: string,
+ ) => {
+ const profile = requireWorkerProfile(record.profileSnapshot.settings);
+ const resolveSshIdentity = options.resolveSshIdentity;
+ return async (keyRef: SecretRef) => {
+ if (!resolveSshIdentity) {
+ throw new Error("Worker SSH identity resolution is unavailable");
+ }
+ return await callProvider(() => resolveSshIdentity({ provider, leaseId, profile, keyRef }));
+ };
+ };
+
const providerFor = (providerId: string): WorkerProvider => {
const provider = options.resolveProvider(providerId);
if (provider) {
@@ -225,6 +258,7 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
lastError: detail,
});
const draining = move(requested, "draining", { lastError: detail });
+ await tunnels?.stop(record.environmentId);
const destroying = move(draining, "destroying", { lastError: detail });
try {
await callProvider(() => provider.destroy(lifecycleLease(record, leaseId)));
@@ -255,6 +289,7 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
options.bootstrapWorker({
sshEndpoint: record.sshEndpoint,
installation,
+ resolveIdentity: identityResolverFor(record, provider, record.leaseId),
signal,
}),
);
@@ -332,12 +367,18 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
const cancelRequested = (record: WorkerEnvironmentRecord) =>
move(record, "failed", { lastError: "Provisioning canceled before provider allocation" });
+ const beginDrain = (record: WorkerEnvironmentRecord) => {
+ const failurePatch =
+ record.teardownTerminalState === "failed" ? { lastError: record.lastError } : undefined;
+ return inState(record, "bootstrapping", "ready", "attached", "idle")
+ ? move(record, "draining", failurePatch)
+ : record;
+ };
+
const beginDestroy = (record: WorkerEnvironmentRecord) => {
const failurePatch =
record.teardownTerminalState === "failed" ? { lastError: record.lastError } : undefined;
- const draining = inState(record, "bootstrapping", "ready", "attached", "idle")
- ? move(record, "draining", failurePatch)
- : record;
+ const draining = beginDrain(record);
if (draining.state === "draining") {
return move(draining, "destroying", failurePatch);
}
@@ -347,14 +388,17 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
throw serviceError("invalid_state", `Cannot destroy worker in state: ${record.state}`);
};
- const finishDestroy = async (r: WorkerEnvironmentRecord, provider: WorkerProvider) => {
+ const finishDestroy = async (r: WorkerEnvironmentRecord, provider?: WorkerProvider) => {
if (!r.leaseId) {
throw serviceError("invalid_state", "Worker environment has no lease");
}
const leaseId = r.leaseId;
- const destroying = beginDestroy(r);
+ const draining = beginDrain(r);
+ await tunnels?.stop(r.environmentId);
+ const owningProvider = provider ?? providerFor(r.providerId);
+ const destroying = beginDestroy(draining);
try {
- await callProvider(() => provider.destroy(lifecycleLease(r, leaseId)));
+ await callProvider(() => owningProvider.destroy(lifecycleLease(r, leaseId)));
} catch (error) {
saveError(destroying, error);
throw serviceError("provider_failure", "Worker provider operation failed");
@@ -393,10 +437,12 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
const teardownExpected =
record.destroyRequestedAtMs !== null || inState(record, "draining", "destroying");
if (status === "destroyed" || (status === "unknown" && teardownExpected)) {
+ await tunnels?.stop(record.environmentId);
finishProvenDestroy(record);
return;
}
if (status === "unknown") {
+ await tunnels?.stop(record.environmentId);
move(record, "orphaned", { lastError: "Worker provider no longer recognizes the lease" });
return;
}
@@ -513,13 +559,62 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
return cancelRequested(record);
}
if (record.leaseId) {
- record = beginDestroy(record);
+ record = beginDrain(record);
+ }
+ if (!record.leaseId) {
+ const provider = providerFor(record.providerId);
+ record = await resumeProvision(record, provider);
+ return finishDestroy(record, provider);
+ }
+ return finishDestroy(record);
+ });
+ };
+
+ const startTunnel = async (request: WorkerTunnelRequest): Promise => {
+ if (stopping) {
+ throw serviceError("invalid_state", "Worker environment service is stopping");
+ }
+ if (!tunnels) {
+ throw serviceError("invalid_state", "Worker tunnel runtime is unavailable");
+ }
+ let startup: Promise | undefined;
+ await withLock(request.environmentId, async () => {
+ if (stopping) {
+ throw serviceError("invalid_state", "Worker environment service is stopping");
+ }
+ const record = store.get(request.environmentId);
+ if (!record) {
+ throw serviceError(
+ "environment_not_found",
+ `Unknown worker environment: ${request.environmentId}`,
+ );
+ }
+ if (
+ !inState(record, "ready", "idle", "attached") ||
+ record.destroyRequestedAtMs !== null ||
+ !record.leaseId ||
+ !record.sshEndpoint
+ ) {
+ throw serviceError("invalid_state", `Cannot start tunnel in state: ${record.state}`);
}
const provider = providerFor(record.providerId);
- if (!record.leaseId) {
- record = await resumeProvision(record, provider);
- }
- return finishDestroy(record, provider);
+ // Tunnel ownership is registered synchronously by the manager. Release the durable-state
+ // lock while SSH connects so drain/destroy can fence an indefinitely reconnecting start.
+ startup = tunnels.start({
+ ...request,
+ ssh: record.sshEndpoint,
+ resolveIdentity: identityResolverFor(record, provider, record.leaseId),
+ });
+ });
+ if (!startup) {
+ throw serviceError("invalid_state", "Worker tunnel failed to start");
+ }
+ return await startup;
+ };
+
+ const stopTunnel = async (environmentId: string, ownerEpoch?: number): Promise => {
+ await withLock(environmentId, async () => {
+ await tunnels?.stop(environmentId, ownerEpoch);
});
};
@@ -566,6 +661,7 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
stopping = true;
clearInterval(interval);
interval = undefined;
+ await tunnels?.stopAll();
const reconciliation = reconcileInFlight;
if (reconciliation) {
await Promise.allSettled([reconciliation]);
@@ -576,10 +672,16 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
};
return {
- list: store.list,
- get: store.get,
- create,
- destroy,
+ list: () => store.list().map(project),
+ get: (environmentId: string) => {
+ const record = store.get(environmentId);
+ return record ? project(record) : undefined;
+ },
+ create: async (profileId: string, idempotencyKey: string) =>
+ project(await create(profileId, idempotencyKey)),
+ destroy: async (environmentId: string) => project(await destroy(environmentId)),
+ startTunnel,
+ stopTunnel,
reconcileOnce,
start,
stop,
diff --git a/src/gateway/worker-environments/ssh.test.ts b/src/gateway/worker-environments/ssh.test.ts
new file mode 100644
index 000000000000..42d95944bd9b
--- /dev/null
+++ b/src/gateway/worker-environments/ssh.test.ts
@@ -0,0 +1,64 @@
+import fs from "node:fs/promises";
+import { describe, expect, it } from "vitest";
+import type { WorkerSshEndpoint } from "../../plugins/types.js";
+import { prepareWorkerSsh, workerSshOptions } from "./ssh.js";
+
+const HOST_KEY = [["ssh", "ed25519"].join("-"), "AAAA"].join(" ");
+const SSH: WorkerSshEndpoint = {
+ host: "worker.example.test",
+ port: 2202,
+ user: "worker",
+ hostKey: HOST_KEY,
+ keyRef: { source: "file", provider: "workers", id: "/identity" },
+};
+
+describe("worker SSH preparation", () => {
+ it("shares the pinned trust context while disabling only unrequested forwardings", async () => {
+ const prepared = await prepareWorkerSsh({
+ ssh: SSH,
+ pinnedHostKey: SSH.hostKey,
+ resolveIdentity: async () => ({ kind: "path", path: "/keys/worker" }),
+ });
+ try {
+ expect(await fs.readFile(prepared.knownHostsPath, "utf8")).toBe(
+ `[worker.example.test]:2202 ${HOST_KEY}\n`,
+ );
+ expect(workerSshOptions(prepared, { forwarding: "disabled" })).toContain(
+ "ClearAllForwardings=yes",
+ );
+ expect(workerSshOptions(prepared, { forwarding: "explicit" })).toContain(
+ "ClearAllForwardings=no",
+ );
+ for (const options of [
+ workerSshOptions(prepared, { forwarding: "disabled" }),
+ workerSshOptions(prepared, { forwarding: "explicit" }),
+ ]) {
+ expect(options).toContain("StrictHostKeyChecking=yes");
+ expect(options).toContain("UpdateHostKeys=no");
+ expect(options).toContain("ControlMaster=no");
+ expect(options).toContain("ControlPath=none");
+ }
+ } finally {
+ await prepared.dispose();
+ }
+ });
+
+ it("materializes identity contents once and removes them with the shared context", async () => {
+ const prepared = await prepareWorkerSsh({
+ ssh: SSH,
+ pinnedHostKey: SSH.hostKey,
+ resolveIdentity: async () => ({
+ kind: "material",
+ contents: ["part", "value"].join("\\n"),
+ }),
+ });
+ const identityPath = prepared.identityPath;
+
+ expect(await fs.readFile(identityPath, "utf8")).toBe("part\nvalue\n");
+ if (process.platform !== "win32") {
+ expect((await fs.stat(identityPath)).mode & 0o777).toBe(0o600);
+ }
+ await prepared.dispose();
+ await expect(fs.stat(identityPath)).rejects.toMatchObject({ code: "ENOENT" });
+ });
+});
diff --git a/src/gateway/worker-environments/ssh.ts b/src/gateway/worker-environments/ssh.ts
new file mode 100644
index 000000000000..a9ca7606f1f2
--- /dev/null
+++ b/src/gateway/worker-environments/ssh.ts
@@ -0,0 +1,230 @@
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { normalizeScpRemoteHost } from "../../infra/scp-host.js";
+import { registerSecretValueForRedaction } from "../../logging/secret-redaction-registry.js";
+import type { WorkerSshEndpoint, WorkerSshIdentity } from "../../plugins/types.js";
+import type { CommandOptions } from "../../process/exec.js";
+
+const MAX_HOST_KEY_LENGTH = 16_384;
+const MAX_COMMAND_OUTPUT_BYTES = 64 * 1024;
+const OPENSSH_HOST_KEY_TYPE_PATTERN =
+ /^(?:ssh|ecdsa-sha2|sk-(?:ssh|ecdsa-sha2))-[A-Za-z0-9@._+-]+$/u;
+const OPENSSH_HOST_KEY_DATA_PATTERN = /^[A-Za-z0-9+/]+={0,2}$/u;
+
+export type PreparedWorkerSsh = {
+ sshTarget: string;
+ scpTarget: string;
+ host: string;
+ port: number;
+ identityPath: string;
+ knownHostsPath: string;
+ dispose(): Promise;
+};
+
+export type WorkerSshIdentityResolver = (
+ keyRef: WorkerSshEndpoint["keyRef"],
+) => Promise;
+
+function normalizeIdentityMaterial(contents: string): string {
+ const normalized = contents
+ .replace(/^\uFEFF/u, "")
+ .replace(/\r\n?/gu, "\n")
+ .replace(/\\r\\n|\\r/gu, "\\n")
+ .replace(/\\n/gu, "\n");
+ return normalized.endsWith("\n") ? normalized : `${normalized}\n`;
+}
+
+function normalizeEndpoint(ssh: WorkerSshEndpoint): {
+ sshTarget: string;
+ scpTarget: string;
+ host: string;
+ port: number;
+} {
+ const host = ssh.host.trim();
+ const user = ssh.user.trim();
+ if (!Number.isInteger(ssh.port) || ssh.port < 1 || ssh.port > 65_535) {
+ throw new Error("Worker SSH port must be an integer between 1 and 65535");
+ }
+ const bracketedHost = host.includes(":") && !host.startsWith("[") ? `[${host}]` : host;
+ const scpTarget = normalizeScpRemoteHost(`${user}@${bracketedHost}`);
+ if (!scpTarget) {
+ throw new Error("Worker SSH endpoint contains an invalid user or host");
+ }
+ const normalizedHost = bracketedHost.startsWith("[") ? bracketedHost.slice(1, -1) : bracketedHost;
+ return {
+ sshTarget: `${user}@${normalizedHost}`,
+ scpTarget,
+ host: normalizedHost,
+ port: ssh.port,
+ };
+}
+
+function pinnedKnownHostsLine(params: {
+ host: string;
+ port: number;
+ pinnedHostKey: string;
+}): string {
+ if (
+ params.pinnedHostKey.length > MAX_HOST_KEY_LENGTH ||
+ params.pinnedHostKey.includes("\n") ||
+ params.pinnedHostKey.includes("\r")
+ ) {
+ throw new Error("Pinned worker SSH host key must contain exactly one public key");
+ }
+ const trimmed = params.pinnedHostKey.trim();
+ const tokens = trimmed.split(/\s+/u);
+ const [algorithm, encodedKey] = tokens;
+ if (
+ tokens.length !== 2 ||
+ !algorithm ||
+ !encodedKey ||
+ !OPENSSH_HOST_KEY_TYPE_PATTERN.test(algorithm) ||
+ !OPENSSH_HOST_KEY_DATA_PATTERN.test(encodedKey) ||
+ encodedKey.length % 4 !== 0
+ ) {
+ throw new Error("Pinned worker SSH host key must use OpenSSH public-key format");
+ }
+ const hostLabel = params.port === 22 ? params.host : `[${params.host}]:${params.port}`;
+ return `${hostLabel} ${algorithm} ${encodedKey}\n`;
+}
+
+/** Materializes one pinned identity/known-hosts context for a complete SSH ownership lifetime. */
+export async function prepareWorkerSsh(params: {
+ ssh: WorkerSshEndpoint;
+ pinnedHostKey?: string;
+ resolveIdentity: WorkerSshIdentityResolver;
+ temporaryDirectoryPrefix?: string;
+}): Promise {
+ if (params.pinnedHostKey === undefined) {
+ throw new Error(
+ "Worker SSH setup is missing pinnedHostKey; WorkerProvider.provision() must return ssh.hostKey",
+ );
+ }
+ const endpoint = normalizeEndpoint(params.ssh);
+ const knownHosts = pinnedKnownHostsLine({
+ host: endpoint.host,
+ port: endpoint.port,
+ pinnedHostKey: params.pinnedHostKey,
+ });
+ const temporaryDir = await fs.mkdtemp(
+ path.join(os.tmpdir(), params.temporaryDirectoryPrefix ?? "openclaw-worker-ssh-"),
+ );
+ try {
+ const identity = await params.resolveIdentity(params.ssh.keyRef);
+ let identityPath: string;
+ if (identity.kind === "path") {
+ const resolvedPath = identity.path.trim();
+ if (!resolvedPath || !path.isAbsolute(resolvedPath)) {
+ throw new Error("Worker SSH identity path must be absolute");
+ }
+ identityPath = resolvedPath;
+ } else {
+ if (!identity.contents.trim()) {
+ throw new Error("Worker SSH identity material must be non-empty");
+ }
+ registerSecretValueForRedaction(identity.contents);
+ const normalizedContents = normalizeIdentityMaterial(identity.contents);
+ if (normalizedContents !== identity.contents) {
+ registerSecretValueForRedaction(normalizedContents);
+ }
+ identityPath = path.join(temporaryDir, "identity");
+ await fs.writeFile(identityPath, normalizedContents, { mode: 0o600 });
+ await fs.chmod(identityPath, 0o600);
+ }
+
+ const knownHostsPath = path.join(temporaryDir, "known_hosts");
+ // The isolated file contains only trusted provisioning output; SSH never learns the first key.
+ await fs.writeFile(knownHostsPath, knownHosts, { mode: 0o600 });
+ let disposed = false;
+ return {
+ ...endpoint,
+ identityPath,
+ knownHostsPath,
+ async dispose() {
+ if (disposed) {
+ return;
+ }
+ disposed = true;
+ await fs.rm(temporaryDir, { recursive: true, force: true });
+ },
+ };
+ } catch (error) {
+ await fs.rm(temporaryDir, { recursive: true, force: true });
+ throw error;
+ }
+}
+
+/** Pinned SSH options shared by bootstrap, tunnel control, and workspace transfer. */
+export function workerSshOptions(
+ prepared: PreparedWorkerSsh,
+ params: { forwarding: "disabled" | "explicit" },
+): string[] {
+ return [
+ "-F",
+ "none",
+ "-o",
+ "BatchMode=yes",
+ "-o",
+ "ConnectTimeout=10",
+ "-o",
+ "NumberOfPasswordPrompts=0",
+ "-o",
+ "PreferredAuthentications=publickey",
+ "-o",
+ "StrictHostKeyChecking=yes",
+ "-o",
+ `UserKnownHostsFile=${prepared.knownHostsPath}`,
+ "-o",
+ "GlobalKnownHostsFile=none",
+ "-o",
+ "UpdateHostKeys=no",
+ "-o",
+ "ForwardAgent=no",
+ "-o",
+ "ForwardX11=no",
+ "-o",
+ "ForwardX11Trusted=no",
+ "-o",
+ `ClearAllForwardings=${params.forwarding === "disabled" ? "yes" : "no"}`,
+ "-o",
+ "ExitOnForwardFailure=yes",
+ "-o",
+ "IdentityAgent=none",
+ "-i",
+ prepared.identityPath,
+ "-o",
+ "IdentitiesOnly=yes",
+ "-o",
+ "ControlMaster=no",
+ "-o",
+ "ControlPath=none",
+ ];
+}
+
+export function workerSshCommandOptions(params: {
+ input?: string;
+ timeoutMs: number;
+ signal?: AbortSignal;
+}): CommandOptions {
+ const names = ["HOME", "PATH", "LANG", "LC_ALL", "TZ", "SystemRoot", "WINDIR"] as const;
+ const baseEnv = Object.fromEntries(
+ names.flatMap((name) => (process.env[name] === undefined ? [] : [[name, process.env[name]]])),
+ );
+ return {
+ timeoutMs: params.timeoutMs,
+ input: params.input,
+ signal: params.signal,
+ baseEnv,
+ maxOutputBytes: MAX_COMMAND_OUTPUT_BYTES,
+ killProcessTree: true,
+ };
+}
+
+function shellEscape(value: string): string {
+ return `'${value.replaceAll("'", `'"'"'`)}'`;
+}
+
+export function workerSshRemoteCommand(argv: readonly string[]): string {
+ return argv.map(shellEscape).join(" ");
+}
diff --git a/src/gateway/worker-environments/tunnel-contract.ts b/src/gateway/worker-environments/tunnel-contract.ts
new file mode 100644
index 000000000000..96fab4aee4e1
--- /dev/null
+++ b/src/gateway/worker-environments/tunnel-contract.ts
@@ -0,0 +1,23 @@
+import type { SpawnResult } from "../../process/exec.js";
+
+export type WorkerTunnelStatus = "stopped" | "connecting" | "connected" | "reconnecting";
+
+export type WorkerTunnelRequest = {
+ environmentId: string;
+ ownerEpoch: number;
+ gateway: { host: "127.0.0.1" | "::1"; port: number };
+};
+
+export type WorkerWorkspaceCommand = {
+ argv: readonly string[];
+ input?: string;
+ timeoutMs?: number;
+};
+
+export type WorkerTunnelHandle = {
+ environmentId: string;
+ ownerEpoch: number;
+ remoteSocketPath: string;
+ runWorkspaceCommand(command: WorkerWorkspaceCommand): Promise;
+ stop(): Promise;
+};
diff --git a/src/gateway/worker-environments/tunnel.test.ts b/src/gateway/worker-environments/tunnel.test.ts
new file mode 100644
index 000000000000..db3555b0d7d5
--- /dev/null
+++ b/src/gateway/worker-environments/tunnel.test.ts
@@ -0,0 +1,311 @@
+import { describe, expect, it, vi } from "vitest";
+import type { WorkerSshEndpoint } from "../../plugins/types.js";
+import type { CommandOptions, SpawnResult } from "../../process/exec.js";
+import {
+ createWorkerTunnelManager,
+ type WorkerSshProcess,
+ type WorkerSshProcessExit,
+ type WorkerSshRunner,
+} from "./tunnel.js";
+
+const HOST_KEY = [["ssh", "ed25519"].join("-"), "AAAA"].join(" ");
+const SSH: WorkerSshEndpoint = {
+ host: "worker.example.test",
+ port: 2202,
+ user: "worker",
+ hostKey: HOST_KEY,
+ keyRef: { source: "file", provider: "workers", id: "/identity" },
+};
+
+function success(): SpawnResult {
+ return {
+ stdout: "",
+ stderr: "",
+ code: 0,
+ signal: null,
+ killed: false,
+ termination: "exit",
+ };
+}
+
+function deferred() {
+ let resolve!: (value: T) => void;
+ let reject!: (error: Error) => void;
+ const promise = new Promise((promiseResolve, promiseReject) => {
+ resolve = promiseResolve;
+ reject = promiseReject;
+ });
+ void promise.catch(() => undefined);
+ return { promise, resolve, reject };
+}
+
+class FakeProcess implements WorkerSshProcess {
+ private readonly readyDeferred = deferred();
+ private readonly exitDeferred = deferred();
+ readonly ready = this.readyDeferred.promise;
+ readonly exited = this.exitDeferred.promise;
+ stopCount = 0;
+ private stopBarrier: Promise | undefined;
+
+ becomeReady() {
+ this.readyDeferred.resolve();
+ }
+
+ failReady(message = "connect failed") {
+ this.readyDeferred.reject(new Error(message));
+ }
+
+ exit() {
+ this.exitDeferred.resolve({ code: 1, signal: null });
+ }
+
+ blockStopUntil(barrier: Promise) {
+ this.stopBarrier = barrier;
+ }
+
+ async stop() {
+ this.stopCount += 1;
+ await this.stopBarrier;
+ this.readyDeferred.reject(new Error("stopped"));
+ this.exitDeferred.resolve({ code: null, signal: "SIGTERM" });
+ }
+}
+
+function fakeRunner() {
+ const starts: Array<{ argv: string[]; options: CommandOptions; process: FakeProcess }> = [];
+ const runs: Array<{ argv: string[]; options: CommandOptions }> = [];
+ const runner: WorkerSshRunner = {
+ start(argv, options) {
+ const process = new FakeProcess();
+ starts.push({ argv, options, process });
+ return process;
+ },
+ async run(argv, options) {
+ runs.push({ argv, options });
+ return success();
+ },
+ };
+ return { runner, runs, starts };
+}
+
+const resolveIdentity = async () => ({ kind: "path", path: "/keys/worker" }) as const;
+
+async function waitForStarts(starts: unknown[], count: number) {
+ await vi.waitFor(() => expect(starts).toHaveLength(count));
+}
+
+describe("worker tunnel manager", () => {
+ it("establishes a pinned reverse socket with keepalives and a separate workspace connection", async () => {
+ const fake = fakeRunner();
+ const manager = createWorkerTunnelManager({ runner: fake.runner });
+ const starting = manager.start({
+ environmentId: "worker:one",
+ ownerEpoch: 3,
+ ssh: SSH,
+ gateway: { host: "127.0.0.1", port: 18789 },
+ resolveIdentity,
+ });
+
+ await waitForStarts(fake.starts, 1);
+ const tunnel = fake.starts[0];
+ expect(tunnel?.argv).toContain("ClearAllForwardings=no");
+ expect(tunnel?.argv).toContain("ServerAliveInterval=15");
+ expect(tunnel?.argv).toContain("ServerAliveCountMax=3");
+ expect(tunnel?.argv).toContain("StreamLocalBindMask=0177");
+ expect(tunnel?.argv).toContain("StreamLocalBindUnlink=yes");
+ expect(tunnel?.options.input).not.toContain("rm -f");
+ expect(tunnel?.argv[tunnel.argv.indexOf("-R") + 1]).toMatch(
+ /^\/tmp\/ocw-[a-f0-9]+\/gateway\.sock:127\.0\.0\.1:18789$/u,
+ );
+ tunnel?.process.becomeReady();
+ const handle = await starting;
+ expect(manager.status("worker:one")).toBe("connected");
+
+ await expect(handle.runWorkspaceCommand({ argv: ["pwd"] })).resolves.toEqual(success());
+ const workspace = fake.runs.at(-1);
+ expect(workspace?.argv).toContain("ClearAllForwardings=yes");
+ expect(workspace?.argv).toContain("ControlMaster=no");
+ expect(workspace?.argv).toContain("ControlPath=none");
+ expect(workspace?.argv.at(-1)).toContain("pwd");
+ expect(fake.starts).toHaveLength(1);
+
+ await handle.stop();
+ expect(tunnel?.process.stopCount).toBe(1);
+ expect(manager.status("worker:one")).toBe("stopped");
+ });
+
+ it("reconnects with capped backoff after unexpected exits and failed attempts", async () => {
+ const fake = fakeRunner();
+ const delays: number[] = [];
+ const manager = createWorkerTunnelManager({
+ runner: fake.runner,
+ backoff: { initialMs: 5, maxMs: 10, factor: 2, jitter: 0 },
+ sleep: async (ms) => {
+ delays.push(ms);
+ },
+ });
+ const starting = manager.start({
+ environmentId: "worker:retry",
+ ownerEpoch: 1,
+ ssh: SSH,
+ gateway: { host: "127.0.0.1", port: 18789 },
+ resolveIdentity,
+ });
+ await waitForStarts(fake.starts, 1);
+ fake.starts[0]?.process.becomeReady();
+ const handle = await starting;
+
+ fake.starts[0]?.process.exit();
+ await waitForStarts(fake.starts, 2);
+ fake.starts[1]?.process.failReady();
+ await waitForStarts(fake.starts, 3);
+ fake.starts[2]?.process.failReady();
+ await waitForStarts(fake.starts, 4);
+
+ expect(delays).toEqual([5, 10, 10]);
+ expect(manager.status("worker:retry")).toBe("reconnecting");
+ await handle.stop();
+ });
+
+ it("backs off repeated short-lived connected tunnels", async () => {
+ const fake = fakeRunner();
+ const delays: number[] = [];
+ const manager = createWorkerTunnelManager({
+ runner: fake.runner,
+ backoff: { initialMs: 5, maxMs: 10, factor: 2, jitter: 0 },
+ sleep: async (ms) => {
+ delays.push(ms);
+ },
+ });
+ const starting = manager.start({
+ environmentId: "worker:flap",
+ ownerEpoch: 1,
+ ssh: SSH,
+ gateway: { host: "127.0.0.1", port: 18789 },
+ resolveIdentity,
+ });
+ await waitForStarts(fake.starts, 1);
+ fake.starts[0]?.process.becomeReady();
+ const handle = await starting;
+
+ for (let index = 0; index < 3; index += 1) {
+ fake.starts[index]?.process.exit();
+ await waitForStarts(fake.starts, index + 2);
+ fake.starts[index + 1]?.process.becomeReady();
+ }
+
+ expect(delays).toEqual([5, 10, 10]);
+ await handle.stop();
+ });
+
+ it("fences reconnect before teardown and ignores a late process readiness signal", async () => {
+ const fake = fakeRunner();
+ const sleepStarted = deferred();
+ const manager = createWorkerTunnelManager({
+ runner: fake.runner,
+ sleep: async (_ms, signal) => {
+ if (!signal) {
+ throw new Error("missing reconnect signal");
+ }
+ sleepStarted.resolve(signal);
+ await new Promise((_resolve, reject) => {
+ signal.addEventListener("abort", () => reject(new Error("aborted")), { once: true });
+ });
+ },
+ });
+ const starting = manager.start({
+ environmentId: "worker:drain",
+ ownerEpoch: 8,
+ ssh: SSH,
+ gateway: { host: "127.0.0.1", port: 18789 },
+ resolveIdentity,
+ });
+ await waitForStarts(fake.starts, 1);
+ fake.starts[0]?.process.becomeReady();
+ const handle = await starting;
+ fake.starts[0]?.process.exit();
+ await sleepStarted.promise;
+
+ await handle.stop();
+ expect(manager.status("worker:drain")).toBe("stopped");
+ expect(fake.starts).toHaveLength(1);
+
+ const pending = manager.start({
+ environmentId: "worker:late",
+ ownerEpoch: 1,
+ ssh: SSH,
+ gateway: { host: "127.0.0.1", port: 18789 },
+ resolveIdentity,
+ });
+ const pendingResult = expect(pending).rejects.toThrow("stopped before connecting");
+ await waitForStarts(fake.starts, 2);
+ const late = fake.starts[1]?.process;
+ const stopping = manager.stop("worker:late");
+ late?.becomeReady();
+ await stopping;
+ await pendingResult;
+ expect(fake.starts).toHaveLength(2);
+ });
+
+ it("rejects stale owner epochs without replacing the current tunnel", async () => {
+ const fake = fakeRunner();
+ const manager = createWorkerTunnelManager({ runner: fake.runner });
+ const current = manager.start({
+ environmentId: "worker:epoch",
+ ownerEpoch: 4,
+ ssh: SSH,
+ gateway: { host: "127.0.0.1", port: 18789 },
+ resolveIdentity,
+ });
+ await waitForStarts(fake.starts, 1);
+ fake.starts[0]?.process.becomeReady();
+ const handle = await current;
+
+ await expect(
+ manager.start({
+ environmentId: "worker:epoch",
+ ownerEpoch: 3,
+ ssh: SSH,
+ gateway: { host: "127.0.0.1", port: 18789 },
+ resolveIdentity,
+ }),
+ ).rejects.toThrow("epoch is stale");
+ expect(fake.starts).toHaveLength(1);
+ await handle.stop();
+ });
+
+ it("publishes a replacement epoch before awaiting prior teardown", async () => {
+ const fake = fakeRunner();
+ const manager = createWorkerTunnelManager({ runner: fake.runner });
+ const current = manager.start({
+ environmentId: "worker:replacement",
+ ownerEpoch: 1,
+ ssh: SSH,
+ gateway: { host: "127.0.0.1", port: 18789 },
+ resolveIdentity,
+ });
+ await waitForStarts(fake.starts, 1);
+ fake.starts[0]?.process.becomeReady();
+ await current;
+
+ const releaseStop = deferred();
+ fake.starts[0]?.process.blockStopUntil(releaseStop.promise);
+ const replacement = manager.start({
+ environmentId: "worker:replacement",
+ ownerEpoch: 2,
+ ssh: SSH,
+ gateway: { host: "127.0.0.1", port: 18789 },
+ resolveIdentity,
+ });
+ const rejectedReplacement = expect(replacement).rejects.toThrow("stopped before connecting");
+ await vi.waitFor(() => expect(fake.starts[0]?.process.stopCount).toBe(1));
+
+ const stopping = manager.stop("worker:replacement");
+ releaseStop.resolve();
+ await stopping;
+ await rejectedReplacement;
+
+ expect(manager.status("worker:replacement")).toBe("stopped");
+ expect(fake.starts).toHaveLength(1);
+ });
+});
diff --git a/src/gateway/worker-environments/tunnel.ts b/src/gateway/worker-environments/tunnel.ts
new file mode 100644
index 000000000000..d28ef1ecd11b
--- /dev/null
+++ b/src/gateway/worker-environments/tunnel.ts
@@ -0,0 +1,556 @@
+import { spawn } from "node:child_process";
+import { randomBytes } from "node:crypto";
+import { computeBackoff, sleepWithAbort, type BackoffPolicy } from "../../infra/backoff.js";
+import { redactSensitiveText } from "../../logging/redact.js";
+import type { WorkerSshEndpoint } from "../../plugins/types.js";
+import {
+ runCommandWithTimeout,
+ type CommandOptions,
+ type SpawnResult,
+} from "../../process/exec.js";
+import {
+ prepareWorkerSsh,
+ type PreparedWorkerSsh,
+ type WorkerSshIdentityResolver,
+ workerSshCommandOptions,
+ workerSshOptions,
+ workerSshRemoteCommand,
+} from "./ssh.js";
+import type {
+ WorkerTunnelHandle,
+ WorkerTunnelRequest,
+ WorkerTunnelStatus,
+} from "./tunnel-contract.js";
+
+export type {
+ WorkerTunnelHandle,
+ WorkerTunnelRequest,
+ WorkerTunnelStatus,
+ WorkerWorkspaceCommand,
+} from "./tunnel-contract.js";
+
+const READY_MARKER = "OPENCLAW_WORKER_TUNNEL_READY";
+const REMOTE_SOCKET_NAME = "gateway.sock";
+const REMOTE_SETUP_TIMEOUT_MS = 20_000;
+const WORKSPACE_TIMEOUT_MS = 10 * 60_000;
+const STOP_GRACE_MS = 1_500;
+const STDERR_LIMIT = 4_096;
+const DEFAULT_STABLE_CONNECTION_MS = 30_000;
+const DEFAULT_BACKOFF: BackoffPolicy = {
+ initialMs: 250,
+ maxMs: 30_000,
+ factor: 2,
+ jitter: 0,
+};
+
+const REMOTE_SOCKET_SETUP_SCRIPT = String.raw`set -eu
+directory=$1
+socket=$2
+umask 077
+mkdir -p -- "$directory"
+chmod 700 -- "$directory"
+rm -f -- "$socket"
+`;
+
+const REMOTE_TUNNEL_READY_SCRIPT = String.raw`set -eu
+socket=$1
+test -S "$socket"
+printf '%s\n' '${READY_MARKER}'
+trap 'exit 0' HUP INT TERM
+while :; do sleep 3600; done
+`;
+
+const REMOTE_SOCKET_CLEANUP_SCRIPT = String.raw`set -eu
+socket=$1
+directory=$2
+rm -f -- "$socket"
+rmdir -- "$directory" 2>/dev/null || true
+`;
+
+export type WorkerSshProcessExit = {
+ code: number | null;
+ signal: NodeJS.Signals | null;
+};
+
+export type WorkerSshProcess = {
+ ready: Promise;
+ exited: Promise;
+ stop(): Promise;
+};
+
+export type WorkerSshRunner = {
+ start(argv: string[], options: CommandOptions): WorkerSshProcess;
+ run(argv: string[], options: CommandOptions): Promise;
+};
+
+export type WorkerTunnelStartRequest = WorkerTunnelRequest & {
+ ssh: WorkerSshEndpoint;
+ resolveIdentity: WorkerSshIdentityResolver;
+};
+
+type TunnelEntry = {
+ environmentId: string;
+ ownerEpoch: number;
+ gateway: WorkerTunnelStartRequest["gateway"];
+ remoteDirectory: string;
+ remoteSocketPath: string;
+ abortController: AbortController;
+ status: Exclude;
+ prepared?: PreparedWorkerSsh;
+ process?: WorkerSshProcess;
+ initialization?: Promise;
+ loop?: Promise;
+ stopPromise?: Promise;
+ ready: Promise;
+ resolveReady: (handle: WorkerTunnelHandle) => void;
+ rejectReady: (error: Error) => void;
+ readySettled: boolean;
+ workspaceTasks: Set>;
+};
+
+export type WorkerTunnelManagerOptions = {
+ runner?: WorkerSshRunner;
+ sleep?: (ms: number, signal?: AbortSignal) => Promise;
+ backoff?: BackoffPolicy;
+ now?: () => number;
+ stableConnectionMs?: number;
+};
+
+function processError(stderr: string): Error {
+ const detail = redactSensitiveText(stderr, { mode: "tools" }).replace(/\s+/gu, " ").trim();
+ return new Error(detail ? `Worker SSH tunnel failed: ${detail}` : "Worker SSH tunnel failed");
+}
+
+/** Production runner that treats the remote post-forward marker as connection readiness. */
+export function createWorkerSshRunner(): WorkerSshRunner {
+ return {
+ run: runCommandWithTimeout,
+ start(argv, options) {
+ const [command, ...args] = argv;
+ if (!command) {
+ throw new Error("Worker SSH runner requires a command");
+ }
+ const child = spawn(command, args, {
+ env: options.baseEnv,
+ stdio: ["pipe", "pipe", "pipe"],
+ windowsHide: true,
+ });
+ let closed = false;
+ let readySettled = false;
+ let resolveReady!: () => void;
+ let rejectReady!: (error: Error) => void;
+ let resolveExited!: (exit: WorkerSshProcessExit) => void;
+ const ready = new Promise((resolve, reject) => {
+ resolveReady = resolve;
+ rejectReady = reject;
+ });
+ const exited = new Promise((resolve) => {
+ resolveExited = resolve;
+ });
+ let stdout = "";
+ let stderr = "";
+ const settleReadyError = () => {
+ if (readySettled) {
+ return;
+ }
+ readySettled = true;
+ rejectReady(processError(stderr));
+ };
+ child.stdout.setEncoding("utf8");
+ child.stdout.on("error", () => {});
+ child.stdout.on("data", (chunk: string) => {
+ if (readySettled) {
+ return;
+ }
+ stdout = `${stdout}${chunk}`.slice(-STDERR_LIMIT);
+ if (stdout.split(/\r?\n/u).includes(READY_MARKER)) {
+ readySettled = true;
+ resolveReady();
+ }
+ });
+ child.stderr.setEncoding("utf8");
+ child.stderr.on("error", () => {});
+ child.stderr.on("data", (chunk: string) => {
+ stderr = `${stderr}${chunk}`.slice(-STDERR_LIMIT);
+ });
+ child.once("error", settleReadyError);
+ child.once("close", (code, signal) => {
+ closed = true;
+ settleReadyError();
+ resolveExited({ code, signal });
+ });
+ child.stdin.on("error", () => {});
+ if (options.input !== undefined) {
+ child.stdin.end(options.input);
+ } else {
+ child.stdin.end();
+ }
+
+ let stopPromise: Promise | undefined;
+ return {
+ ready,
+ exited,
+ stop() {
+ return (stopPromise ??= (async () => {
+ if (closed) {
+ return;
+ }
+ child.kill("SIGTERM");
+ let timer: ReturnType | undefined;
+ await Promise.race([
+ exited,
+ new Promise((resolve) => {
+ timer = setTimeout(resolve, STOP_GRACE_MS);
+ timer.unref?.();
+ }),
+ ]);
+ clearTimeout(timer);
+ if (!closed) {
+ child.kill("SIGKILL");
+ await exited;
+ }
+ })());
+ },
+ };
+ },
+ };
+}
+
+function success(result: SpawnResult): boolean {
+ return result.termination === "exit" && result.code === 0;
+}
+
+function validateStartRequest(request: WorkerTunnelStartRequest): void {
+ if (!request.environmentId.trim()) {
+ throw new Error("Worker tunnel environment id must be non-empty");
+ }
+ if (!Number.isSafeInteger(request.ownerEpoch) || request.ownerEpoch < 0) {
+ throw new Error("Worker tunnel owner epoch must be a non-negative safe integer");
+ }
+ if (
+ !Number.isInteger(request.gateway.port) ||
+ request.gateway.port < 1 ||
+ request.gateway.port > 65_535
+ ) {
+ throw new Error("Worker tunnel gateway port must be an integer between 1 and 65535");
+ }
+}
+
+function remoteTargetHost(host: WorkerTunnelStartRequest["gateway"]["host"]): string {
+ return host === "::1" ? `[${host}]` : host;
+}
+
+/** Owns process-local reverse tunnels and fences all delayed work on stop or owner replacement. */
+export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = {}) {
+ const runner = options.runner ?? createWorkerSshRunner();
+ const sleep = options.sleep ?? sleepWithAbort;
+ const backoff = options.backoff ?? DEFAULT_BACKOFF;
+ const now = options.now ?? Date.now;
+ const stableConnectionMs = options.stableConnectionMs ?? DEFAULT_STABLE_CONNECTION_MS;
+ const entries = new Map();
+ const claimedOwnerEpochs = new Map();
+
+ const isCurrent = (entry: TunnelEntry) =>
+ entries.get(entry.environmentId) === entry && !entry.abortController.signal.aborted;
+
+ const sshCommand = (
+ prepared: PreparedWorkerSsh,
+ params: { input: string; remoteArgs: readonly string[]; signal?: AbortSignal },
+ ) => ({
+ argv: [
+ "ssh",
+ ...workerSshOptions(prepared, { forwarding: "disabled" as const }),
+ "-a",
+ "-x",
+ "-T",
+ "-p",
+ String(prepared.port),
+ "--",
+ prepared.sshTarget,
+ workerSshRemoteCommand(["sh", "-s", "--", ...params.remoteArgs]),
+ ],
+ options: workerSshCommandOptions({
+ input: params.input,
+ timeoutMs: REMOTE_SETUP_TIMEOUT_MS,
+ signal: params.signal,
+ }),
+ });
+
+ const prepareRemoteSocket = async (entry: TunnelEntry) => {
+ const prepared = entry.prepared;
+ if (!prepared) {
+ throw new Error("Worker tunnel SSH context is unavailable");
+ }
+ const command = sshCommand(prepared, {
+ input: REMOTE_SOCKET_SETUP_SCRIPT,
+ remoteArgs: [entry.remoteDirectory, entry.remoteSocketPath],
+ signal: entry.abortController.signal,
+ });
+ const result = await runner.run(command.argv, command.options);
+ if (!success(result)) {
+ throw processError(result.stderr || result.stdout);
+ }
+ };
+
+ const cleanupRemoteSocket = async (entry: TunnelEntry) => {
+ if (!entry.prepared) {
+ return;
+ }
+ const command = sshCommand(entry.prepared, {
+ input: REMOTE_SOCKET_CLEANUP_SCRIPT,
+ remoteArgs: [entry.remoteSocketPath, entry.remoteDirectory],
+ });
+ await runner.run(command.argv, command.options).catch(() => undefined);
+ };
+
+ const createHandle = (entry: TunnelEntry): WorkerTunnelHandle => ({
+ environmentId: entry.environmentId,
+ ownerEpoch: entry.ownerEpoch,
+ remoteSocketPath: entry.remoteSocketPath,
+ async runWorkspaceCommand(command) {
+ if (!isCurrent(entry) || !entry.prepared || entry.status !== "connected") {
+ throw new Error("Worker tunnel owner is no longer connected");
+ }
+ const task = runner.run(
+ [
+ "ssh",
+ ...workerSshOptions(entry.prepared, { forwarding: "disabled" }),
+ "-a",
+ "-x",
+ "-T",
+ "-p",
+ String(entry.prepared.port),
+ "--",
+ entry.prepared.sshTarget,
+ workerSshRemoteCommand(command.argv),
+ ],
+ workerSshCommandOptions({
+ input: command.input,
+ timeoutMs: command.timeoutMs ?? WORKSPACE_TIMEOUT_MS,
+ signal: entry.abortController.signal,
+ }),
+ );
+ entry.workspaceTasks.add(task);
+ void task.then(
+ () => entry.workspaceTasks.delete(task),
+ () => entry.workspaceTasks.delete(task),
+ );
+ return await task;
+ },
+ stop: () => stop(entry.environmentId, entry.ownerEpoch),
+ });
+
+ const connect = async (entry: TunnelEntry): Promise => {
+ const prepared = entry.prepared;
+ if (!prepared) {
+ throw new Error("Worker tunnel SSH context is unavailable");
+ }
+ await prepareRemoteSocket(entry);
+ if (!isCurrent(entry)) {
+ throw new Error("Worker tunnel owner changed during connection");
+ }
+ const target = `${remoteTargetHost(entry.gateway.host)}:${entry.gateway.port}`;
+ return runner.start(
+ [
+ "ssh",
+ ...workerSshOptions(prepared, { forwarding: "explicit" }),
+ "-a",
+ "-x",
+ "-T",
+ "-o",
+ "ServerAliveInterval=15",
+ "-o",
+ "ServerAliveCountMax=3",
+ "-o",
+ "StreamLocalBindMask=0177",
+ "-o",
+ "StreamLocalBindUnlink=yes",
+ "-R",
+ `${entry.remoteSocketPath}:${target}`,
+ "-p",
+ String(prepared.port),
+ "--",
+ prepared.sshTarget,
+ workerSshRemoteCommand(["sh", "-s", "--", entry.remoteSocketPath]),
+ ],
+ workerSshCommandOptions({
+ input: REMOTE_TUNNEL_READY_SCRIPT,
+ timeoutMs: Number.MAX_SAFE_INTEGER,
+ signal: entry.abortController.signal,
+ }),
+ );
+ };
+
+ const reconnectLoop = async (entry: TunnelEntry) => {
+ let retryAttempt = 0;
+ while (isCurrent(entry)) {
+ entry.status = retryAttempt === 0 ? "connecting" : "reconnecting";
+ let child: WorkerSshProcess | undefined;
+ try {
+ child = await connect(entry);
+ entry.process = child;
+ await child.ready;
+ if (!isCurrent(entry)) {
+ await child.stop();
+ return;
+ }
+ entry.status = "connected";
+ if (!entry.readySettled) {
+ entry.readySettled = true;
+ entry.resolveReady(createHandle(entry));
+ }
+ const connectedAtMs = now();
+ await child.exited;
+ if (now() - connectedAtMs >= stableConnectionMs) {
+ retryAttempt = 0;
+ }
+ } catch {
+ await child?.stop().catch(() => undefined);
+ } finally {
+ if (entry.process === child) {
+ entry.process = undefined;
+ }
+ }
+ if (!isCurrent(entry)) {
+ return;
+ }
+ entry.status = "reconnecting";
+ retryAttempt += 1;
+ try {
+ await sleep(computeBackoff(backoff, retryAttempt), entry.abortController.signal);
+ } catch {
+ return;
+ }
+ }
+ };
+
+ const stopEntry = (entry: TunnelEntry): Promise => {
+ if (entry.stopPromise) {
+ return entry.stopPromise;
+ }
+ entry.stopPromise = (async () => {
+ if (entries.get(entry.environmentId) === entry) {
+ entries.delete(entry.environmentId);
+ }
+ entry.abortController.abort(new Error("Worker tunnel owner stopped"));
+ if (!entry.readySettled) {
+ entry.readySettled = true;
+ entry.rejectReady(new Error("Worker tunnel stopped before connecting"));
+ }
+ await entry.process?.stop().catch(() => undefined);
+ await entry.initialization?.catch(() => undefined);
+ await entry.process?.stop().catch(() => undefined);
+ await Promise.allSettled(entry.workspaceTasks);
+ await entry.loop?.catch(() => undefined);
+ await cleanupRemoteSocket(entry);
+ await entry.prepared?.dispose().catch(() => undefined);
+ })();
+ return entry.stopPromise;
+ };
+
+ async function start(request: WorkerTunnelStartRequest): Promise {
+ validateStartRequest(request);
+ const claimedEpoch = claimedOwnerEpochs.get(request.environmentId);
+ if (claimedEpoch !== undefined && request.ownerEpoch < claimedEpoch) {
+ throw new Error("Worker tunnel owner epoch is stale");
+ }
+ claimedOwnerEpochs.set(request.environmentId, request.ownerEpoch);
+ const current = entries.get(request.environmentId);
+ if (current) {
+ if (request.ownerEpoch < current.ownerEpoch) {
+ throw new Error("Worker tunnel owner epoch is stale");
+ }
+ if (request.ownerEpoch === current.ownerEpoch) {
+ return await current.ready;
+ }
+ }
+
+ let resolveReady!: (handle: WorkerTunnelHandle) => void;
+ let rejectReady!: (error: Error) => void;
+ const ready = new Promise((resolve, reject) => {
+ resolveReady = resolve;
+ rejectReady = reject;
+ });
+ void ready.catch(() => undefined);
+ const remoteDirectory = `/tmp/ocw-${randomBytes(8).toString("hex")}`;
+ const entry: TunnelEntry = {
+ environmentId: request.environmentId,
+ ownerEpoch: request.ownerEpoch,
+ gateway: request.gateway,
+ remoteDirectory,
+ remoteSocketPath: `${remoteDirectory}/${REMOTE_SOCKET_NAME}`,
+ abortController: new AbortController(),
+ status: "connecting",
+ ready,
+ resolveReady,
+ rejectReady,
+ readySettled: false,
+ workspaceTasks: new Set(),
+ };
+ // Publish the new epoch before any teardown await. Stop/drain always sees the newest owner and
+ // can fence its initialization even while the previous epoch is still shutting down.
+ entries.set(request.environmentId, entry);
+ entry.initialization = (async () => {
+ if (current) {
+ await stopEntry(current);
+ }
+ if (!isCurrent(entry)) {
+ return;
+ }
+ entry.prepared = await prepareWorkerSsh({
+ ssh: request.ssh,
+ pinnedHostKey: request.ssh.hostKey,
+ resolveIdentity: request.resolveIdentity,
+ temporaryDirectoryPrefix: "openclaw-worker-tunnel-",
+ });
+ if (!isCurrent(entry)) {
+ await entry.prepared.dispose();
+ entry.prepared = undefined;
+ return;
+ }
+ entry.loop = reconnectLoop(entry);
+ void entry.loop.catch((error: unknown) => {
+ if (!entry.readySettled) {
+ entry.readySettled = true;
+ entry.rejectReady(error instanceof Error ? error : new Error("Worker tunnel failed"));
+ }
+ });
+ })();
+ void entry.initialization.catch((error: unknown) => {
+ if (!entry.readySettled) {
+ entry.readySettled = true;
+ entry.rejectReady(error instanceof Error ? error : new Error("Worker tunnel failed"));
+ }
+ void stopEntry(entry);
+ });
+ return await entry.ready;
+ }
+
+ async function stop(environmentId: string, ownerEpoch?: number): Promise {
+ const entry = entries.get(environmentId);
+ if (!entry || (ownerEpoch !== undefined && ownerEpoch !== entry.ownerEpoch)) {
+ return;
+ }
+ await stopEntry(entry);
+ }
+
+ async function stopAll(): Promise {
+ const current = [...entries.values()];
+ for (const entry of current) {
+ entries.delete(entry.environmentId);
+ entry.abortController.abort(new Error("Worker tunnel manager stopped"));
+ }
+ await Promise.all(current.map(stopEntry));
+ }
+
+ return {
+ start,
+ stop,
+ stopAll,
+ status(environmentId: string): WorkerTunnelStatus {
+ return entries.get(environmentId)?.status ?? "stopped";
+ },
+ };
+}
+
+export type WorkerTunnelManager = ReturnType;
diff --git a/src/plugin-sdk/plugin-entry.ts b/src/plugin-sdk/plugin-entry.ts
index 6d0c0195e1c9..06e9dcfd5217 100644
--- a/src/plugin-sdk/plugin-entry.ts
+++ b/src/plugin-sdk/plugin-entry.ts
@@ -52,6 +52,8 @@ export type WorkerLeaseStatus = import("../plugins/types.js").WorkerLeaseStatus;
export type WorkerProfile = import("../plugins/types.js").WorkerProfile;
export type WorkerProvider = import("../plugins/types.js").WorkerProvider;
export type WorkerSshEndpoint = import("../plugins/types.js").WorkerSshEndpoint;
+export type WorkerSshIdentity = import("../plugins/types.js").WorkerSshIdentity;
+export type WorkerSshIdentityRequest = import("../plugins/types.js").WorkerSshIdentityRequest;
export { WorkerProviderError } from "../plugins/types.js";
export type ProviderAugmentModelCatalogContext =
import("../plugins/types.js").ProviderAugmentModelCatalogContext;
diff --git a/src/plugins/types.ts b/src/plugins/types.ts
index 13d70f8927a3..6cb750d98f8f 100644
--- a/src/plugins/types.ts
+++ b/src/plugins/types.ts
@@ -1278,6 +1278,18 @@ export type WorkerSshEndpoint = {
keyRef: SecretRef;
};
+/** Resolved SSH client identity. Providers may return a local path or ephemeral material. */
+export type WorkerSshIdentity =
+ | { kind: "path"; path: string }
+ | { kind: "material"; contents: string };
+
+/** Durable context supplied when a worker provider resolves the identity it minted. */
+export type WorkerSshIdentityRequest = {
+ leaseId: string;
+ profile: WorkerProfile;
+ keyRef: SecretRef;
+};
+
/** Durable lease identity and endpoint returned by a successful provision operation. */
export type WorkerLease = {
leaseId: string;
@@ -1310,6 +1322,11 @@ export type WorkerProvider = {
provision: (profile: WorkerProfile, operationId: string) => Promise;
/** Throws on transient/indeterminate failures; `unknown` means authoritative absence. */
inspect: (lease: { leaseId: string; profile: WorkerProfile }) => Promise;
+ /**
+ * Resolves provider-owned dynamic identities. When absent, the gateway uses its generic
+ * SecretRef resolver; when present, failures are authoritative and never fall back.
+ */
+ resolveSshIdentity?: (request: WorkerSshIdentityRequest) => Promise;
renew?: (leaseId: string) => Promise;
/** Idempotent; resolves only after the provider can prove teardown. */
destroy: (lease: { leaseId: string; profile: WorkerProfile }) => Promise;
diff --git a/src/plugins/worker-provider-registry.test.ts b/src/plugins/worker-provider-registry.test.ts
index 1a5b156fb7d9..b4b2845bae43 100644
--- a/src/plugins/worker-provider-registry.test.ts
+++ b/src/plugins/worker-provider-registry.test.ts
@@ -104,6 +104,23 @@ describe("worker provider registry", () => {
);
});
+ it("rejects a non-function optional SSH identity resolver", () => {
+ const pluginRegistry = createTestRegistry();
+ const provider = {
+ ...createWorkerProvider("static-ssh"),
+ resolveSshIdentity: "later",
+ } 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 resolveSshIdentity must be a function",
+ }),
+ );
+ });
+
it("rejects invalid provider ids", () => {
const pluginRegistry = createTestRegistry();
diff --git a/src/plugins/worker-provider-registry.ts b/src/plugins/worker-provider-registry.ts
index 0ca1899e9b2d..369de25f3d96 100644
--- a/src/plugins/worker-provider-registry.ts
+++ b/src/plugins/worker-provider-registry.ts
@@ -77,6 +77,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.resolveSshIdentity !== undefined &&
+ typeof provider.resolveSshIdentity !== "function"
+ ) {
+ return {
+ ok: false,
+ message: "worker provider registration resolveSshIdentity must be a function",
+ };
+ }
const id = normalizeCapabilityProviderId(provider.id);
if (!id) {
return { ok: false, message: "worker provider registration missing valid id" };