From bb1ce585149aa08b882a46c3e91fdc6b9ad0ea0b Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 17 Aug 2026 18:19:52 -0700 Subject: [PATCH] refactor(gateway): remove obsolete reverse worker tunnel (#125465) * refactor(gateway): remove reverse worker tunnel * test(gateway): align worker transport expectations * docs(gateway): clarify cloud worker ingress --- config/assertion-safety-baseline.txt | 3 +- docs/gateway/cloud-workers.md | 2 +- docs/plan/cloud-workers.md | 19 +- docs/plan/runners.md | 7 +- src/gateway/server-http-upgrades.ts | 35 -- src/gateway/server-http.ts | 5 +- src/gateway/server-lifecycle.ts | 2 - src/gateway/server-runtime-state-prepare.ts | 6 - src/gateway/server-runtime-state.ts | 49 +- src/gateway/server-transport-bridge.ts | 1 - .../server-worker-environment-startup.test.ts | 2 - .../server-worker-environment-startup.ts | 3 - src/gateway/server.preauth-hardening.test.ts | 74 +-- .../server.public-worker-ingress.test.ts | 42 +- ...ver.sessions.worker-original-order.test.ts | 41 +- src/gateway/server/ws-connection.test.ts | 54 +- src/gateway/server/ws-connection.ts | 7 +- .../message-handler.worker.test.ts | 29 +- .../server/ws-connection/worker-connection.ts | 8 +- src/gateway/server/ws-types.ts | 3 - .../environment-access.test.ts | 4 +- .../worker-environments/environment-access.ts | 10 +- .../worker-environments/node-worker-tunnel.ts | 19 +- .../service.test-support.ts | 2 - src/gateway/worker-environments/service.ts | 2 - src/gateway/worker-environments/ssh.ts | 17 - .../worker-environments/tunnel-contract.ts | 12 +- .../tunnel.test-support.ts | 24 +- .../worker-environments/tunnel.test.ts | 511 ++---------------- src/gateway/worker-environments/tunnel.ts | 445 +-------------- ...rker-turn-launcher-claim-admission.test.ts | 4 +- ...-turn-launcher-reclaimed-placement.test.ts | 70 ++- .../worker-turn-launcher.ts | 3 + .../workspace-sync-reconnect.test.ts | 219 -------- .../workspace-sync-tunnel.test.ts | 16 +- .../worker-fault-injection.test-support.ts | 2 +- src/worker/worker.fault-injection.test.ts | 6 +- 37 files changed, 207 insertions(+), 1551 deletions(-) delete mode 100644 src/gateway/worker-environments/workspace-sync-reconnect.test.ts diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index 94c57af168c6..f77ace3e082a 100644 --- a/config/assertion-safety-baseline.txt +++ b/config/assertion-safety-baseline.txt @@ -3083,8 +3083,7 @@ src/gateway/server-reload-utils.ts 1 src/gateway/server-resident-registry.ts 1 src/gateway/server-restart-sentinel-agent-delivery.ts 3 src/gateway/server-runtime-handles.ts 11 -src/gateway/server-runtime-state-prepare.ts 3 -src/gateway/server-runtime-state.ts 1 +src/gateway/server-runtime-state-prepare.ts 2 src/gateway/server-session-events.ts 4 src/gateway/server-startup-bootstrap.ts 1 src/gateway/server-startup-config-helpers.ts 1 diff --git a/docs/gateway/cloud-workers.md b/docs/gateway/cloud-workers.md index 84c51d3a7280..5bb61636fe03 100644 --- a/docs/gateway/cloud-workers.md +++ b/docs/gateway/cloud-workers.md @@ -239,7 +239,7 @@ The bundled Crabbox provider does not support Cloud Worker Desktop after node tr ## Security model -- **Closed worker ingress.** In worker-turn mode, workers speak a dedicated protocol over their authenticated node connection with a closed method allowlist — a worker cannot call operator RPCs. +- **Closed worker ingress.** In worker-turn mode, the enrolled node launches the worker child, which dials the Gateway's authenticated public worker route and speaks a dedicated protocol with a closed method allowlist — a worker cannot call operator RPCs. - **Gateway-owned tool authority.** In worker-turn mode, the Gateway projects current profile, provider, agent, group, sender, sandbox, delegation, inherited, and runtime-cap policy over the worker's fixed coding-tool catalog before every turn. The launch envelope carries only that final closed-vocabulary subset. Explicitly capped scheduled turns reuse their trusted owner-group context without sending that identity to the box or reapplying a fresh sender overlay. Tools outside the worker catalog remain unavailable; an empty result runs with no tools. - **Minted credentials, hashed at rest.** Each dispatch mints a worker credential; the Gateway stores only its hash. Credential rotation and owner-epoch fencing guarantee at most one live owner per session — a stale worker that reconnects is fenced, never merged. - **Environment-bound enrollment.** One short-lived node-only setup credential is bound to the durable environment before allocation. Its first authenticated Ed25519 device identity is recorded atomically with setup completion; replay cannot substitute an unrelated node. diff --git a/docs/plan/cloud-workers.md b/docs/plan/cloud-workers.md index 7975cafdea26..a8399ba667c6 100644 --- a/docs/plan/cloud-workers.md +++ b/docs/plan/cloud-workers.md @@ -1,15 +1,16 @@ --- -summary: Run agent sessions on ephemeral SSH-reachable machines with gateway-proxied inference and live sidebar streaming. -title: Cloud workers plan +summary: Historical design record for cloud workers before convergence onto node-backed worker turns. +title: Cloud workers historical plan read_when: - - Designing or implementing cloud worker provisioning, worker mode, or session handoff - - Changing environments.*, the worker protocol, transcript ingestion, or inference proxy RPCs - - Reviewing security posture of remote agent execution + - Reviewing the design history behind cloud worker provisioning and session placement + - Comparing the superseded SSH reverse-tunnel proposal with the current node-backed architecture --- ## Status -Proposal, revision 3. Not implemented. Direction agreed 2026-07; revision 2 incorporated adversarial review findings (dedicated worker protocol, placement/environment state machines, git-aware inbound sync, one-way v1 handoff, controlled-egress security wording). Revision 3 settles the sync ownership model (worker authors commits, gateway adopts and publishes), adds a no-git plain sync mode, fixes worker exec at full-within-box, moves internet policy to provision time, and restores agent dispatch to milestone 3. +Superseded historical proposal. The implemented architecture is documented in [Runners and execution environments](/plan/runners) and [Cloud workers](/gateway/cloud-workers): Crabbox provisions a node-backed `worker-turn` lease, the worker child dials the Gateway's authenticated public worker route, and workspace transfer uses the node channel. The former dedicated loopback listener, SSH reverse-forward carrier, and SSH-launched worker-turn path have been removed. SSH remains a separate `remote-exec` workspace transport and desktop carrier. + +The sections below preserve the pre-convergence design record and are not the current runtime contract. ## Problem @@ -94,9 +95,11 @@ 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 SSH +### 3. Historical transport proposal: everything over SSH -The gateway owns connectivity; the worker requires nothing but sshd: +This section describes the superseded carrier. Current `worker-turn` environments use authenticated node connectivity; only `remote-exec` workspace operations retain pinned SSH. + +The original proposal had the gateway own connectivity while the worker required 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 connections with the same pinned trust material so rsync cannot head-of-line-block token streams. diff --git a/docs/plan/runners.md b/docs/plan/runners.md index 8cc77a5ea839..cd0e6e7ccab6 100644 --- a/docs/plan/runners.md +++ b/docs/plan/runners.md @@ -132,12 +132,7 @@ channel. ### Worker ingress on the public endpoint (milestone 5) -Today the worker ingress is a dedicated loopback-only listener reached via -`ssh -R`; the main ingress rejects worker frames. For node runners the same -admission is exposed on a path-tagged upgrade route on the public TLS -endpoint (`connectionKind = "worker"` forced by route instead of listener). -The loopback listener stays for SSH-provisioned cloud workers until -milestone 10. +Worker admission is exposed only on a path-tagged upgrade route on the public TLS endpoint (`connectionKind = "worker"` is forced by the route). Node-hosted worker children dial that endpoint directly. The former loopback listener and SSH reverse-forward carrier were removed after Crabbox converged onto node-backed worker turns; SSH remains only for `remote-exec` workspace transport and separately owned desktop tunnels. Hardening that ships with the exposure, not after it: diff --git a/src/gateway/server-http-upgrades.ts b/src/gateway/server-http-upgrades.ts index 9784218c8a27..8626e6864588 100644 --- a/src/gateway/server-http-upgrades.ts +++ b/src/gateway/server-http-upgrades.ts @@ -41,8 +41,6 @@ import type { PreauthConnectionBudget } from "./server/preauth-connection-budget import { markPublicWorkerIngress } from "./server/public-worker-ingress-context.js"; import { GATEWAY_WS_CONNECTION_KIND_PROPERTY, - GATEWAY_WS_PREAUTH_BUDGET_PROPERTY, - GATEWAY_WS_WORKER_INGRESS_PROPERTY, type GatewayIngressWebSocket, type GatewayWsClient, } from "./server/ws-types.js"; @@ -269,7 +267,6 @@ export function attachGatewayUpgradeHandler(opts: { ingressName: "Worker", prepareSocket: (workerSocket) => { workerSocket[GATEWAY_WS_CONNECTION_KIND_PROPERTY] = "worker"; - workerSocket[GATEWAY_WS_WORKER_INGRESS_PROPERTY] = "public"; markPublicWorkerIngress(workerSocket, { clientIp: requestClientIp, rateLimiter: publicRateLimiter, @@ -442,35 +439,3 @@ export function attachGatewayUpgradeHandler(opts: { }); }); } - -/** Attach the loopback-only worker ingress and force every accepted socket into worker mode. */ -export function attachWorkerGatewayUpgradeHandler(params: { - httpServer: HttpServer; - wss: WebSocketServer; - preauthConnectionBudget: PreauthConnectionBudget; - log?: { warn: (message: string) => void }; -}): void { - params.httpServer.on("upgrade", (req, socket, head) => { - try { - handleBudgetedGatewayWebSocketUpgrade({ - req, - socket, - head, - wss: params.wss, - preauthConnectionBudget: params.preauthConnectionBudget, - preauthBudgetKey: req.socket.remoteAddress, - ingressName: "Worker", - prepareSocket: (workerSocket) => { - workerSocket[GATEWAY_WS_CONNECTION_KIND_PROPERTY] = "worker"; - workerSocket[GATEWAY_WS_PREAUTH_BUDGET_PROPERTY] = params.preauthConnectionBudget; - workerSocket[GATEWAY_WS_WORKER_INGRESS_PROPERTY] = "loopback"; - }, - }); - } catch (error) { - params.log?.warn( - `worker websocket upgrade failed: ${error instanceof Error ? error.message : String(error)}`, - ); - socket.destroy(); - } - }); -} diff --git a/src/gateway/server-http.ts b/src/gateway/server-http.ts index 27c23b4cf05c..3a925bec0f7c 100644 --- a/src/gateway/server-http.ts +++ b/src/gateway/server-http.ts @@ -645,7 +645,4 @@ export function createGatewayHttpServer(opts: { return httpServer; } -export { - attachGatewayUpgradeHandler, - attachWorkerGatewayUpgradeHandler, -} from "./server-http-upgrades.js"; +export { attachGatewayUpgradeHandler } from "./server-http-upgrades.js"; diff --git a/src/gateway/server-lifecycle.ts b/src/gateway/server-lifecycle.ts index baf18d10ad4d..6f252f47472c 100644 --- a/src/gateway/server-lifecycle.ts +++ b/src/gateway/server-lifecycle.ts @@ -51,7 +51,6 @@ export async function prepareGatewayLifecycle(params: { const { runtime, port, log, logCron, diagnosticsEnabled, shutdownRuntime } = params; const { minimalTestGateway, - workerGatewayEndpoint, transportBridge, sessionMessageSubscribers, isConnectionActive, @@ -88,7 +87,6 @@ export async function prepareGatewayLifecycle(params: { bindDeviceNodeControl, workerPlacementRuntime, } = runtime; - workerGatewayEndpoint.resolve = transportBridge.getWorkerIngressEndpoint; const subscribeSessionMessageEvents: GatewayRequestContext["subscribeSessionMessageEvents"] = ( connId, sessionKey, diff --git a/src/gateway/server-runtime-state-prepare.ts b/src/gateway/server-runtime-state-prepare.ts index 8d8d7d386079..9ace248fb10f 100644 --- a/src/gateway/server-runtime-state-prepare.ts +++ b/src/gateway/server-runtime-state-prepare.ts @@ -122,9 +122,6 @@ export async function prepareGatewayKernelState(params: { !(nodeCommandConfig?.deny ?? []).some( (command) => command.trim() === NODE_DESKTOP_STREAM_COMMAND, ); - const workerGatewayEndpoint = { - resolve: (() => undefined) as () => { host: "127.0.0.1" | "::1"; port: number } | undefined, - }; const desktopSessionRegistry = shouldStartWorkerEnvironmentService || hostDesktopEnabled || nodeDesktopObserveAvailable ? createDesktopSessionRegistry() @@ -155,7 +152,6 @@ export async function prepareGatewayKernelState(params: { const workerModule = await loadWorkerEnvironmentStartupModule(); return await workerModule.createGatewayWorkerEnvironmentRuntime({ getPluginRegistry: () => pluginRuntime.registry, - resolveWorkerGateway: () => workerGatewayEndpoint.resolve(), desktopSessionRegistry, startup: workerEnvironmentStartup, log, @@ -573,11 +569,9 @@ export async function prepareGatewayKernelState(params: { sessionEventSubscribers, sessionMessageSubscribers, isConnectionActive, - getWorkerIngressEndpoint: transportBridge.getWorkerIngressEndpoint, getTailscaleIngressEndpoint: transportBridge.getTailscaleIngressEndpoint, getMcpAppSandboxPort: transportBridge.getMcpAppSandboxPort, ensureSandboxHostPort: transportBridge.ensureSandboxHostPort, getPortalService: transportBridge.getPortalService, - workerGatewayEndpoint, }; } diff --git a/src/gateway/server-runtime-state.ts b/src/gateway/server-runtime-state.ts index 62be452627e5..a1970c1af449 100644 --- a/src/gateway/server-runtime-state.ts +++ b/src/gateway/server-runtime-state.ts @@ -1,12 +1,6 @@ // Gateway HTTP/WebSocket runtime state factory. // Builds one server runtime with lazy plugin route handlers. -import { - createServer as createHttpServer, - type IncomingMessage, - type Server as HttpServer, - type ServerResponse, -} from "node:http"; -import type { AddressInfo } from "node:net"; +import type { IncomingMessage, Server as HttpServer, ServerResponse } from "node:http"; import type { Duplex } from "node:stream"; import { WebSocketServer } from "ws"; import { resolveSandboxHostPort } from "../agents/sandbox-host.js"; @@ -33,11 +27,7 @@ import { createSandboxHostHttpServer } from "./mcp-app-sandbox-http.js"; import { isLoopbackHost, resolveGatewayListenHosts } from "./net.js"; import { createGatewayPortalService, type GatewayPortalService } from "./portals/portal-service.js"; import { MAX_PREAUTH_PAYLOAD_BYTES } from "./server-constants.js"; -import { - attachGatewayUpgradeHandler, - attachWorkerGatewayUpgradeHandler, - createGatewayHttpServer, -} from "./server-http.js"; +import { attachGatewayUpgradeHandler, createGatewayHttpServer } from "./server-http.js"; import type { GatewayRequestContext } from "./server-methods/types.js"; import type { HookClientIpConfig, HooksRequestHandler } from "./server/hooks-request-handler.js"; import { listenGatewayHttpServer } from "./server/http-listen.js"; @@ -149,7 +139,6 @@ export async function createGatewayHttpTransport(params: { wss: WebSocketServer; preauthConnectionBudget: PreauthConnectionBudget; portalService: GatewayPortalService; - getWorkerIngressEndpoint: () => { host: "127.0.0.1"; port: number } | undefined; getTailscaleIngressEndpoint: () => GatewayTailscaleIngressEndpoint | undefined; getMcpAppSandboxPort: () => number | undefined; ensureSandboxHostPort: () => Promise; @@ -285,7 +274,6 @@ export async function createGatewayHttpTransport(params: { maxPayload: MAX_PREAUTH_PAYLOAD_BYTES, }); const preauthConnectionBudget = createPreauthConnectionBudget(); - const workerPreauthConnectionBudget = createPreauthConnectionBudget(); const httpServers: HttpServer[] = []; const gatewayHttpServers: HttpServer[] = []; @@ -371,21 +359,6 @@ export async function createGatewayHttpTransport(params: { httpServers.push(tailscaleHttpServer); } let tailscaleIngressEndpoint: GatewayTailscaleIngressEndpoint | undefined; - let workerIngressPort: number | undefined; - const workerHttpServer = params.workerIngressEnabled - ? createHttpServer((_req, res) => { - res.statusCode = 404; - res.end("Not Found"); - }) - : undefined; - if (workerHttpServer) { - attachWorkerGatewayUpgradeHandler({ - httpServer: workerHttpServer, - wss, - preauthConnectionBudget: workerPreauthConnectionBudget, - log: params.log, - }); - } const httpServer = gatewayHttpServers[0]; if (!httpServer) { throw new Error("Gateway HTTP server failed to start"); @@ -541,20 +514,6 @@ export async function createGatewayHttpTransport(params: { if (params.cfg.mcp?.apps?.enabled === true) { await startSandboxHost(); } - if (workerHttpServer) { - await listenGatewayHttpServer({ - httpServer: workerHttpServer, - bindHost: "127.0.0.1", - port: 0, - retryEaddrinuse: false, - }); - const address = workerHttpServer.address() as AddressInfo | null; - if (!address || typeof address === "string") { - throw new Error("Worker gateway ingress failed to resolve its loopback port"); - } - workerIngressPort = address.port; - httpServers.push(workerHttpServer); - } startListeningComplete = true; })(); await startListeningPromise; @@ -567,10 +526,6 @@ export async function createGatewayHttpTransport(params: { wss, preauthConnectionBudget, portalService, - getWorkerIngressEndpoint: () => - workerIngressPort === undefined - ? undefined - : { host: "127.0.0.1" as const, port: workerIngressPort }, getTailscaleIngressEndpoint: () => tailscaleIngressEndpoint, getMcpAppSandboxPort: () => mcpAppSandboxPort, ensureSandboxHostPort, diff --git a/src/gateway/server-transport-bridge.ts b/src/gateway/server-transport-bridge.ts index b77ed1e31d29..578f2d085adc 100644 --- a/src/gateway/server-transport-bridge.ts +++ b/src/gateway/server-transport-bridge.ts @@ -12,7 +12,6 @@ export function createGatewayTransportBridge() { }, current: () => current, getPortalService: () => current?.portalService, - getWorkerIngressEndpoint: () => current?.getWorkerIngressEndpoint(), getTailscaleIngressEndpoint: () => current?.getTailscaleIngressEndpoint(), getMcpAppSandboxPort: () => current?.getMcpAppSandboxPort(), ensureSandboxHostPort: async () => { diff --git a/src/gateway/server-worker-environment-startup.test.ts b/src/gateway/server-worker-environment-startup.test.ts index fada217770d7..74faef038728 100644 --- a/src/gateway/server-worker-environment-startup.test.ts +++ b/src/gateway/server-worker-environment-startup.test.ts @@ -34,7 +34,6 @@ describe("gateway worker environment startup", () => { const startup = await loadGatewayWorkerEnvironmentStartupState(); const runtime = await createGatewayWorkerEnvironmentRuntime({ getPluginRegistry: () => ({ workerProviders: new Map() }), - resolveWorkerGateway: () => undefined, desktopSessionRegistry: createDesktopSessionRegistry({ lingerMs: 1 }), startup, log: { child: () => ({ warn: () => {} }) }, @@ -95,7 +94,6 @@ describe("gateway worker environment startup", () => { const runtime = await createGatewayWorkerEnvironmentRuntime({ getPluginRegistry: () => ({ workerProviders: new Map() }), - resolveWorkerGateway: () => undefined, desktopSessionRegistry: createDesktopSessionRegistry({ lingerMs: 1 }), startup, log: { child: () => ({ warn: () => {} }) }, diff --git a/src/gateway/server-worker-environment-startup.ts b/src/gateway/server-worker-environment-startup.ts index 4de03f6274e7..efe6a0667391 100644 --- a/src/gateway/server-worker-environment-startup.ts +++ b/src/gateway/server-worker-environment-startup.ts @@ -34,7 +34,6 @@ type WorkerEnvironmentStore = ReturnType< typeof import("./worker-environments/store.js").createWorkerEnvironmentStore >; type WorkerEnvironmentRecord = ReturnType[number]; -type WorkerGatewayEndpoint = { host: "127.0.0.1" | "::1"; port: number } | undefined; type WorkerEnvironmentLogger = { child: (name: string) => { warn: (message: string) => void }; }; @@ -105,7 +104,6 @@ export async function loadGatewayWorkerEnvironmentStartupState(): Promise Pick; - resolveWorkerGateway: () => WorkerGatewayEndpoint; desktopSessionRegistry: DesktopSessionRegistry; startup: GatewayWorkerEnvironmentStartupState; log: WorkerEnvironmentLogger; @@ -258,7 +256,6 @@ export async function createGatewayWorkerEnvironmentRuntime(params: { tunnelManager: workerTunnelManager, nodeTunnelManager: nodeWorkerTunnelManager, stopNodeWorkerBundleTransfers: () => nodeWorkerBundleTransfer.closeAll(), - resolveWorkerGateway: params.resolveWorkerGateway, applyTranscriptCommit: createWorkerTranscriptCommitter({ getConfig: getRuntimeConfig, }).commit, diff --git a/src/gateway/server.preauth-hardening.test.ts b/src/gateway/server.preauth-hardening.test.ts index f4c4f8656d64..a40425da606d 100644 --- a/src/gateway/server.preauth-hardening.test.ts +++ b/src/gateway/server.preauth-hardening.test.ts @@ -24,11 +24,7 @@ import { captureEnv, setTestEnvValue } from "../test-utils/env.js"; import { createWorkerConnection } from "../worker/worker-connection.js"; import type { ResolvedGatewayAuth } from "./auth.js"; import { MAX_PREAUTH_PAYLOAD_BYTES } from "./server-constants.js"; -import { - attachGatewayUpgradeHandler, - attachWorkerGatewayUpgradeHandler, - createGatewayHttpServer, -} from "./server-http.js"; +import { attachGatewayUpgradeHandler, createGatewayHttpServer } from "./server-http.js"; import { createPreauthConnectionBudget } from "./server/preauth-connection-budget.js"; import { attachGatewayWsConnectionHandler } from "./server/ws-connection.js"; import { @@ -39,7 +35,6 @@ import type { WorkerConnectionService } from "./server/ws-connection/worker-conn import { GATEWAY_WS_CONNECTION_KIND_PROPERTY, GATEWAY_WS_PREAUTH_BUDGET_PROPERTY, - GATEWAY_WS_WORKER_INGRESS_PROPERTY, type GatewayIngressWebSocket, type GatewayWsClient, } from "./server/ws-types.js"; @@ -133,50 +128,6 @@ async function expectIdlePreauthSocketClose() { } describe("gateway pre-auth hardening", () => { - it("tags worker-only upgrades with the trusted ingress kind and budget", async () => { - const httpServer = http.createServer(); - const wss = new WebSocketServer({ maxPayload: 1024, noServer: true }); - const workerBudget = createPreauthConnectionBudget(1); - const accepted = new Promise((resolve) => { - wss.once("connection", (socket) => { - resolve(socket as GatewayIngressWebSocket); - }); - }); - attachWorkerGatewayUpgradeHandler({ - httpServer, - wss, - preauthConnectionBudget: workerBudget, - }); - await new Promise((resolve) => { - httpServer.listen(0, "127.0.0.1", resolve); - }); - const address = httpServer.address(); - const port = typeof address === "object" && address ? address.port : 0; - const client = new WebSocket(`ws://127.0.0.1:${port}`); - - try { - await new Promise((resolve, reject) => { - client.once("open", resolve); - client.once("error", reject); - }); - const socket = await accepted; - expect(socket[GATEWAY_WS_CONNECTION_KIND_PROPERTY]).toBe("worker"); - expect(socket[GATEWAY_WS_PREAUTH_BUDGET_PROPERTY]).toBe(workerBudget); - expect(socket[GATEWAY_WS_WORKER_INGRESS_PROPERTY]).toBe("loopback"); - } finally { - client.close(); - await new Promise((resolve) => { - client.once("close", () => resolve()); - }); - await new Promise((resolve) => { - wss.close(() => resolve()); - }); - await new Promise((resolve, reject) => { - httpServer.close((error) => (error ? reject(error) : resolve())); - }); - } - }); - it("reserves the public worker path before plugin upgrade routing", async () => { const clients = new Set(); const resolvedAuth: ResolvedGatewayAuth = { mode: "none", allowTailscale: false }; @@ -217,7 +168,6 @@ describe("gateway pre-auth hardening", () => { }); const socket = await accepted; expect(socket[GATEWAY_WS_CONNECTION_KIND_PROPERTY]).toBe("worker"); - expect(socket[GATEWAY_WS_WORKER_INGRESS_PROPERTY]).toBe("public"); expect(socket[GATEWAY_WS_PREAUTH_BUDGET_PROPERTY]).toBeUndefined(); expect(pluginUpgrade).not.toHaveBeenCalled(); } finally { @@ -443,14 +393,26 @@ describe("gateway pre-auth hardening", () => { } }); - it("rejects worker websocket upgrades after suspension is prepared", async () => { - const httpServer = http.createServer(); + it("rejects public worker websocket upgrades after suspension is prepared", async () => { + const clients = new Set(); + const resolvedAuth: ResolvedGatewayAuth = { mode: "none", allowTailscale: false }; + const httpServer = createGatewayHttpServer({ + clients, + controlUiEnabled: false, + controlUiBasePath: "/__control__", + openAiChatCompletionsEnabled: false, + openResponsesEnabled: false, + handleHooksRequest: async () => false, + resolvedAuth, + }); const wss = new WebSocketServer({ maxPayload: 1024, noServer: true }); - wss.on("connection", (socket) => socket.close()); - attachWorkerGatewayUpgradeHandler({ + attachGatewayUpgradeHandler({ httpServer, wss, + clients, preauthConnectionBudget: createPreauthConnectionBudget(1), + resolvedAuth, + workerIngressEnabled: true, }); await new Promise((resolve) => { httpServer.listen(0, "127.0.0.1", resolve); @@ -461,7 +423,7 @@ describe("gateway pre-auth hardening", () => { expect(suspension?.commit()).toBe(true); try { - await expect(requestUpgradeRejection(port)).resolves.toEqual({ + await expect(requestUpgradeRejection(port, WORKER_PUBLIC_INGRESS_PATH)).resolves.toEqual({ status: 503, body: "Worker websocket admission closed", }); diff --git a/src/gateway/server.public-worker-ingress.test.ts b/src/gateway/server.public-worker-ingress.test.ts index f28048b4735c..2c5c7a84d57a 100644 --- a/src/gateway/server.public-worker-ingress.test.ts +++ b/src/gateway/server.public-worker-ingress.test.ts @@ -14,11 +14,7 @@ import { } from "../../packages/gateway-protocol/src/index.js"; import { createAuthRateLimiter } from "./auth-rate-limit.js"; import type { ResolvedGatewayAuth } from "./auth.js"; -import { - attachGatewayUpgradeHandler, - attachWorkerGatewayUpgradeHandler, - createGatewayHttpServer, -} from "./server-http.js"; +import { attachGatewayUpgradeHandler, createGatewayHttpServer } from "./server-http.js"; import { createPreauthConnectionBudget } from "./server/preauth-connection-budget.js"; import { attachGatewayWsConnectionHandler } from "./server/ws-connection.js"; import { createGatewayWsTestLogger } from "./server/ws-connection.test-helpers.js"; @@ -404,42 +400,6 @@ describe("public worker ingress", () => { }); }); - it("shares the gateway preauth budget without affecting loopback worker ingress", async () => { - await withHarness({ preauthLimit: 1 }, async (harness) => { - const publicWorker = new WebSocket(harness.url()); - await waitForOpen(publicWorker); - - await expect(requestUpgradeRejection(harness.port, "/")).resolves.toEqual({ - status: 503, - body: "Too many unauthenticated sockets", - }); - - const loopbackServer = http.createServer(); - attachWorkerGatewayUpgradeHandler({ - httpServer: loopbackServer, - wss: harness.wss, - preauthConnectionBudget: createPreauthConnectionBudget(1), - }); - await new Promise((resolve) => { - loopbackServer.listen(0, "127.0.0.1", resolve); - }); - const loopbackPort = (loopbackServer.address() as AddressInfo).port; - const loopbackWorker = new WebSocket(`ws://127.0.0.1:${loopbackPort}`); - try { - await waitForOpen(loopbackWorker); - } finally { - const loopbackClose = waitForClose(loopbackWorker); - const publicClose = waitForClose(publicWorker); - loopbackWorker.close(); - publicWorker.close(); - await Promise.all([loopbackClose, publicClose]); - await new Promise((resolve, reject) => { - loopbackServer.close((error) => (error ? reject(error) : resolve())); - }); - } - }); - }); - it("rate-limits parallel invalid admissions before repeated store work", async () => { await withHarness({ rateLimitMaxAttempts: 2 }, async (harness) => { const sockets = Array.from({ length: 6 }, () => new WebSocket(harness.url())); diff --git a/src/gateway/server.sessions.worker-original-order.test.ts b/src/gateway/server.sessions.worker-original-order.test.ts index 8a6eb0a739bb..d6ea4efda160 100644 --- a/src/gateway/server.sessions.worker-original-order.test.ts +++ b/src/gateway/server.sessions.worker-original-order.test.ts @@ -28,7 +28,7 @@ import { type WorkerEnvironmentService, } from "./worker-environments/service.js"; import { createWorkerEnvironmentStore } from "./worker-environments/store.js"; -import type { WorkerSshProcess, WorkerSshRunner } from "./worker-environments/tunnel-ssh-runner.js"; +import type { WorkerSshRunner } from "./worker-environments/tunnel-ssh-runner.js"; import { createWorkerTunnelManager } from "./worker-environments/tunnel.js"; import { prepareLocalWorkspaceRsyncBoundary } from "./worker-environments/tunnel.test-support.js"; import { rsyncArgvPort, sshArgvPort } from "./worker-environments/worker-ssh-argv.test-support.js"; @@ -106,25 +106,8 @@ function argvPort(argv: readonly string[]): number { return port!; } -class ConnectedProcess implements WorkerSshProcess { - readonly ready = Promise.resolve(); - readonly exited: Promise<{ code: number | null; signal: NodeJS.Signals | null }>; - private resolveExit!: (exit: { code: number | null; signal: NodeJS.Signals | null }) => void; - - constructor() { - this.exited = new Promise((resolve) => { - this.resolveExit = resolve; - }); - } - - async stop(): Promise { - this.resolveExit({ code: null, signal: "SIGTERM" }); - } -} - class OriginalOrderSshRunner implements WorkerSshRunner { readonly events: string[] = []; - readonly starts: string[][] = []; private bootstrapOperationToken: string | undefined; constructor(private readonly remoteHome: string) {} @@ -149,10 +132,8 @@ class OriginalOrderSshRunner implements WorkerSshRunner { return path.join(this.remoteHome, ".openclaw-worker", BUNDLE_HASH, "bootstrap-receipt.json"); } - start(argv: string[]): WorkerSshProcess { - this.starts.push(argv); - this.events.push(`tunnel:start:${argvPort(argv)}`); - return new ConnectedProcess(); + start(): never { + throw new Error("remote-exec workspace transport must not start a persistent SSH process"); } async run(argv: string[], options: CommandOptions): Promise { @@ -186,10 +167,6 @@ class OriginalOrderSshRunner implements WorkerSshRunner { this.events.push(`bootstrap:cleanup:${port}`); return success(); } - if (argv[0] === "ssh" && input.includes("unsafe worker tunnel directory")) { - this.events.push(`tunnel:prepare:${port}`); - return port === PRIMARY_PORT ? transportFailure() : success(); - } if (argv[0] === "rsync") { this.events.push(`workspace:transfer:${port}`); if (argv.some((arg) => arg.startsWith("--rsync-path="))) { @@ -348,10 +325,7 @@ test("preserves ordered fallback through restart, workspace sync, and safe sessi database = openOpenClawStateDatabase({ env: { OPENCLAW_STATE_DIR: stateDir } }); const environmentStore = createWorkerEnvironmentStore({ database, now: () => 2_000 }); const placements = createWorkerSessionPlacementStore({ database, now: () => 3_000 }); - tunnelManager = createWorkerTunnelManager({ - runner, - backoff: { initialMs: 1, maxMs: 1, factor: 1, jitter: 0 }, - }); + tunnelManager = createWorkerTunnelManager({ runner }); const environmentService = createWorkerEnvironmentService({ store: environmentStore, getConfig: () => ({ @@ -397,7 +371,6 @@ test("preserves ordered fallback through restart, workspace sync, and safe sessi }, resolveSshIdentity: async () => ({ kind: "path", path: "/keys/worker" }), tunnelManager, - resolveWorkerGateway: () => ({ host: "127.0.0.1", port: 18_789 }), generateWorkerCredential: () => "original-order-credential", liveEvents: { apply: () => ({ ok: true, result: { ackedSeq: 1 } }), @@ -455,7 +428,6 @@ test("preserves ordered fallback through restart, workspace sync, and safe sessi executionMode: "remote-exec", }); expect(active).toMatchObject({ state: "active", environmentId: ENVIRONMENT_ID }); - expect(runner.starts).toHaveLength(1); await expect(fs.stat(runner.bootstrapUploadPath)).rejects.toMatchObject({ code: "ENOENT" }); await expect(fs.readFile(runner.bootstrapReceiptPath, "utf8")).resolves.toBe( `${JSON.stringify(RECEIPT)}\n`, @@ -502,10 +474,7 @@ test("preserves ordered fallback through restart, workspace sync, and safe sessi `bootstrap:transfer:${FALLBACK_PORT}`, `bootstrap:install:${FALLBACK_PORT}`, `bootstrap:cleanup:${FALLBACK_PORT}`, - `tunnel:prepare:${PRIMARY_PORT}`, - `tunnel:prepare:${FALLBACK_PORT}`, - `tunnel:start:${FALLBACK_PORT}`, - `workspace:transfer:${FALLBACK_PORT}`, + `workspace:transfer:${PRIMARY_PORT}`, "placement:active", "workspace:quiesce", "workspace:renew-quiescence", diff --git a/src/gateway/server/ws-connection.test.ts b/src/gateway/server/ws-connection.test.ts index db1f2eb36b36..43d4320044ab 100644 --- a/src/gateway/server/ws-connection.test.ts +++ b/src/gateway/server/ws-connection.test.ts @@ -52,11 +52,7 @@ vi.mock("../talk-session-registry.js", () => ({ import { markPublicWorkerIngress } from "./public-worker-ingress-context.js"; import { attachGatewayWsConnectionHandler } from "./ws-connection.js"; import { resolveSharedGatewaySessionGeneration } from "./ws-shared-generation.js"; -import { - GATEWAY_WS_CONNECTION_KIND_PROPERTY, - GATEWAY_WS_PREAUTH_BUDGET_PROPERTY, - GATEWAY_WS_WORKER_INGRESS_PROPERTY, -} from "./ws-types.js"; +import { GATEWAY_WS_CONNECTION_KIND_PROPERTY } from "./ws-types.js"; async function waitForLazyMessageHandler() { await vi.dynamicImportSettled(); @@ -116,7 +112,7 @@ describe("attachGatewayWsConnectionHandler", () => { vi.useRealTimers(); }); - it("keeps loopback worker sockets off the legacy challenge, plugin surface, and gateway budget", async () => { + it("keeps public worker sockets off the legacy challenge and plugin surface", async () => { const socket = createGatewayWsTestSocket(); const previous = { socket: { terminate: vi.fn() }, @@ -124,13 +120,16 @@ describe("attachGatewayWsConnectionHandler", () => { }; const clients = new Set([previous]); const gatewayBudget = { release: vi.fn() }; - const workerBudget = { release: vi.fn() }; + const rateLimiter = { check: vi.fn() }; const getPluginNodeCapabilities = vi.fn(() => [{ surface: "canvas" }]); const buildRequestContext = vi.fn(() => createGatewayWsTestRequestContext() as never); Object.assign(socket, { [GATEWAY_WS_CONNECTION_KIND_PROPERTY]: "worker", - [GATEWAY_WS_PREAUTH_BUDGET_PROPERTY]: workerBudget, - __openclawPreauthBudgetKey: "127.0.0.1", + __openclawPreauthBudgetKey: "203.0.113.10", + }); + markPublicWorkerIngress(socket as never, { + clientIp: "203.0.113.10", + rateLimiter: rateLimiter as never, }); await connectTestWs({ @@ -146,8 +145,12 @@ describe("attachGatewayWsConnectionHandler", () => { expect(socket.send).not.toHaveBeenCalled(); expect(getPluginNodeCapabilities).not.toHaveBeenCalled(); const handler = firstAttachedWorkerHandlerParams() as { + publicAdmission: { clientIp: string; rateLimiter: unknown }; setClient(client: never): boolean; }; + expect(handler).toMatchObject({ + publicAdmission: { clientIp: "203.0.113.10", rateLimiter }, + }); const client = { socket, connect: { client: { id: "openclaw-worker", mode: "worker" } }, @@ -160,39 +163,6 @@ describe("attachGatewayWsConnectionHandler", () => { expect(attachGatewayWsMessageHandlerMock).not.toHaveBeenCalled(); socket.emit("close", 1000, Buffer.alloc(0)); expect(buildRequestContext).not.toHaveBeenCalled(); - expect(workerBudget.release).toHaveBeenCalledWith("127.0.0.1"); - expect(gatewayBudget.release).not.toHaveBeenCalled(); - }); - - it("uses the main budget and public admission context for public worker sockets", async () => { - const socket = createGatewayWsTestSocket(); - const gatewayBudget = { release: vi.fn() }; - const rateLimiter = { check: vi.fn() }; - Object.assign(socket, { - [GATEWAY_WS_CONNECTION_KIND_PROPERTY]: "worker", - [GATEWAY_WS_WORKER_INGRESS_PROPERTY]: "public", - __openclawPreauthBudgetKey: "203.0.113.10", - }); - markPublicWorkerIngress(socket as never, { - clientIp: "203.0.113.10", - rateLimiter: rateLimiter as never, - }); - - await connectTestWs({ - socket, - options: { - preauthConnectionBudget: gatewayBudget as never, - }, - }); - - const handler = firstAttachedWorkerHandlerParams() as { - publicAdmission: { clientIp: string; rateLimiter: unknown }; - setClient(client: never): boolean; - }; - expect(handler).toMatchObject({ - publicAdmission: { clientIp: "203.0.113.10", rateLimiter }, - }); - expect(handler.setClient({ socket } as never)).toBe(true); expect(gatewayBudget.release).toHaveBeenCalledWith("203.0.113.10"); }); diff --git a/src/gateway/server/ws-connection.ts b/src/gateway/server/ws-connection.ts index 9063a37bbdda..a2b488c47ea8 100644 --- a/src/gateway/server/ws-connection.ts +++ b/src/gateway/server/ws-connection.ts @@ -62,10 +62,8 @@ import { resolveSharedGatewaySessionGeneration } from "./ws-shared-generation.js import { GATEWAY_WS_CONNECTION_KIND_PROPERTY, GATEWAY_WS_PREAUTH_BUDGET_PROPERTY, - GATEWAY_WS_WORKER_INGRESS_PROPERTY, WS_HANDSHAKE_PHASES, type GatewayIngressWebSocket, - type GatewayWorkerIngress, type GatewayWsClient, type WsHandshakePhase, } from "./ws-types.js"; @@ -210,10 +208,8 @@ export function attachGatewayWsConnectionHandler(params: AttachGatewayWsConnecti const connId = randomUUID(); const ingressSocket = socket as GatewayIngressWebSocket; const connectionKind = ingressSocket[GATEWAY_WS_CONNECTION_KIND_PROPERTY] ?? "gateway"; - const workerIngress: GatewayWorkerIngress = - ingressSocket[GATEWAY_WS_WORKER_INGRESS_PROPERTY] ?? "loopback"; const publicWorkerIngress = - workerIngress === "public" ? takePublicWorkerIngress(socket) : undefined; + connectionKind === "worker" ? takePublicWorkerIngress(socket) : undefined; const connectionPreauthBudget = ingressSocket[GATEWAY_WS_PREAUTH_BUDGET_PROPERTY] ?? preauthConnectionBudget; const { remoteAddr, remotePort, localAddr, localPort, endpoint } = resolveSocketAddress(socket); @@ -663,7 +659,6 @@ export function attachGatewayWsConnectionHandler(params: AttachGatewayWsConnecti connId, service: workerConnectionService, isStartupPending, - ingress: workerIngress, send, close, isClosed: () => closed, diff --git a/src/gateway/server/ws-connection/message-handler.worker.test.ts b/src/gateway/server/ws-connection/message-handler.worker.test.ts index 9ffa3988e4f5..b690d438b1af 100644 --- a/src/gateway/server/ws-connection/message-handler.worker.test.ts +++ b/src/gateway/server/ws-connection/message-handler.worker.test.ts @@ -164,7 +164,6 @@ function attachHarness( commitFailure?: WorkerTranscriptCommitErrorReason; identity?: WorkerConnectionIdentity; liveFailure?: WorkerLiveEventErrorDetails; - ingress?: "loopback" | "public"; omitPublicAdmission?: boolean; rateLimiter?: AuthRateLimiter; onInferenceLaunch?: (sink: InferenceSink) => void; @@ -232,11 +231,9 @@ function attachHarness( socket: socket as unknown as WebSocket, connId: "worker-connection", service, - ingress: options.ingress ?? "loopback", - publicAdmission: - options.ingress === "public" && !options.omitPublicAdmission - ? { clientIp: "203.0.113.10", rateLimiter: options.rateLimiter } - : undefined, + publicAdmission: options.omitPublicAdmission + ? undefined + : { clientIp: "203.0.113.10", rateLimiter: options.rateLimiter }, send: (frame) => responses.push(frame), close, isClosed: () => false, @@ -300,8 +297,13 @@ describe("dedicated worker websocket protocol", () => { const harness = attachHarness({ admissionFailure: reason }); harness.sendConnect(); - await waitForWorkerProtocol(() => expect(harness.close).toHaveBeenCalledWith(1008, reason)); - expect(harness.responses[0]).toMatchObject({ ok: false, error: { details: { reason } } }); + await waitForWorkerProtocol(() => + expect(harness.close).toHaveBeenCalledWith(1008, "invalid-handshake"), + ); + expect(harness.responses[0]).toMatchObject({ + ok: false, + error: { details: { reason: "invalid-handshake" } }, + }); expect(harness.logWsControl.warn).toHaveBeenCalledWith( `worker admission rejected reason=${reason}`, ); @@ -309,7 +311,7 @@ describe("dedicated worker websocket protocol", () => { }); it("fails closed when public ingress context is missing", async () => { - const harness = attachHarness({ ingress: "public", omitPublicAdmission: true }); + const harness = attachHarness({ omitPublicAdmission: true }); harness.sendConnect(); await waitForWorkerProtocol(() => @@ -328,7 +330,6 @@ describe("dedicated worker websocket protocol", () => { const rateLimiter = createRateLimiter({ recordFailure }); const harness = attachHarness({ admissionFailure: internalReason, - ingress: "public", rateLimiter, }); harness.sendConnect(); @@ -352,7 +353,7 @@ describe("dedicated worker websocket protocol", () => { const rateLimiter = createRateLimiter({ check: vi.fn(() => ({ allowed: false, remaining: 0, retryAfterMs: 12_000 })), }); - const harness = attachHarness({ ingress: "public", rateLimiter }); + const harness = attachHarness({ rateLimiter }); harness.sendConnect(); await waitForWorkerProtocol(() => @@ -371,7 +372,7 @@ describe("dedicated worker websocket protocol", () => { it("resets public credential failures after successful admission", async () => { const reset = vi.fn(); const rateLimiter = createRateLimiter({ reset }); - const harness = attachHarness({ ingress: "public", rateLimiter }); + const harness = attachHarness({ rateLimiter }); await admit(harness); expect(reset).toHaveBeenCalledWith("203.0.113.10", "worker-admission"); @@ -382,7 +383,6 @@ describe("dedicated worker websocket protocol", () => { const recordFailure = vi.fn(); const rateLimiter = createRateLimiter({ reset, recordFailure }); const harness = attachHarness({ - ingress: "public", rateLimiter, validationFailure: "credential-replaced", }); @@ -683,8 +683,9 @@ describe("dedicated worker websocket protocol", () => { harness.sendConnect(); await waitForWorkerProtocol(() => - expect(harness.close).toHaveBeenCalledWith(1008, "credential-replaced"), + expect(harness.close).toHaveBeenCalledWith(1008, "invalid-handshake"), ); + expect(harness.setCloseCause).toHaveBeenCalledWith("credential-replaced"); expect(harness.setClient).not.toHaveBeenCalled(); }); diff --git a/src/gateway/server/ws-connection/worker-connection.ts b/src/gateway/server/ws-connection/worker-connection.ts index 8e27d7bd1df7..c852eb0f986e 100644 --- a/src/gateway/server/ws-connection/worker-connection.ts +++ b/src/gateway/server/ws-connection/worker-connection.ts @@ -58,7 +58,7 @@ import { AUTH_RATE_LIMIT_SCOPE_WORKER_ADMISSION } from "../../auth-rate-limit.js import type { WorkerConnectionIdentity } from "../../worker-environments/connection-identity.js"; import { MAX_RUNNING_WORKER_SESSION_TOOL_OPERATIONS } from "../../worker-environments/placement-session-tool-operations.js"; import type { PublicWorkerIngressContext } from "../public-worker-ingress-context.js"; -import type { GatewayWorkerIngress, GatewayWsClient, WsHandshakePhase } from "../ws-types.js"; +import type { GatewayWsClient, WsHandshakePhase } from "../ws-types.js"; import { runWorkerAdmissionBoundary } from "./worker-admission-boundary.js"; import { buildWorkerHello, @@ -141,7 +141,6 @@ type WorkerWsMessageHandlerParams = { connId: string; service?: WorkerConnectionService; isStartupPending?: () => boolean; - ingress: GatewayWorkerIngress; send(frame: unknown): void; close(code?: number, reason?: string): void; isClosed(): boolean; @@ -399,8 +398,7 @@ export function attachWorkerWsMessageHandler(params: WorkerWsMessageHandlerParam }) => { const internalReason = rejection.internalReason ?? rejection.reason; const wireReason: WorkerProtocolCloseReason = - (rejection.opaqueOnPublicIngress && params.publicAdmission) || - rejection.reason === "rate-limited" + rejection.opaqueOnPublicIngress || rejection.reason === "rate-limited" ? "invalid-handshake" : rejection.reason; const wireError = @@ -430,7 +428,7 @@ export function attachWorkerWsMessageHandler(params: WorkerWsMessageHandlerParam }); return; } - if (params.ingress === "public" && !params.publicAdmission) { + if (!params.publicAdmission) { rejectAdmission({ id, reason: "invalid-handshake", diff --git a/src/gateway/server/ws-types.ts b/src/gateway/server/ws-types.ts index 9c8f4e9f5831..d7c5100ea982 100644 --- a/src/gateway/server/ws-types.ts +++ b/src/gateway/server/ws-types.ts @@ -7,15 +7,12 @@ import type { WorkerConnectionIdentity } from "../worker-environments/connection export const GATEWAY_WS_CONNECTION_KIND_PROPERTY = "__openclawConnectionKind"; export const GATEWAY_WS_PREAUTH_BUDGET_PROPERTY = "__openclawPreauthBudget"; -export const GATEWAY_WS_WORKER_INGRESS_PROPERTY = "__openclawWorkerIngress"; type GatewayWsConnectionKind = "gateway" | "worker"; -export type GatewayWorkerIngress = "loopback" | "public"; export type GatewayIngressWebSocket = WebSocket & { [GATEWAY_WS_CONNECTION_KIND_PROPERTY]?: GatewayWsConnectionKind; [GATEWAY_WS_PREAUTH_BUDGET_PROPERTY]?: { release(clientIp: string | undefined): void; }; - [GATEWAY_WS_WORKER_INGRESS_PROPERTY]?: GatewayWorkerIngress; __openclawPreauthBudgetClaimed?: boolean; __openclawPreauthBudgetKey?: string; }; diff --git a/src/gateway/worker-environments/environment-access.test.ts b/src/gateway/worker-environments/environment-access.test.ts index 1897b6829011..9b6d967e400f 100644 --- a/src/gateway/worker-environments/environment-access.test.ts +++ b/src/gateway/worker-environments/environment-access.test.ts @@ -13,7 +13,7 @@ type WorkerEnvironmentServiceError = support.WorkerEnvironmentServiceError; describe("worker environment service", () => { support.setupWorkerEnvironmentServiceSuite(); - it("projects live tunnel status and fences the tunnel before provider teardown", async () => { + it("projects live workspace transport status and fences it before provider teardown", async () => { support.seedReady("worker-tunnel", undefined, true); const order: string[] = []; let tunnelStatus: "stopped" | "connected" = "stopped"; @@ -24,7 +24,6 @@ describe("worker environment service", () => { return { environmentId: request.environmentId, ownerEpoch: request.ownerEpoch, - launchTurn: vi.fn(), runWorkspaceCommand: vi.fn(), syncWorkspace: vi.fn(), stop: async () => {}, @@ -55,7 +54,6 @@ describe("worker environment service", () => { expect(tunnelManager.start).toHaveBeenCalledWith( expect.objectContaining({ bundleHash: support.BUNDLE_HASH, - gateway: { host: "127.0.0.1", port: 18_789 }, sharedHost: true, }), ); diff --git a/src/gateway/worker-environments/environment-access.ts b/src/gateway/worker-environments/environment-access.ts index 9544314e2a7a..052f07f0dc73 100644 --- a/src/gateway/worker-environments/environment-access.ts +++ b/src/gateway/worker-environments/environment-access.ts @@ -22,7 +22,6 @@ type WorkerEnvironmentAccessOptions = { prepareCurrentBundle: () => Promise; tunnelManager?: WorkerTunnelManager; nodeTunnelManager?: NodeWorkerTunnelManager; - resolveWorkerGateway?: () => { host: "127.0.0.1" | "::1"; port: number } | undefined; now: () => number; identityResolverFor: ( record: WorkerEnvironmentRecord, @@ -161,17 +160,12 @@ export function createWorkerEnvironmentAccess(options: WorkerEnvironmentAccessOp if (!tunnels) { throw serviceError("invalid_state", "Worker SSH tunnel runtime is unavailable"); } - const gateway = options.resolveWorkerGateway?.(); - if (!gateway) { - throw serviceError("invalid_state", "Worker gateway ingress is unavailable"); - } const provider = providerFor(record.providerId); - // 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. + // Workspace ownership is registered synchronously by the manager. Release the durable-state + // lock while SSH identity material is prepared so drain/destroy can fence initialization. startup = tunnels.start({ ...request, bundleHash: currentBundle.bundleHash, - gateway, ssh: record.sshEndpoint, sharedHost: record.sharedHost, resolveIdentity: identityResolverFor(record, provider, record.leaseId), diff --git a/src/gateway/worker-environments/node-worker-tunnel.ts b/src/gateway/worker-environments/node-worker-tunnel.ts index ab4d335af8ea..4f6348de8989 100644 --- a/src/gateway/worker-environments/node-worker-tunnel.ts +++ b/src/gateway/worker-environments/node-worker-tunnel.ts @@ -30,8 +30,9 @@ import type { NodeWorkspaceTransferService } from "./node-workspace-transfer-ser import type { WorkerSessionTurnClaim } from "./placement-record.js"; import type { WorkerEnvironmentRecord } from "./store.js"; import type { - WorkerTunnelHandle, WorkerTunnelStatus, + WorkerTurnLaunchRequest, + WorkerTurnTunnelHandle, WorkerWorkspaceCommand, } from "./tunnel-contract.js"; import { boundedWorkerError } from "./worker-error.js"; @@ -72,7 +73,7 @@ type NodeWorkerLaunch = (request: { gatewayNamespace: string; expectedBundleHash: string; placementGeneration: number; - descriptor: Parameters[0]["plan"]; + descriptor: WorkerTurnLaunchRequest["plan"]; }; isDispatchAuthorized: () => boolean; isCancellationAuthorized: () => boolean; @@ -113,10 +114,10 @@ type NodeWorkerTunnelStartRequest = { type NodeTunnelEntry = NodeWorkerTunnelStartRequest & { abortController: AbortController; gatewayNamespace: string; - handle?: WorkerTunnelHandle; + handle?: WorkerTurnTunnelHandle; initialization?: Promise; launchTasks: Set>; - readiness: Deferred; + readiness: Deferred; stopPromise?: Promise; }; @@ -311,7 +312,7 @@ export function createNodeWorkerTunnelManager(options: NodeWorkerTunnelManagerOp const createHandle = ( entry: Omit, restoredWorkspace: NodeWorkerWorkspaceBinding | undefined, - ): { handle: WorkerTunnelHandle; validateRestoredWorkspace: () => Promise } => { + ): { handle: WorkerTurnTunnelHandle; validateRestoredWorkspace: () => Promise } => { let workspaceReady = restoredWorkspace !== undefined; const exec = async (command: Parameters[2]) => { if (!workspaceReady) { @@ -379,7 +380,7 @@ export function createNodeWorkerTunnelManager(options: NodeWorkerTunnelManagerOp } }; const reconcileWorkspace = async ( - request: Parameters[0], + request: Parameters[0], ) => { const pending = request.journal.load(); if (pending) { @@ -514,7 +515,7 @@ export function createNodeWorkerTunnelManager(options: NodeWorkerTunnelManagerOp await fsp.rm(uploaded.stagingRoot, { recursive: true, force: true }); } }; - const handle: WorkerTunnelHandle = { + const handle: WorkerTurnTunnelHandle = { environmentId: entry.environmentId, ownerEpoch: entry.ownerEpoch, launchTurn: async (request) => { @@ -633,7 +634,7 @@ export function createNodeWorkerTunnelManager(options: NodeWorkerTunnelManagerOp bindWorkspaceBindingResolver(resolver: NodeWorkerWorkspaceBindingResolver): void { resolveWorkspaceBinding = resolver; }, - async start(request: NodeWorkerTunnelStartRequest): Promise { + async start(request: NodeWorkerTunnelStartRequest): Promise { const current = entries.get(request.environmentId); if (current) { if (request.ownerEpoch < current.ownerEpoch) { @@ -651,7 +652,7 @@ export function createNodeWorkerTunnelManager(options: NodeWorkerTunnelManagerOp return current.readiness.promise; // Share restored-workspace validation without false readiness. } } - const readiness = createDeferredCore(); + const readiness = createDeferredCore(); void readiness.promise.catch(() => undefined); const entry: NodeTunnelEntry = { ...request, diff --git a/src/gateway/worker-environments/service.test-support.ts b/src/gateway/worker-environments/service.test-support.ts index b26a5ec7d7d8..ac47658002be 100644 --- a/src/gateway/worker-environments/service.test-support.ts +++ b/src/gateway/worker-environments/service.test-support.ts @@ -170,7 +170,6 @@ export function createService( | "ensureNodeWorkerBundle" | "prepareNodeEnrollment" | "retireNodeEnrollment" - | "resolveWorkerGateway" | "tunnelManager" | "generateWorkerCredential" | "liveEvents" @@ -188,7 +187,6 @@ export function createService( prepareInstallation: testState.prepareInstallation, bootstrapWorker: testState.bootstrapWorker, resolveSshIdentity: async () => ({ kind: "path", path: "/keys/worker" }), - resolveWorkerGateway: () => ({ host: "127.0.0.1", port: 18_789 }), generateWorkerCredential: () => CREDENTIAL, executeInference: async () => ({ type: "error", diff --git a/src/gateway/worker-environments/service.ts b/src/gateway/worker-environments/service.ts index 58e34346be03..aead56b9cb5c 100644 --- a/src/gateway/worker-environments/service.ts +++ b/src/gateway/worker-environments/service.ts @@ -101,7 +101,6 @@ type WorkerEnvironmentServiceOptions = { bootstrapCallTimeoutMs?: number; workerCredentialTtlMs?: number; generateWorkerCredential?: (bytes: number) => string; - resolveWorkerGateway?: () => { host: "127.0.0.1" | "::1"; port: number } | undefined; now?: () => number; logger?: { warn: (message: string) => void }; applyTranscriptCommit?: (params: { @@ -309,7 +308,6 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService prepareCurrentBundle: async () => await options.prepareInstallation("bundle"), tunnelManager: options.tunnelManager, nodeTunnelManager: options.nodeTunnelManager, - resolveWorkerGateway: options.resolveWorkerGateway, now, identityResolverFor: providerLifecycle.identityResolverFor, inState, diff --git a/src/gateway/worker-environments/ssh.ts b/src/gateway/worker-environments/ssh.ts index fb48c0a623bc..f39d8ccbc58e 100644 --- a/src/gateway/worker-environments/ssh.ts +++ b/src/gateway/worker-environments/ssh.ts @@ -258,23 +258,6 @@ export async function runWorkerSshCandidates( return lastResult!; } -/** Moves a reconnect to the next candidate without overwriting a newer concurrent selection. */ -export function advanceWorkerSshAfterTransportExit( - prepared: PreparedWorkerSsh, - failedPort: number, - exit: { code: number | null; signal: NodeJS.Signals | null }, -): boolean { - if (exit.code !== 255 || exit.signal !== null || prepared.port !== failedPort) { - return false; - } - const nextPort = workerSshCandidatePorts(prepared)[1]; - if (nextPort === undefined) { - return false; - } - prepared.selectPort(nextPort); - return true; -} - /** Pinned SSH options shared by bootstrap, tunnel control, and workspace transfer. */ export function workerSshOptions( prepared: PreparedWorkerSsh, diff --git a/src/gateway/worker-environments/tunnel-contract.ts b/src/gateway/worker-environments/tunnel-contract.ts index 5f69e4f3b95b..912f88c7657c 100644 --- a/src/gateway/worker-environments/tunnel-contract.ts +++ b/src/gateway/worker-environments/tunnel-contract.ts @@ -99,7 +99,7 @@ export type WorkerWorkspaceQuiescence = { resume(): Promise; }; -type WorkerTurnLaunchRequest = { +export type WorkerTurnLaunchRequest = { plan: WorkerLaunchPlan; turnClaim: WorkerSessionTurnClaim; timeoutMs?: number; @@ -107,10 +107,10 @@ type WorkerTurnLaunchRequest = { onDispatchReady?: () => void; }; -export type WorkerTunnelHandle = { +export type WorkerWorkspaceTunnelHandle = { environmentId: string; ownerEpoch: number; - launchTurn(request: WorkerTurnLaunchRequest): Promise; + launchTurn?: never; runWorkspaceCommand(command: WorkerWorkspaceCommand): Promise; quiesceWorkspace(remoteWorkspaceDir: string): Promise; syncWorkspace(request: WorkerWorkspaceSyncRequest): Promise; @@ -119,3 +119,9 @@ export type WorkerTunnelHandle = { ): Promise; stop(): Promise; }; + +export type WorkerTurnTunnelHandle = Omit & { + launchTurn(request: WorkerTurnLaunchRequest): Promise; +}; + +export type WorkerTunnelHandle = WorkerWorkspaceTunnelHandle | WorkerTurnTunnelHandle; diff --git a/src/gateway/worker-environments/tunnel.test-support.ts b/src/gateway/worker-environments/tunnel.test-support.ts index dc38ce8c1d4b..4882403531b7 100644 --- a/src/gateway/worker-environments/tunnel.test-support.ts +++ b/src/gateway/worker-environments/tunnel.test-support.ts @@ -319,14 +319,6 @@ export function localWorkspaceRunner( return await runCommandWithTimeout(localArgv, options); } if (argv[0] === "ssh") { - if ( - typeof options.input === "string" && - options.input.includes("unsafe worker tunnel directory") - ) { - const result = success(); - onCommandCompleted?.(argv, result); - return result; - } const remoteCommand = argv.at(-1); if (!remoteCommand) { throw new Error("missing test SSH remote command"); @@ -360,8 +352,7 @@ export async function waitForStarts(starts: unknown[], count: number) { await waitForFast(() => expect(starts).toHaveLength(count)); } -type TunnelTestFake = Pick, "runner" | "starts">; -type TunnelManagerOptions = NonNullable[0]>; +type TunnelTestFake = Pick, "runner">; type TunnelManager = ReturnType; export function startTestTunnel( @@ -377,7 +368,6 @@ export function startTestTunnel( bundleHash: BUNDLE_HASH, ssh, sharedHost, - gateway: { host: "127.0.0.1", port: 18789 }, resolveIdentity, }); } @@ -389,21 +379,15 @@ export async function startConnectedTunnel( options: { ssh?: WorkerSshEndpoint; sharedHost?: boolean; - manager?: Omit; - beforeReady?: (start: TunnelTestFake["starts"][number]) => void; } = {}, ) { - const manager = createWorkerTunnelManager({ ...options.manager, runner: fake.runner }); - const starting = startTestTunnel( + const manager = createWorkerTunnelManager({ runner: fake.runner }); + const handle = await startTestTunnel( manager, environmentId, ownerEpoch, options.ssh, options.sharedHost, ); - await waitForStarts(fake.starts, 1); - const start = fake.starts[0]!; - options.beforeReady?.(start); - start.process.becomeReady(); - return { manager, handle: await starting, start }; + return { manager, handle }; } diff --git a/src/gateway/worker-environments/tunnel.test.ts b/src/gateway/worker-environments/tunnel.test.ts index 0c6022d1ee82..9381555e6d9a 100644 --- a/src/gateway/worker-environments/tunnel.test.ts +++ b/src/gateway/worker-environments/tunnel.test.ts @@ -1,25 +1,16 @@ import { describe, expect, it, vi } from "vitest"; -import { - WORKER_PROTOCOL_FEATURES, - WORKER_RPC_SET_VERSION, -} from "../../../packages/gateway-protocol/src/schema/worker-admission.js"; -import { parseWorkerLaunchPlan } from "../../worker/launch-descriptor.js"; import { createWorkerSshRunner } from "./tunnel-ssh-runner.js"; import { createWorkerTunnelManager } from "./tunnel.js"; import { - BUNDLE_HASH, PWD_COMMAND, SSH, deferred, fakeRunner, resolveIdentity, - startConnectedTunnel, startTestTunnel, success, - waitForFast, waitForStarts, } from "./tunnel.test-support.js"; -import { sshArgvPort } from "./worker-ssh-argv.test-support.js"; describe("worker tunnel manager", () => { it("cascades only an epoch-matched environment stop into the desktop tunnel owner", async () => { @@ -53,89 +44,24 @@ describe("worker tunnel manager", () => { expect(close).toHaveBeenCalledWith(1012, "desktop tunnel closed"); }); - it("establishes a pinned reverse socket with keepalives and a separate workspace connection", async () => { + it("prepares pinned workspace SSH without starting a persistent tunnel", async () => { const fake = fakeRunner(); - const { manager, handle, start: tunnel } = await startConnectedTunnel(fake, "worker:one", 3); - 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?.options.input).toContain("sleep 15; printf '.'"); - expect(tunnel?.options.input).toContain("remote command received SIGHUP"); - expect(tunnel?.argv[tunnel.argv.indexOf("-R") + 1]).toMatch( - /^\/tmp\/ocw-[a-f0-9]{16}-3\/gateway\.sock:127\.0\.0\.1:18789$/u, - ); + const manager = createWorkerTunnelManager({ runner: fake.runner }); + const handle = await startTestTunnel(manager, "worker:one", 3); + expect(manager.status("worker:one")).toBe("connected"); + expect(fake.starts).toHaveLength(0); + expect(handle.launchTurn).toBeUndefined(); await expect(handle.runWorkspaceCommand(PWD_COMMAND)).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).not.toContain("-R"); expect(workspace?.argv.at(-1)).toContain("pwd"); - expect(fake.starts).toHaveLength(1); - const plan = parseWorkerLaunchPlan({ - version: 4, - admission: { - environmentId: "worker:one", - credential: "worker-credential-fixture", - sessionId: "session-1", - ownerEpoch: 3, - rpcSetVersion: WORKER_RPC_SET_VERSION, - handshake: { - bundleHash: BUNDLE_HASH, - openclawVersion: "2026.8.13", - protocolFeatures: [...WORKER_PROTOCOL_FEATURES], - }, - }, - assignment: { - agentId: "main", - operationalRunInstance: { instanceId: "instance-1", runId: "run-1" }, - agentRuntimeIdentityToken: "runtime-token", - runId: "run-1", - turnId: "turn-1", - prompt: "inspect", - suppressPromptTranscript: true, - workspaceDir: "/worker/workspace", - modelRef: { provider: "openai", model: "gpt-5.6-luna" }, - inferenceOptions: {}, - initialMessages: [], - transcript: { baseLeafId: null, nextSeq: 1 }, - liveEvents: { ackedSeq: 0, nextSeq: 1 }, - toolAuthority: { allowedToolNames: [] }, - }, - }); - const onDispatchReady = vi.fn(); - await expect( - handle.launchTurn({ - plan, - turnClaim: { - sessionId: plan.admission.sessionId, - claimId: "claim-1", - runId: plan.assignment.runId, - placementGeneration: 1, - owner: { - kind: "worker", - environmentId: plan.admission.environmentId, - ownerEpoch: plan.admission.ownerEpoch, - }, - }, - timeoutMs: 123, - onDispatchReady, - }), - ).resolves.toEqual(success()); - expect(onDispatchReady).toHaveBeenCalledOnce(); - const launch = fake.runs.at(-1); - const remoteLaunchCommand = launch?.argv.at(-1) ?? ""; - expect(remoteLaunchCommand).toContain("'sh' '-c'"); - expect(remoteLaunchCommand).toContain('exec node "$HOME/.openclaw-worker/$1/worker.mjs"'); - expect(remoteLaunchCommand).toContain(`'${BUNDLE_HASH}'`); - expect(launch?.options.input).toContain('"connectionEndpoint":{"kind":"unix"'); - expect(launch?.options.timeoutMs).toBeGreaterThan(0); - expect(launch?.options.timeoutMs).toBeLessThanOrEqual(123); + await handle.stop(); - expect(tunnel?.process.stopCount).toBe(1); expect(manager.status("worker:one")).toBe("stopped"); }); @@ -151,7 +77,8 @@ describe("worker tunnel manager", () => { } return undefined; }); - const { handle } = await startConnectedTunnel(fake, "worker:quiescence-renewal", 3); + const manager = createWorkerTunnelManager({ runner: fake.runner }); + const handle = await startTestTunnel(manager, "worker:quiescence-renewal", 3); vi.useFakeTimers(); try { @@ -179,9 +106,8 @@ describe("worker tunnel manager", () => { } return undefined; }); - const { handle } = await startConnectedTunnel(fake, "worker:shared-quiescence", 3, { - sharedHost: true, - }); + const manager = createWorkerTunnelManager({ runner: fake.runner }); + const handle = await startTestTunnel(manager, "worker:shared-quiescence", 3, SSH, true); const quiescence = await handle.quiesceWorkspace("/home/worker/workspace"); await quiescence.assertActive(); @@ -196,404 +122,41 @@ describe("worker tunnel manager", () => { await handle.stop(); }); - it("reconnects with capped backoff after unexpected exits and failed attempts", async () => { + it("fences stale owners when a replacement epoch takes ownership", async () => { const fake = fakeRunner(); - const delays: number[] = []; - const { manager, handle } = await startConnectedTunnel(fake, "worker:retry", 1, { - manager: { - backoff: { initialMs: 5, maxMs: 10, factor: 2, jitter: 0 }, - sleep: async (ms) => { - delays.push(ms); - }, - }, - }); - - 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("reports stopped when the reconnect loop settles without removing its entry", async () => { - const fake = fakeRunner(); - const { manager, handle } = await startConnectedTunnel(fake, "worker:settled-loop", 1, { - manager: { - sleep: async () => { - throw new Error("retry scheduler stopped"); - }, - }, - }); - - fake.starts[0]?.process.exit(); - await waitForFast(() => expect(manager.status("worker:settled-loop")).toBe("stopped")); - - await handle.stop(); - }); - - it("times out a marker-less SSH child and retries", async () => { - vi.useFakeTimers(); - const fake = fakeRunner(); - const manager = createWorkerTunnelManager({ runner: fake.runner, sleep: async () => {} }); - const starting = startTestTunnel(manager, "worker:ready-timeout", 1); - const rejected = expect(starting).rejects.toThrow("stopped before connecting"); - - try { - await waitForStarts(fake.starts, 1); - await vi.advanceTimersByTimeAsync(60_000); - await waitForStarts(fake.starts, 2); - - expect(fake.starts[0]?.process.stopCount).toBe(1); - expect(manager.status("worker:ready-timeout")).toBe("reconnecting"); - } finally { - await manager.stop("worker:ready-timeout"); - await rejected; - vi.useRealTimers(); - } - }); - - it("reconnects on the next advertised port after SSH transport exit 255", async () => { - const fake = fakeRunner(); - const manager = createWorkerTunnelManager({ - runner: fake.runner, - sleep: async () => {}, - }); - const request = { - bundleHash: BUNDLE_HASH, - environmentId: "worker:port-reconnect", - ownerEpoch: 1, - ssh: { ...SSH, port: 2222, fallbackPorts: [22] }, - gateway: { host: "127.0.0.1", port: 18789 }, - resolveIdentity, - } as const; - const starting = manager.start(request); - await waitForStarts(fake.starts, 1); - expect(sshArgvPort(fake.starts[0]!.argv)).toBe(2222); - fake.starts[0]!.process.becomeReady(); - await starting; - - fake.starts[0]!.process.exit(255); - await waitForStarts(fake.starts, 2); - expect(sshArgvPort(fake.starts[1]!.argv)).toBe(22); - expect(sshArgvPort(fake.runs.at(-1)!.argv)).toBe(22); - const reconnecting = manager.start(request); - const reconnectSettled = vi.fn(); - void reconnecting.then(reconnectSettled, reconnectSettled); - await Promise.resolve(); - await Promise.resolve(); - expect(reconnectSettled).not.toHaveBeenCalled(); - - fake.starts[1]!.process.becomeReady(); - const handle = await reconnecting; - await expect(handle.runWorkspaceCommand(PWD_COMMAND)).resolves.toEqual(success()); - expect(sshArgvPort(fake.runs.at(-1)!.argv)).toBe(22); - await handle.stop(); - }); - - it("waits before stateful dispatch and aborts reconnect waits on owner stop", async () => { - const fake = fakeRunner(); - const sleepStarted = deferred(); - const { handle } = await startConnectedTunnel(fake, "worker:reconnect-command-policy", 1, { - manager: { - sleep: async (_ms, signal) => { - if (!signal) { - throw new Error("missing reconnect signal"); - } - sleepStarted.resolve(signal); - await new Promise((_resolve, reject) => { - signal.addEventListener( - "abort", - () => - reject( - signal.reason instanceof Error - ? signal.reason - : new Error("reconnect sleep aborted"), - ), - { once: true }, - ); - }); - }, - }, - }); - fake.starts[0]!.process.exit(255); - await sleepStarted.promise; - - const idempotent = handle.runWorkspaceCommand(PWD_COMMAND); - const idempotentResult = expect(idempotent).rejects.toThrow( - "Worker tunnel owner is no longer connected", - ); - const stateful = handle.runWorkspaceCommand({ ...PWD_COMMAND, transportRetry: "never" }); - const statefulSettled = vi.fn(); - void stateful.then(statefulSettled, statefulSettled); - await Promise.resolve(); - await Promise.resolve(); - expect(statefulSettled).not.toHaveBeenCalled(); - - await handle.stop(); - await idempotentResult; - await expect(stateful).rejects.toThrow("Worker tunnel owner is no longer connected"); - }); - - it("does not dispatch a stateful command cancelled during reconnect", async () => { - const fake = fakeRunner(); - const releaseReconnect = deferred(); - const { handle, manager } = await startConnectedTunnel(fake, "worker:cancel-reconnect", 1, { - manager: { - sleep: async () => await releaseReconnect.promise, - }, - }); - const controller = new AbortController(); - const onDispatchReady = vi.fn(); - - try { - fake.starts[0]!.process.exit(255); - await waitForFast(() => - expect(manager.status("worker:cancel-reconnect")).toBe("reconnecting"), - ); - const command = handle.runWorkspaceCommand({ - ...PWD_COMMAND, - transportRetry: "never", - signal: controller.signal, - onDispatchReady, - }); - const settled = vi.fn(); - void command.then(settled, settled); - - controller.abort(new Error("turn cancelled")); - await waitForFast(() => expect(settled).toHaveBeenCalledOnce(), { timeout: 100 }); - await expect(command).rejects.toThrow("turn cancelled"); - - releaseReconnect.resolve(); - await waitForStarts(fake.starts, 2); - fake.starts[1]!.process.becomeReady(); - await Promise.resolve(); - await Promise.resolve(); - expect(onDispatchReady).not.toHaveBeenCalled(); - expect(fake.runs.filter((run) => run.argv.at(-1)?.includes("'pwd'"))).toHaveLength(0); - } finally { - releaseReconnect.resolve(); - await handle.stop(); - } - }); - - it("does not replay a stateful command after an ambiguous transport exit", async () => { - const fake = fakeRunner((argv) => - argv.at(-1)?.includes("'pwd'") ? { ...success(), code: 255 } : undefined, - ); - const { handle } = await startConnectedTunnel(fake, "worker:stateful-no-replay", 1, { - ssh: { ...SSH, port: 2222, fallbackPorts: [22] }, - }); - - try { - await expect( - handle.runWorkspaceCommand({ ...PWD_COMMAND, transportRetry: "never" }), - ).resolves.toMatchObject({ code: 255 }); - expect(fake.runs.filter((run) => run.argv.at(-1)?.includes("'pwd'"))).toHaveLength(1); - } finally { - await handle.stop(); - } - }); - - it("shares setup and best-effort stop cleanup deadlines across fallback candidates", async () => { - let nowMs = 1_000; - const dateNow = vi.spyOn(Date, "now").mockImplementation(() => nowMs); - const setupAttempts: Array<{ port: number; timeoutMs: number }> = []; - const cleanupAttempts: Array<{ port: number; timeoutMs: number }> = []; - const fake = fakeRunner((argv, options) => { - const port = sshArgvPort(argv); - if (port === undefined) { - throw new Error("missing tunnel SSH port"); - } - if ( - typeof options.input === "string" && - options.input.includes("unsafe worker tunnel directory") - ) { - const timeoutMs = options.timeoutMs; - if (timeoutMs === undefined) { - throw new Error("missing tunnel setup timeout"); - } - setupAttempts.push({ port, timeoutMs }); - if (setupAttempts.length === 1) { - nowMs += 7_000; - return { ...success("", "primary transport unavailable"), code: 255 }; - } - return success(); - } - if (typeof options.input === "string" && options.input.includes('rmdir -- "$directory"')) { - const timeoutMs = options.timeoutMs; - if (timeoutMs === undefined) { - throw new Error("missing tunnel cleanup timeout"); - } - cleanupAttempts.push({ port, timeoutMs }); - if (cleanupAttempts.length === 1) { - nowMs += 5_000; - return { ...success("", "selected transport unavailable"), code: 255 }; - } - return success(); - } - return undefined; - }); - const manager = createWorkerTunnelManager({ runner: fake.runner, sleep: async () => {} }); - try { - const starting = startTestTunnel(manager, "worker:operation-deadline", 1, { - ...SSH, - port: 2222, - fallbackPorts: [22], - }); - await waitForStarts(fake.starts, 1); - expect(sshArgvPort(fake.starts[0]!.argv)).toBe(22); - fake.starts[0]!.process.becomeReady(); - const handle = await starting; - - fake.starts[0]!.process.exit(); - await waitForStarts(fake.starts, 2); - expect(sshArgvPort(fake.starts[1]!.argv)).toBe(22); - fake.starts[1]!.process.becomeReady(); - await handle.stop(); - expect(setupAttempts).toEqual([ - { port: 2222, timeoutMs: 20_000 }, - { port: 22, timeoutMs: 13_000 }, - { port: 22, timeoutMs: 20_000 }, - ]); - expect(cleanupAttempts).toEqual([ - { port: 22, timeoutMs: 20_000 }, - { port: 2222, timeoutMs: 15_000 }, - ]); - expect(manager.status("worker:operation-deadline")).toBe("stopped"); - } finally { - dateNow.mockRestore(); - await manager.stopAll(); - } - }); - - it("backs off repeated short-lived connected tunnels", async () => { - const fake = fakeRunner(); - const delays: number[] = []; - const { handle } = await startConnectedTunnel(fake, "worker:flap", 1, { - manager: { - backoff: { initialMs: 5, maxMs: 10, factor: 2, jitter: 0 }, - sleep: async (ms) => { - delays.push(ms); - }, - }, - }); - - 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, handle } = await startConnectedTunnel(fake, "worker:drain", 8, { - manager: { - 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 }); - }); - }, - }, - }); - fake.starts[0]?.process.exit(); - await sleepStarted.promise; - - const reconnecting = manager.start({ - bundleHash: BUNDLE_HASH, - environmentId: "worker:drain", - ownerEpoch: 8, - ssh: SSH, - gateway: { host: "127.0.0.1", port: 18789 }, - resolveIdentity, - }); - const reconnectResult = expect(reconnecting).rejects.toThrow("stopped before connecting"); - await handle.stop(); - await reconnectResult; - expect(manager.status("worker:drain")).toBe("stopped"); - expect(fake.starts).toHaveLength(1); - - const pending = startTestTunnel(manager, "worker:late", 1); - 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, handle } = await startConnectedTunnel(fake, "worker:epoch", 4); + const manager = createWorkerTunnelManager({ runner: fake.runner }); + const stale = await startTestTunnel(manager, "worker:epoch", 4); await expect(startTestTunnel(manager, "worker:epoch", 3)).rejects.toThrow("epoch is stale"); - expect(fake.starts).toHaveLength(1); - await handle.stop(); + + const replacement = await startTestTunnel(manager, "worker:epoch", 5); + await expect(stale.runWorkspaceCommand(PWD_COMMAND)).rejects.toThrow( + "Worker tunnel owner is no longer connected", + ); + await expect(replacement.runWorkspaceCommand(PWD_COMMAND)).resolves.toEqual(success()); + expect(replacement.ownerEpoch).toBe(5); + expect(manager.status("worker:epoch")).toBe("connected"); + await replacement.stop(); }); - it("publishes a replacement epoch before awaiting prior teardown", async () => { + it("fails initialization that loses ownership before identity preparation completes", async () => { + const identity = deferred>>(); const fake = fakeRunner(); - const manager = createWorkerTunnelManager({ runner: fake.runner, sleep: async () => {} }); - const initialRequest = { - bundleHash: BUNDLE_HASH, - environmentId: "worker:replacement", + const manager = createWorkerTunnelManager({ runner: fake.runner }); + const starting = manager.start({ + environmentId: "worker:pending", ownerEpoch: 1, + bundleHash: "a".repeat(64), ssh: SSH, - gateway: { host: "127.0.0.1", port: 18789 }, - resolveIdentity, - } as const; - const current = manager.start(initialRequest); - await waitForStarts(fake.starts, 1); - fake.starts[0]?.process.becomeReady(); - await current; - - fake.starts[0]?.process.exit(); - await waitForStarts(fake.starts, 2); - const staleReconnect = fake.starts[1]!.process; - const staleOwnerStart = manager.start(initialRequest); - const staleOwnerResult = expect(staleOwnerStart).rejects.toThrow("stopped before connecting"); - const releaseStop = deferred(); - staleReconnect.blockStopUntil(releaseStop.promise); - const replacement = manager.start({ - bundleHash: BUNDLE_HASH, - environmentId: "worker:replacement", - ownerEpoch: 2, - ssh: SSH, - gateway: { host: "127.0.0.1", port: 18789 }, - resolveIdentity, + resolveIdentity: async () => await identity.promise, }); - const replacementSettled = vi.fn(); - void replacement.then(replacementSettled, replacementSettled); - await waitForFast(() => expect(staleReconnect.stopCount).toBeGreaterThan(0)); - await staleOwnerResult; - staleReconnect.becomeReady(); - await Promise.resolve(); - await Promise.resolve(); - expect(replacementSettled).not.toHaveBeenCalled(); - releaseStop.resolve(); - await waitForStarts(fake.starts, 3); - expect(replacementSettled).not.toHaveBeenCalled(); - fake.starts[2]!.process.becomeReady(); - const handle = await replacement; + const stopping = manager.stop("worker:pending", 1); + identity.resolve(await resolveIdentity()); - expect(handle.ownerEpoch).toBe(2); - expect(manager.status("worker:replacement")).toBe("connected"); - await handle.stop(); + await stopping; + await expect(starting).rejects.toThrow("Worker tunnel owner is no longer connected"); + expect(manager.status("worker:pending")).toBe("stopped"); }); }); diff --git a/src/gateway/worker-environments/tunnel.ts b/src/gateway/worker-environments/tunnel.ts index 17fbe89b2eb3..95e83d9c3273 100644 --- a/src/gateway/worker-environments/tunnel.ts +++ b/src/gateway/worker-environments/tunnel.ts @@ -1,94 +1,21 @@ -import { RetrySupervisor } from "../../../packages/retry/src/index.js"; -import { sleepWithAbort, type BackoffPolicy } from "../../infra/backoff.js"; -import { withTimeout } from "../../infra/fs-safe.js"; -import { createSubsystemLogger } from "../../logging/subsystem.js"; import type { WorkerSshEndpoint } from "../../plugins/types.js"; -import type { SpawnResult } from "../../process/exec.js"; -import { createDeferredCore, type Deferred } from "../../shared/deferred.js"; -import { completeWorkerLaunchDescriptor } from "../../worker/launch-descriptor.js"; import type { DesktopSessionRegistry } from "../desktop/session-registry.js"; import { createWorkerDesktopTunnels } from "./desktop-tunnel.js"; -import { - advanceWorkerSshAfterTransportExit, - prepareWorkerSsh, - type PreparedWorkerSsh, - runWorkerSshCandidates, - type WorkerSshIdentityResolver, - workerSshCommandOptions, - workerSshOptions, - workerSshRemoteCommand, -} from "./ssh.js"; +import { prepareWorkerSsh, type PreparedWorkerSsh, type WorkerSshIdentityResolver } from "./ssh.js"; import { WorkerTunnelOwnerDisconnectedError, type WorkerTunnelHandle, type WorkerTunnelRequest, + type WorkerWorkspaceTunnelHandle, type WorkerTunnelStatus, } from "./tunnel-contract.js"; -import { - createWorkerSshRunner, - type WorkerSshProcess, - type WorkerSshRunner, - workerSshProcessError, - WORKER_TUNNEL_READY_MARKER, -} from "./tunnel-ssh-runner.js"; -import { boundedWorkerError } from "./worker-error.js"; -import { stableWorkerPathComponent } from "./workspace-sync-helpers.js"; +import { createWorkerSshRunner, type WorkerSshRunner } from "./tunnel-ssh-runner.js"; import { createWorkerWorkspaceActions } from "./workspace-sync.js"; export type { WorkerTunnelHandle } from "./tunnel-contract.js"; -const REMOTE_SOCKET_NAME = "gateway.sock"; -const REMOTE_SETUP_TIMEOUT_MS = 20_000; -// A live SSH process without the remote marker is not a usable tunnel. Bound each attempt so the -// retry supervisor can move on instead of pinning the environment forever. -const TUNNEL_READY_TIMEOUT_MS = 60_000; -const DEFAULT_STABLE_CONNECTION_MS = 30_000; -const DEFAULT_BACKOFF: BackoffPolicy = { - initialMs: 250, - maxMs: 30_000, - factor: 2, - jitter: 0, -}; -const tunnelLog = createSubsystemLogger("gateway/worker-tunnel"); - -const REMOTE_SOCKET_SETUP_SCRIPT = String.raw`set -eu -directory=$1 -socket=$2 -umask 077 -if [ -e "$directory" ] || [ -L "$directory" ]; then - if [ ! -d "$directory" ] || [ -L "$directory" ]; then - printf '%s\n' 'unsafe worker tunnel directory' >&2 - exit 2 - fi -else - mkdir -- "$directory" -fi -chmod 700 "$directory" # no "--": BSD/macOS chmod treats it as a filename; path is script-owned and absolute -rm -f -- "$socket" -`; - -const REMOTE_TUNNEL_READY_SCRIPT = String.raw`set -eu -socket=$1 -test -S "$socket" -printf '%s\n' '${WORKER_TUNNEL_READY_MARKER}' -trap 'printf "%s\n" "worker tunnel remote command received SIGHUP" >&2; exit 129' HUP -trap 'printf "%s\n" "worker tunnel remote command received SIGINT" >&2; exit 130' INT -trap 'printf "%s\n" "worker tunnel remote command received SIGTERM" >&2; exit 143' TERM -# ServerAlive messages protect the SSH transport, not an idle session channel. Keep the control -# channel active too so provider sshd ChannelTimeout policies cannot retire a healthy tunnel. -while :; do sleep 15; printf '.'; done -`; - -const REMOTE_SOCKET_CLEANUP_SCRIPT = String.raw`set -eu -socket=$1 -directory=$2 -rm -f -- "$socket" -rmdir -- "$directory" 2>/dev/null || true -`; -const WORKER_LAUNCH_SCRIPT = 'exec node "$HOME/.openclaw-worker/$1/worker.mjs"'; type WorkerTunnelStartRequest = WorkerTunnelRequest & { bundleHash: string; - gateway: { host: "127.0.0.1" | "::1"; port: number }; ssh: WorkerSshEndpoint; sharedHost?: boolean; resolveIdentity: WorkerSshIdentityResolver; @@ -98,35 +25,20 @@ type TunnelEntry = { bundleHash: string; environmentId: string; ownerEpoch: number; - gateway: WorkerTunnelStartRequest["gateway"]; sharedHost: boolean; - remoteDirectory: string; - remoteSocketPath: string; abortController: AbortController; status: Exclude; prepared?: PreparedWorkerSsh; - process?: WorkerSshProcess; - initialization?: Promise; - loop?: Promise; - loopSettled: boolean; + initialization?: Promise; stopPromise?: Promise; - readiness: Deferred; workspaceTasks: Set>; }; type WorkerTunnelManagerOptions = { runner?: WorkerSshRunner; desktopSessionRegistry?: DesktopSessionRegistry; - sleep?: (ms: number, signal?: AbortSignal) => Promise; - backoff?: BackoffPolicy; - now?: () => number; - stableConnectionMs?: number; }; -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"); @@ -134,26 +46,11 @@ function validateStartRequest(request: WorkerTunnelStartRequest): void { 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. */ +/** Owns SSH workspace state for remote-exec environments and fences replacement epochs. */ 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 desktop = createWorkerDesktopTunnels({ runner, ...(options.desktopSessionRegistry ? { registry: options.desktopSessionRegistry } : {}), @@ -164,99 +61,10 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = const isCurrent = (entry: TunnelEntry) => entries.get(entry.environmentId) === entry && !entry.abortController.signal.aborted; - const sshCommand = ( - prepared: PreparedWorkerSsh, - params: { - input: string; - port: number; - remoteArgs: readonly string[]; - timeoutMs: number; - signal?: AbortSignal; - }, - ) => ({ - argv: [ - "ssh", - ...workerSshOptions(prepared, { forwarding: "disabled" as const }), - "-a", - "-x", - "-T", - "-p", - String(params.port), - "--", - prepared.sshTarget, - workerSshRemoteCommand(["sh", "-s", "--", ...params.remoteArgs]), - ], - options: workerSshCommandOptions({ - input: params.input, - timeoutMs: params.timeoutMs, - signal: params.signal, - }), - }); - - const prepareRemoteSocket = async (entry: TunnelEntry) => { - const prepared = entry.prepared; - if (!prepared) { - throw new Error("Worker tunnel SSH context is unavailable"); - } - const result = await runWorkerSshCandidates( - prepared, - REMOTE_SETUP_TIMEOUT_MS, - async (port, remainingTimeoutMs) => { - const command = sshCommand(prepared, { - input: REMOTE_SOCKET_SETUP_SCRIPT, - port, - remoteArgs: [entry.remoteDirectory, entry.remoteSocketPath], - timeoutMs: remainingTimeoutMs, - signal: entry.abortController.signal, - }); - return await runner.run(command.argv, command.options); - }, - ); - if (!success(result)) { - throw workerSshProcessError(result.stderr || result.stdout); - } - }; - - const cleanupRemoteSocket = async (entry: TunnelEntry) => { - const prepared = entry.prepared; - if (!prepared) { - return; - } - await runWorkerSshCandidates( - prepared, - REMOTE_SETUP_TIMEOUT_MS, - async (port, remainingTimeoutMs) => { - const command = sshCommand(prepared, { - input: REMOTE_SOCKET_CLEANUP_SCRIPT, - port, - remoteArgs: [entry.remoteSocketPath, entry.remoteDirectory], - timeoutMs: remainingTimeoutMs, - }); - return await runner.run(command.argv, command.options); - }, - ).catch(() => undefined); - }; - - const createHandle = (entry: TunnelEntry): WorkerTunnelHandle => { - const getPrepared = () => - isCurrent(entry) && entry.status === "connected" ? entry.prepared : undefined; - // Handles outlive individual SSH children. Wait only on this owner's current barrier; - // replacement or stop makes the entry non-current and must remain fail-closed. + const createHandle = (entry: TunnelEntry): WorkerWorkspaceTunnelHandle => { const waitForPrepared = async (): Promise => { - while (isCurrent(entry)) { - const prepared = getPrepared(); - if (prepared) { - return prepared; - } - const readiness = entry.readiness; - try { - await readiness.promise; - } catch (error) { - if (!isCurrent(entry)) { - break; - } - throw error; - } + if (isCurrent(entry) && entry.status === "connected" && entry.prepared) { + return entry.prepared; } throw new WorkerTunnelOwnerDisconnectedError(); }; @@ -272,193 +80,11 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = return { environmentId: entry.environmentId, ownerEpoch: entry.ownerEpoch, - launchTurn: (request) => - workspace.runWorkspaceCommand({ - transportRetry: "never", - argv: ["sh", "-c", WORKER_LAUNCH_SCRIPT, "openclaw-worker", entry.bundleHash], - input: JSON.stringify( - completeWorkerLaunchDescriptor(request.plan, { - kind: "unix", - socketPath: entry.remoteSocketPath, - }), - ), - timeoutMs: request.timeoutMs, - signal: request.signal, - onDispatchReady: request.onDispatchReady, - }), ...workspace, stop: () => stop(entry.environmentId, entry.ownerEpoch), }; }; - const connect = async ( - entry: TunnelEntry, - ): Promise<{ port: number; process: WorkerSshProcess }> => { - 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}`; - const port = prepared.port; - const process = 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(port), - "--", - prepared.sshTarget, - workerSshRemoteCommand(["sh", "-s", "--", entry.remoteSocketPath]), - ], - workerSshCommandOptions({ - input: REMOTE_TUNNEL_READY_SCRIPT, - timeoutMs: Number.MAX_SAFE_INTEGER, - signal: entry.abortController.signal, - }), - ); - return { port, process }; - }; - - const reconnectLoop = async (entry: TunnelEntry) => { - const reconnectSupervisor = new RetrySupervisor(backoff); - while (isCurrent(entry)) { - entry.status = reconnectSupervisor.attempts === 0 ? "connecting" : "reconnecting"; - const attempt = reconnectSupervisor.attempts + 1; - const reconnecting = entry.status === "reconnecting"; - const connectStartedAtMs = now(); - if (reconnecting) { - tunnelLog.warn("worker tunnel reconnect attempt started", { - environmentId: entry.environmentId, - ownerEpoch: entry.ownerEpoch, - attempt, - status: entry.status, - port: entry.prepared?.port, - workspaceTaskCount: entry.workspaceTasks.size, - }); - } - let child: WorkerSshProcess | undefined; - let childPort: number | undefined; - try { - const connection = await connect(entry); - child = connection.process; - childPort = connection.port; - entry.process = child; - await withTimeout(child.ready, TUNNEL_READY_TIMEOUT_MS, { - message: "Worker tunnel did not become ready within 60 seconds", - }); - if (!isCurrent(entry)) { - await child.stop(); - return; - } - entry.status = "connected"; - if (reconnecting) { - tunnelLog.info("worker tunnel reconnected", { - environmentId: entry.environmentId, - ownerEpoch: entry.ownerEpoch, - attempt, - port: childPort, - durationMs: now() - connectStartedAtMs, - }); - } - const connectionReadiness = entry.readiness; - connectionReadiness.resolve(createHandle(entry)); - const connectedAtMs = now(); - const exit = await child.exited.finally(() => { - if (isCurrent(entry) && entry.readiness === connectionReadiness) { - // Each established child owns one readiness barrier. Replace it as soon as that child - // is lost so same-owner callers wait for the reconnect instead of using a stale handle. - entry.status = "reconnecting"; - const readiness = createDeferredCore(); - void readiness.promise.catch(() => undefined); - entry.readiness = readiness; - } - }); - if (isCurrent(entry) && entry.workspaceTasks.size > 0) { - tunnelLog.warn("worker tunnel SSH child exited during workspace operation", { - environmentId: entry.environmentId, - ownerEpoch: entry.ownerEpoch, - exitCode: exit.code, - signal: exit.signal, - ...(exit.stderrTail ? { stderrTail: exit.stderrTail } : {}), - workspaceTaskCount: entry.workspaceTasks.size, - }); - } - if (entry.prepared) { - advanceWorkerSshAfterTransportExit(entry.prepared, childPort, exit); - } - if (now() - connectedAtMs >= stableConnectionMs) { - reconnectSupervisor.reset(); - } - } catch (error) { - if (child && childPort !== undefined) { - let stopError: unknown; - let stopFailed = false; - const stopping = child.stop().catch((failure: unknown) => { - stopFailed = true; - stopError = failure; - }); - let exit = await Promise.race([ - child.exited.catch(() => undefined), - stopping.then(() => undefined), - ]); - await stopping; - if (stopFailed) { - // A failed stop means the SSH child may still be running. Never drop it from - // tracking and never retry over it — wait for its real exit first, and keep - // that late exit so transport-exit port rotation still advances. - tunnelLog.warn("worker tunnel stop failed; waiting for SSH child exit", { - environmentId: entry.environmentId, - error: boundedWorkerError(stopError), - connectError: boundedWorkerError(error), - }); - exit = (await child.exited.catch(() => undefined)) ?? exit; - } - if (exit && entry.prepared) { - advanceWorkerSshAfterTransportExit(entry.prepared, childPort, exit); - } - } - if (isCurrent(entry)) { - tunnelLog.warn("worker tunnel connect attempt failed", { - environmentId: entry.environmentId, - attempt: reconnectSupervisor.attempts + 1, - error: boundedWorkerError(error), - }); - } - } finally { - if (entry.process === child) { - entry.process = undefined; - } - } - if (!isCurrent(entry)) { - return; - } - entry.status = "reconnecting"; - try { - const retry = reconnectSupervisor.next(entry.abortController.signal)!; - await sleep(retry.delayMs, retry.signal); - } catch { - return; - } - } - }; - const stopEntry = (entry: TunnelEntry): Promise => { if (entry.stopPromise) { return entry.stopPromise; @@ -468,13 +94,8 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = entries.delete(entry.environmentId); } entry.abortController.abort(new Error("Worker tunnel owner stopped")); - entry.readiness.reject(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; @@ -493,61 +114,48 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = throw new Error("Worker tunnel owner epoch is stale"); } if (request.ownerEpoch === current.ownerEpoch) { - return await current.readiness.promise; + return await current.initialization!; } } - const environmentKey = stableWorkerPathComponent(request.environmentId, 16); - const remoteDirectory = `/tmp/ocw-${environmentKey}-${request.ownerEpoch}`; - const readiness = createDeferredCore(); - void readiness.promise.catch(() => undefined); const entry: TunnelEntry = { environmentId: request.environmentId, bundleHash: request.bundleHash, ownerEpoch: request.ownerEpoch, - gateway: request.gateway, sharedHost: request.sharedHost === true, - remoteDirectory, - remoteSocketPath: `${remoteDirectory}/${REMOTE_SOCKET_NAME}`, abortController: new AbortController(), status: "connecting", - loopSettled: false, - readiness, 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. + // Publish the new owner before waiting for prior teardown so stop/drain can fence initialization. entries.set(request.environmentId, entry); entry.initialization = (async () => { if (current) { await stopEntry(current); } if (!isCurrent(entry)) { - return; + throw new WorkerTunnelOwnerDisconnectedError(); } - entry.prepared = await prepareWorkerSsh({ + const prepared = await prepareWorkerSsh({ ssh: request.ssh, pinnedHostKey: request.ssh.hostKey, resolveIdentity: request.resolveIdentity, - temporaryDirectoryPrefix: "openclaw-worker-tunnel-", + temporaryDirectoryPrefix: "openclaw-worker-workspace-", }); if (!isCurrent(entry)) { - await entry.prepared.dispose(); - entry.prepared = undefined; - return; + await prepared.dispose(); + throw new WorkerTunnelOwnerDisconnectedError(); } - entry.loop = reconnectLoop(entry).finally(() => { - entry.loopSettled = true; - }); - void entry.loop.catch((error: unknown) => { - entry.readiness.reject(error instanceof Error ? error : new Error("Worker tunnel failed")); - }); + entry.prepared = prepared; + entry.status = "connected"; + return createHandle(entry); })(); - void entry.initialization.catch((error: unknown) => { - entry.readiness.reject(error instanceof Error ? error : new Error("Worker tunnel failed")); - void stopEntry(entry); - }); - return await entry.readiness.promise; + try { + return await entry.initialization; + } catch (error) { + await stopEntry(entry); + throw error; + } } async function stop(environmentId: string, ownerEpoch?: number): Promise { @@ -560,8 +168,8 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = async function stopAll(): Promise { const current = [...entries.values()]; + entries.clear(); for (const entry of current) { - entries.delete(entry.environmentId); entry.abortController.abort(new Error("Worker tunnel manager stopped")); } await Promise.all([...current.map(stopEntry), desktop.stopAll()]); @@ -573,8 +181,7 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = stop, stopAll, status(environmentId: string): WorkerTunnelStatus { - const entry = entries.get(environmentId); - return !entry || entry.loopSettled ? "stopped" : entry.status; + return entries.get(environmentId)?.status ?? "stopped"; }, }; } diff --git a/src/gateway/worker-environments/worker-turn-launcher-claim-admission.test.ts b/src/gateway/worker-environments/worker-turn-launcher-claim-admission.test.ts index 07e9ad6f6039..c889aa14be15 100644 --- a/src/gateway/worker-environments/worker-turn-launcher-claim-admission.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher-claim-admission.test.ts @@ -5,7 +5,7 @@ import type { SpawnResult } from "../../process/exec.js"; import { completeWorkerLaunchDescriptor } from "../../worker/launch-descriptor.js"; import { completeReclaimedWorkspaceTeardown } from "./placement-teardown.js"; import { createWorkerSessionPlacementGate } from "./placement-worker-gate.js"; -import type { WorkerTunnelHandle } from "./tunnel-contract.js"; +import type { WorkerTurnLaunchRequest } from "./tunnel-contract.js"; import { ENVIRONMENT_ID, MANIFEST_REF, @@ -191,7 +191,7 @@ describe("worker turn launcher claim admission", () => { killed: false; termination: "exit"; }>(); - const launchTurn = vi.fn((request: Parameters[0]) => { + const launchTurn = vi.fn((request: WorkerTurnLaunchRequest) => { request.onDispatchReady?.(); commandStarted.resolve(); return commandFinished.promise; diff --git a/src/gateway/worker-environments/worker-turn-launcher-reclaimed-placement.test.ts b/src/gateway/worker-environments/worker-turn-launcher-reclaimed-placement.test.ts index 2c11e0c86029..29b4bbdee52c 100644 --- a/src/gateway/worker-environments/worker-turn-launcher-reclaimed-placement.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher-reclaimed-placement.test.ts @@ -21,7 +21,7 @@ import { import { getCommandLaneSnapshot, setCommandLaneConcurrency } from "../../process/command-queue.js"; import type { SpawnResult } from "../../process/exec.js"; import { createWorkerSessionPlacementGate } from "./placement-worker-gate.js"; -import type { WorkerTunnelHandle } from "./tunnel-contract.js"; +import type { WorkerTurnLaunchRequest } from "./tunnel-contract.js"; import { ENVIRONMENT_ID, MANIFEST_REF, @@ -82,41 +82,39 @@ describe("worker turn launcher reclaimed placement", () => { } return active; }; - const launchTurn = vi.fn( - async (request: Parameters[0]): Promise => { - request.onDispatchReady?.(); - workerStarted.resolve(); - await resumeWorker.promise; - expect(placements.get(SESSION_ID)).toMatchObject({ - state: "active", - turnClaim: { owner: "worker", runId }, - }); - const completed = openSessionManager(); - const leafId = completed.appendMessage( - makeAgentAssistantMessage({ - content: [{ type: "text", text: "Redispatched worker reply" }], - timestamp: 51, - }), - ); - createWorkerSessionPlacementGate(placements).updateAckCursors({ - claim: request.turnClaim, - transcriptSeq: 2, - liveSeq: 1, - }); - return { - stdout: JSON.stringify({ - status: "completed", - transcriptLeafId: leafId, - transcriptNextSeq: (placements.get(SESSION_ID)?.lastTranscriptAckCursor ?? 0) + 1, - }), - stderr: "", - code: 0, - signal: null, - killed: false, - termination: "exit", - }; - }, - ); + const launchTurn = vi.fn(async (request: WorkerTurnLaunchRequest): Promise => { + request.onDispatchReady?.(); + workerStarted.resolve(); + await resumeWorker.promise; + expect(placements.get(SESSION_ID)).toMatchObject({ + state: "active", + turnClaim: { owner: "worker", runId }, + }); + const completed = openSessionManager(); + const leafId = completed.appendMessage( + makeAgentAssistantMessage({ + content: [{ type: "text", text: "Redispatched worker reply" }], + timestamp: 51, + }), + ); + createWorkerSessionPlacementGate(placements).updateAckCursors({ + claim: request.turnClaim, + transcriptSeq: 2, + liveSeq: 1, + }); + return { + stdout: JSON.stringify({ + status: "completed", + transcriptLeafId: leafId, + transcriptNextSeq: (placements.get(SESSION_ID)?.lastTranscriptAckCursor ?? 0) + 1, + }), + stderr: "", + code: 0, + signal: null, + killed: false, + termination: "exit", + }; + }); const environments: WorkerTurnEnvironmentService = { get: vi.fn(() => attachedEnvironment()), acquireTurnCredential: vi.fn(async () => credential()), diff --git a/src/gateway/worker-environments/worker-turn-launcher.ts b/src/gateway/worker-environments/worker-turn-launcher.ts index d39bb6dfe679..5177d17a35fc 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.ts @@ -286,6 +286,9 @@ async function executeWorkerTurn(params: { handoffAbort.abort(handoffError); } }; + if (!tunnel.launchTurn) { + throw new Error("Worker tunnel does not support worker turns"); + } const processPromise = tunnel.launchTurn({ plan, turnClaim: params.turnClaim, diff --git a/src/gateway/worker-environments/workspace-sync-reconnect.test.ts b/src/gateway/worker-environments/workspace-sync-reconnect.test.ts deleted file mode 100644 index 8284453a0773..000000000000 --- a/src/gateway/worker-environments/workspace-sync-reconnect.test.ts +++ /dev/null @@ -1,219 +0,0 @@ -import fs from "node:fs/promises"; -import path from "node:path"; -import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; -import type { WorkerWorkspaceQuiescence } from "./tunnel-contract.js"; -import { - deferred, - fakeRunner, - localWorkspaceRunner, - memoryWorkspaceJournal, - startConnectedTunnel, - waitForFast, - waitForStarts, -} from "./tunnel.test-support.js"; -import { verifyReconciledWorkspaceFinal } from "./workspace-finalize.js"; - -const tunnelWarn = vi.hoisted(() => vi.fn()); -vi.mock("../../logging/subsystem.js", async (importOriginal) => { - const actual = await importOriginal(); - return { - ...actual, - createSubsystemLogger: (subsystem: string) => { - const logger = actual.createSubsystemLogger(subsystem); - return subsystem === "gateway/worker-tunnel" ? { ...logger, warn: tunnelWarn } : logger; - }, - }; -}); - -const tempDirs = useAutoCleanupTempDirTracker(afterEach); - -describe("worker workspace reconnect", () => { - beforeEach(() => { - tunnelWarn.mockClear(); - }); - - it("reconciles a completed result across a same-owner SSH reconnect", async () => { - const root = tempDirs.make("openclaw-worker-reconcile-reconnect-"); - const localPath = path.join(root, "local"); - const remoteHome = path.join(root, "remote-home"); - await Promise.all([fs.mkdir(localPath), fs.mkdir(remoteHome)]); - await fs.writeFile(path.join(localPath, "result.txt"), "before\n"); - - const releaseReconnect = deferred(); - let disconnectAfterManifest = false; - let manifestCount = 0; - const fake = localWorkspaceRunner(remoteHome, undefined, (argv) => { - if (disconnectAfterManifest && argv.at(-1)?.includes("'memo-v1'") && ++manifestCount === 1) { - disconnectAfterManifest = false; - fake.starts[0]!.process.exit(255); - } - }); - const { handle, manager } = await startConnectedTunnel(fake, "worker:reconcile-reconnect", 13, { - manager: { - sleep: async (_ms, signal) => { - await Promise.race([ - releaseReconnect.promise, - new Promise((_resolve, reject) => { - signal?.addEventListener( - "abort", - () => - reject( - signal.reason instanceof Error - ? signal.reason - : new Error("reconnect sleep aborted"), - ), - { once: true }, - ); - }), - ]); - }, - }, - }); - - try { - const synced = await handle.syncWorkspace({ - localPath, - sessionId: "session:reconcile-reconnect", - generation: 1, - }); - await fs.writeFile(path.join(synced.remoteWorkspaceDir, "result.txt"), "after\n"); - const reconciliation = await handle.reconcileWorkspace({ - localPath, - remoteWorkspaceDir: synced.remoteWorkspaceDir, - baseManifestRef: synced.manifestRef, - journal: memoryWorkspaceJournal(), - }); - const quiescence: WorkerWorkspaceQuiescence = { - assertActive: async () => { - const result = await handle.runWorkspaceCommand({ - transportRetry: "never", - argv: ["pwd"], - }); - expect(result.code).toBe(0); - }, - resume: async () => {}, - }; - manifestCount = 0; - disconnectAfterManifest = true; - const finalizing = verifyReconciledWorkspaceFinal(reconciliation, quiescence); - const finalizationSettled = vi.fn(); - void finalizing.then(finalizationSettled, finalizationSettled); - - await waitForFast(() => - expect(manager.status("worker:reconcile-reconnect")).toBe("reconnecting"), - ); - await new Promise((resolve) => { - setTimeout(resolve, 100); - }); - expect(finalizationSettled).not.toHaveBeenCalled(); - - releaseReconnect.resolve(); - await waitForStarts(fake.starts, 2); - fake.starts[1]!.process.becomeReady(); - - await expect(finalizing).resolves.toBeDefined(); - await expect(fs.readFile(path.join(localPath, "result.txt"), "utf8")).resolves.toBe( - "after\n", - ); - } finally { - releaseReconnect.resolve(); - await handle.stop(); - } - }); - - it("waits for a same-owner reconnect before initial workspace sync", async () => { - const root = tempDirs.make("openclaw-worker-sync-reconnect-"); - const localPath = path.join(root, "local"); - const remoteHome = path.join(root, "remote-home"); - await Promise.all([fs.mkdir(localPath), fs.mkdir(remoteHome)]); - await fs.writeFile(path.join(localPath, "input.txt"), "ready\n"); - - const releaseReconnect = deferred(); - const fake = localWorkspaceRunner(remoteHome); - const { handle, manager } = await startConnectedTunnel(fake, "worker:sync-reconnect", 15, { - manager: { - sleep: async () => await releaseReconnect.promise, - }, - }); - - try { - fake.starts[0]!.process.exit(255); - await waitForFast(() => expect(manager.status("worker:sync-reconnect")).toBe("reconnecting")); - const syncing = handle.syncWorkspace({ - localPath, - sessionId: "session:sync-reconnect", - generation: 1, - }); - const syncSettled = vi.fn(); - void syncing.then(syncSettled, syncSettled); - await new Promise((resolve) => { - setTimeout(resolve, 100); - }); - expect(syncSettled).not.toHaveBeenCalled(); - - releaseReconnect.resolve(); - await waitForStarts(fake.starts, 2); - fake.starts[1]!.process.becomeReady(); - await expect(syncing).resolves.toMatchObject({ - mode: "plain", - remoteWorkspaceDir: expect.any(String), - manifestRef: expect.stringMatching(/^sha256:/u), - }); - } finally { - releaseReconnect.resolve(); - await handle.stop(); - } - }); - - it("logs an in-flight child exit and the reconnect attempt before readiness", async () => { - const commandStarted = deferred(); - const releaseCommand = deferred(); - const fake = fakeRunner(async (argv) => { - if (argv.at(-1)?.includes("'pwd'")) { - commandStarted.resolve(); - await releaseCommand.promise; - } - return undefined; - }); - const { handle } = await startConnectedTunnel(fake, "worker:reconnect-diagnostics", 14, { - manager: { sleep: async () => {} }, - }); - - try { - const running = handle.runWorkspaceCommand({ transportRetry: "idempotent", argv: ["pwd"] }); - await commandStarted.promise; - fake.starts[0]!.process.exit(255, "ssh transport closed"); - - await waitForStarts(fake.starts, 2); - expect(tunnelWarn).toHaveBeenCalledWith( - "worker tunnel SSH child exited during workspace operation", - expect.objectContaining({ - environmentId: "worker:reconnect-diagnostics", - ownerEpoch: 14, - exitCode: 255, - signal: null, - stderrTail: "ssh transport closed", - workspaceTaskCount: 1, - }), - ); - expect(tunnelWarn).toHaveBeenCalledWith( - "worker tunnel reconnect attempt started", - expect.objectContaining({ - environmentId: "worker:reconnect-diagnostics", - ownerEpoch: 14, - attempt: 2, - status: "reconnecting", - port: expect.any(Number), - workspaceTaskCount: 1, - }), - ); - fake.starts[1]!.process.becomeReady(); - releaseCommand.resolve(); - await expect(running).resolves.toMatchObject({ code: 0 }); - } finally { - releaseCommand.resolve(); - await handle.stop(); - } - }); -}); diff --git a/src/gateway/worker-environments/workspace-sync-tunnel.test.ts b/src/gateway/worker-environments/workspace-sync-tunnel.test.ts index 9db10532b390..060ecfc0bb5e 100644 --- a/src/gateway/worker-environments/workspace-sync-tunnel.test.ts +++ b/src/gateway/worker-environments/workspace-sync-tunnel.test.ts @@ -21,7 +21,6 @@ import { startConnectedTunnel, success, waitForFast, - waitForStarts, workspaceSetup, } from "./tunnel.test-support.js"; import { rsyncArgvPort, sshArgvPort } from "./worker-ssh-argv.test-support.js"; @@ -177,7 +176,6 @@ describe("worker tunnel manager", () => { }); const { handle } = await startConnectedTunnel(fake, "worker:fallback-sync", 1, { ssh: endpoint, - beforeReady: (start) => expect(sshArgvPort(start.argv)).toBe(2222), }); try { @@ -209,7 +207,7 @@ describe("worker tunnel manager", () => { const knownHostsOption = fake.runs[0]!.argv.find((value) => value.startsWith("UserKnownHostsFile="), )!; - for (const connection of [...freshConnections, ...fake.starts]) { + for (const connection of freshConnections) { expect(connection.argv.join(" ")).toContain(identityPath); expect(connection.argv.join(" ")).toContain(knownHostsOption); } @@ -386,17 +384,13 @@ describe("worker tunnel manager", () => { }, ); const manager = createWorkerTunnelManager({ runner: fake.runner }); - const starting = manager.start({ + const handle = await manager.start({ bundleHash: BUNDLE_HASH, environmentId: "worker:convergent-sync", ownerEpoch: 1, ssh: { ...SSH, port: 2222, fallbackPorts: [22] }, - gateway: { host: "127.0.0.1", port: 18789 }, resolveIdentity, }); - await waitForStarts(fake.starts, 1); - fake.starts[0]!.process.becomeReady(); - const handle = await starting; try { const syncing = handle.syncWorkspace({ @@ -592,17 +586,13 @@ describe("worker tunnel manager", () => { }, ); const manager = createWorkerTunnelManager({ runner: fake.runner }); - const starting = manager.start({ + const handle = await manager.start({ bundleHash: BUNDLE_HASH, environmentId: "worker:retry-owner", ownerEpoch: 1, ssh: { ...SSH, port: 2222, fallbackPorts: [22] }, - gateway: { host: "127.0.0.1", port: 18789 }, resolveIdentity, }); - await waitForStarts(fake.starts, 1); - fake.starts[0]!.process.becomeReady(); - const handle = await starting; try { await expect( diff --git a/src/worker/worker-fault-injection.test-support.ts b/src/worker/worker-fault-injection.test-support.ts index c95bcd4953ab..2ce69b5b7666 100644 --- a/src/worker/worker-fault-injection.test-support.ts +++ b/src/worker/worker-fault-injection.test-support.ts @@ -657,7 +657,7 @@ export class ComposedGatewayHarness { return result; }, } as workerServer.WorkerConnectionService, - ingress: "loopback", + publicAdmission: { clientIp: "127.0.0.1", rateLimiter: undefined }, send: (frame) => this.send(socket, frame), close: (code = 1000, reason = "") => socket.close(code, reason), isClosed: () => closed || socket.readyState === WebSocket.CLOSED, diff --git a/src/worker/worker.fault-injection.test.ts b/src/worker/worker.fault-injection.test.ts index d1747ff9d993..b942d0e0c380 100644 --- a/src/worker/worker.fault-injection.test.ts +++ b/src/worker/worker.fault-injection.test.ts @@ -232,11 +232,11 @@ describe("cloud worker milestone 2 fault injection", () => { const commit = current.transcript.commit([transcriptMessage("restart transcript")]); const fencedCommit = expect(commit).rejects.toMatchObject({ name: "WorkerAdmissionError", - reason: "placement-mismatch", + reason: "invalid-handshake", }); const fencedInference = expect(inference).rejects.toMatchObject({ name: "WorkerAdmissionError", - reason: "placement-mismatch", + reason: "invalid-handshake", }); await commitEntered.promise; for (const delta of ["tail-a", "tail-b"]) { @@ -258,7 +258,7 @@ describe("cloud worker milestone 2 fault injection", () => { kind: "failed", error: expect.objectContaining({ name: WorkerAdmissionError.name, - reason: "placement-mismatch", + reason: "invalid-handshake", }), }); expect(harness.providerCalls).toBe(1);