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

@@ -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

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.

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.

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:
 

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();
-    }
-  });
-}

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";

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,

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,
   };
 }

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<number>;
@@ -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,

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 () => {

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: () => {} }) },

src/gateway/server-worker-environment-startup.ts

@@ -34,7 +34,6 @@ type WorkerEnvironmentStore = ReturnType<
   typeof import("./worker-environments/store.js").createWorkerEnvironmentStore
 >;
 type WorkerEnvironmentRecord = ReturnType<WorkerEnvironmentStore["list"]>[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<Gatewa
 
 export async function createGatewayWorkerEnvironmentRuntime(params: {
   getPluginRegistry: () => Pick<PluginRegistry, "workerProviders">;
-  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,

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<GatewayIngressWebSocket>((resolve) => {
-      wss.once("connection", (socket) => {
-        resolve(socket as GatewayIngressWebSocket);
-      });
-    });
-    attachWorkerGatewayUpgradeHandler({
-      httpServer,
-      wss,
-      preauthConnectionBudget: workerBudget,
-    });
-    await new Promise<void>((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<void>((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<void>((resolve) => {
-        client.once("close", () => resolve());
-      });
-      await new Promise<void>((resolve) => {
-        wss.close(() => resolve());
-      });
-      await new Promise<void>((resolve, reject) => {
-        httpServer.close((error) => (error ? reject(error) : resolve()));
-      });
-    }
-  });
-
   it("reserves the public worker path before plugin upgrade routing", async () => {
     const clients = new Set<GatewayWsClient>();
     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<GatewayWsClient>();
+    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<void>((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",
       });

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<void>((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<void>((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()));

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<void> {
-    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<SpawnResult> {
@@ -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",

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<unknown>([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");
   });
 

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,

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();
   });
 

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",

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;
 };

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,
       }),
     );

src/gateway/worker-environments/environment-access.ts

@@ -22,7 +22,6 @@ type WorkerEnvironmentAccessOptions = {
   prepareCurrentBundle: () => Promise<ExpectedWorkerBuild>;
   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),

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<WorkerTunnelHandle["launchTurn"]>[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<void>;
   launchTasks: Set<Promise<unknown>>;
-  readiness: Deferred<WorkerTunnelHandle>;
+  readiness: Deferred<WorkerTurnTunnelHandle>;
   stopPromise?: Promise<void>;
 };
 
@@ -311,7 +312,7 @@ export function createNodeWorkerTunnelManager(options: NodeWorkerTunnelManagerOp
   const createHandle = (
     entry: Omit<NodeTunnelEntry, "handle" | "readiness" | "initialization">,
     restoredWorkspace: NodeWorkerWorkspaceBinding | undefined,
-  ): { handle: WorkerTunnelHandle; validateRestoredWorkspace: () => Promise<void> } => {
+  ): { handle: WorkerTurnTunnelHandle; validateRestoredWorkspace: () => Promise<void> } => {
     let workspaceReady = restoredWorkspace !== undefined;
     const exec = async (command: Parameters<typeof runWorkspaceCommand>[2]) => {
       if (!workspaceReady) {
@@ -379,7 +380,7 @@ export function createNodeWorkerTunnelManager(options: NodeWorkerTunnelManagerOp
       }
     };
     const reconcileWorkspace = async (
-      request: Parameters<WorkerTunnelHandle["reconcileWorkspace"]>[0],
+      request: Parameters<WorkerTurnTunnelHandle["reconcileWorkspace"]>[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<WorkerTunnelHandle> {
+    async start(request: NodeWorkerTunnelStartRequest): Promise<WorkerTurnTunnelHandle> {
       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<WorkerTunnelHandle>();
+      const readiness = createDeferredCore<WorkerTurnTunnelHandle>();
       void readiness.promise.catch(() => undefined);
       const entry: NodeTunnelEntry = {
         ...request,

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",

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,

src/gateway/worker-environments/ssh.ts

@@ -258,23 +258,6 @@ export async function runWorkerSshCandidates<T extends WorkerSshCommandResult>(
   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,

src/gateway/worker-environments/tunnel-contract.ts

@@ -99,7 +99,7 @@ export type WorkerWorkspaceQuiescence = {
   resume(): Promise<void>;
 };
 
-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<SpawnResult>;
+  launchTurn?: never;
   runWorkspaceCommand(command: WorkerWorkspaceCommand): Promise<SpawnResult>;
   quiesceWorkspace(remoteWorkspaceDir: string): Promise<WorkerWorkspaceQuiescence>;
   syncWorkspace(request: WorkerWorkspaceSyncRequest): Promise<WorkerWorkspaceSyncResult>;
@@ -119,3 +119,9 @@ export type WorkerTunnelHandle = {
   ): Promise<WorkerWorkspaceReconcileResult>;
   stop(): Promise<void>;
 };
+
+export type WorkerTurnTunnelHandle = Omit<WorkerWorkspaceTunnelHandle, "launchTurn"> & {
+  launchTurn(request: WorkerTurnLaunchRequest): Promise<SpawnResult>;
+};
+
+export type WorkerTunnelHandle = WorkerWorkspaceTunnelHandle | WorkerTurnTunnelHandle;

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<ReturnType<typeof fakeRunner>, "runner" | "starts">;
-type TunnelManagerOptions = NonNullable<Parameters<typeof createWorkerTunnelManager>[0]>;
+type TunnelTestFake = Pick<ReturnType<typeof fakeRunner>, "runner">;
 type TunnelManager = ReturnType<typeof createWorkerTunnelManager>;
 
 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<TunnelManagerOptions, "runner">;
-    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 };
 }

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<AbortSignal>();
-    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<void>((_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<void>();
-    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<AbortSignal>();
-    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<void>((_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<Awaited<ReturnType<typeof resolveIdentity>>>();
     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<void>();
-    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");
   });
 });
 

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<WorkerTunnelStatus, "stopped">;
   prepared?: PreparedWorkerSsh;
-  process?: WorkerSshProcess;
-  initialization?: Promise<void>;
-  loop?: Promise<void>;
-  loopSettled: boolean;
+  initialization?: Promise<WorkerTunnelHandle>;
   stopPromise?: Promise<void>;
-  readiness: Deferred<WorkerTunnelHandle>;
   workspaceTasks: Set<Promise<unknown>>;
 };
 
 type WorkerTunnelManagerOptions = {
   runner?: WorkerSshRunner;
   desktopSessionRegistry?: DesktopSessionRegistry;
-  sleep?: (ms: number, signal?: AbortSignal) => Promise<void>;
-  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<PreparedWorkerSsh> => {
-      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<WorkerTunnelHandle>();
-            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<void> => {
     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<WorkerTunnelHandle>();
-    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<void> {
@@ -560,8 +168,8 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions =
 
   async function stopAll(): Promise<void> {
     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";
     },
   };
 }

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<WorkerTunnelHandle["launchTurn"]>[0]) => {
+    const launchTurn = vi.fn((request: WorkerTurnLaunchRequest) => {
       request.onDispatchReady?.();
       commandStarted.resolve();
       return commandFinished.promise;

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<WorkerTunnelHandle["launchTurn"]>[0]): Promise<SpawnResult> => {
-        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<SpawnResult> => {
+      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()),

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,

src/gateway/worker-environments/workspace-sync-reconnect.test.ts

@@ -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<typeof import("../../logging/subsystem.js")>();
-  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<void>();
-    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<never>((_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<void>((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<void>();
-    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<void>((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<void>();
-    const releaseCommand = deferred<void>();
-    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();
-    }
-  });
-});

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(

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,

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);