fix(gateway): unbreak cloud session creation — stale chat metadata, unbounded tunnel hangs, swallowed dispatch errors (#120926)

* fix(gateway): rebuild chat metadata when auth-profile snapshots change

chat.metadata cached a prepared generation built before the runtime
auth-profile store snapshot was published (empty-store fallback) and
generationFactsMatch never compared auth state, so the Control UI showed
"No models available" after a gateway restart until an unrelated config
edit. Capture per-agent auth snapshot revisions in the generation facts,
subscribe the metadata lifecycle to auth-store mutations, and run one
awaited revision-aware catch-up refresh after listener registration so
publications that precede registration are still observed.

* fix(gateway): bound cloud worker tunnel startup and surface dispatch failure detail

Live stress-testing the Control UI cloud flow found sessions.dispatch
hanging unbounded (observed 12+ min) when the worker SSH tunnel could
not connect: the runner's exited promise settled only on "close" (a
spawn error never settled it), the reconnect loop swallowed every
failure with no logging, the poisoned ready promise was re-handed to
every later dispatch, and dispatch errors dropped the actionable reason
recorded in worker_environments.last_error.

- settle exited on the real "exit" event, wire the owner abort signal
  into spawn, bound stop()'s post-SIGKILL wait, and fail stop honestly
  when termination is unconfirmed instead of fabricating an exit
- add a 60s per-attempt readiness deadline and log each failed connect
  attempt (bounded, redacted); keep an unconfirmed child tracked and
  wait for its real exit before retrying
- add a 3-minute epoch-fenced startTunnel deadline with a typed,
  actionable error; detach its cleanup so the deadline holds
- append the bounded recorded reason to the five dispatch-visible
  worker environment error messages (docs already promise these)
- remote socket setup: drop "--" from chmod (BSD/macOS chmod treats it
  as a filename), which blocked every tunnel to a macOS worker host
- workspace quiescence: tolerate EPERM without crashing the protocol
  while keeping unsignalable freeze targets counted as live so
  quiescence fails closed

* test(gateway): type worker child kill mock

src/agents/auth-profiles.ts

@@ -63,6 +63,7 @@ export {
   ensureAuthProfileStoreWithoutExternalProfiles,
   getPreparedRuntimeAuthProfileStoreSnapshot,
   getRuntimeAuthProfileStoreSnapshot,
+  getRuntimeAuthProfileStoreSnapshotRevision,
   hasAuthProfileStoreSourceForProvider,
   hasAnyAuthProfileStoreSource,
   hasLocalAuthProfileStoreSource,

src/agents/auth-profiles/store.ts

@@ -1326,6 +1326,8 @@ export function getPreparedRuntimeAuthProfileStoreSnapshot(
   return getPreparedRuntimeAuthProfileStoreSnapshotImpl(agentDir, inheritedAuthDir);
 }
 
+export { getRuntimeAuthProfileStoreSnapshotRevision };
+
 /** Replace runtime auth-profile snapshots, used by tests and prepared runtimes. */
 export function replaceRuntimeAuthProfileStoreSnapshots(
   entries: Array<{ agentDir?: string; store: AuthProfileStore }>,

src/gateway/server-chat-metadata-lifecycle.ts

@@ -46,11 +46,15 @@ export async function createGatewayChatMetadataLifecycle(params: {
     if (params.minimalTestGateway) {
       return undefined;
     }
-    const [{ registerPreparedModelRuntimePublicationListener }, { registerSkillsChangeListener }] =
-      await Promise.all([
-        import("../agents/prepared-model-runtime.js"),
-        import("../skills/runtime/refresh.js"),
-      ]);
+    const [
+      { registerRuntimeAuthProfileStoreMutationListener },
+      { registerPreparedModelRuntimePublicationListener },
+      { registerSkillsChangeListener },
+    ] = await Promise.all([
+      import("../agents/auth-profiles/runtime-snapshots.js"),
+      import("../agents/prepared-model-runtime.js"),
+      import("../skills/runtime/refresh.js"),
+    ]);
     const unregisterPreparedModelRuntimePublication =
       registerPreparedModelRuntimePublicationListener((event) => {
         if (event.phase === "invalidated") {
@@ -67,8 +71,14 @@ export async function createGatewayChatMetadataLifecycle(params: {
       runtime.invalidate();
       refreshLogged();
     });
+    const unregisterRuntimeAuthProfileStoreMutation =
+      registerRuntimeAuthProfileStoreMutationListener(() => {
+        runtime.invalidate();
+        refreshLogged();
+      });
     return {
       stop: async () => {
+        unregisterRuntimeAuthProfileStoreMutation();
         unregisterPreparedModelRuntimePublication();
         unregisterSkillsChange();
       },
@@ -85,6 +95,13 @@ export async function createGatewayChatMetadataLifecycle(params: {
       if (sidecar) {
         sidecars.push(sidecar);
       }
+      // Auth/model snapshots published before listener registration are otherwise never
+      // observed; one awaited catch-up refresh reconciles facts (revision-aware, no-op when
+      // fresh) so post-attach reads cannot serve a pre-publication generation. Failures are
+      // logged, not thrown: startup must not die on a metadata build error.
+      await runtime.refresh().catch((error: unknown) => {
+        params.log.warn(`chat metadata catch-up refresh failed: ${String(error)}`);
+      });
     },
     read: runtime.read,
     readStartup: runtime.readStartup,

src/gateway/server-methods/chat-metadata-runtime.test.ts

@@ -37,16 +37,22 @@ function createHarness(
   let owner = createOwner(config, "first");
   let skillsVersion = 1;
   let pluginRegistryVersion = 1;
-  const authStore: AuthProfileStore = { version: 1, profiles: {} };
+  let authStore: AuthProfileStore | undefined = { version: 1, profiles: {} };
+  let authStoreRevision = 1;
   const getPreparedOwner = vi.fn(() => owner);
   const getPreparedAuthStore = vi.fn(() => authStore);
+  const getAuthStoreRevision = vi.fn(() => authStoreRevision);
   const getSkillsVersion = vi.fn(() => skillsVersion);
   const getPluginRegistryVersion = vi.fn(() => pluginRegistryVersion);
   const buildCommands = vi.fn(async () => ({
     commands: [{ name: `command-${skillsVersion}-${pluginRegistryVersion}` }],
   }));
   const buildProjection = vi.fn(
-    async ({ facts }: { facts: { owner: PreparedModelRuntimeSnapshot } }) => ({
+    async ({
+      facts,
+    }: {
+      facts: { authStore: AuthProfileStore; owner: PreparedModelRuntimeSnapshot };
+    }) => ({
       modelCatalog: facts.owner.modelCatalog.entries,
       models: facts.owner.modelCatalog.entries,
     }),
@@ -67,6 +73,7 @@ function createHarness(
     deps: {
       getPreparedOwner,
       getPreparedAuthStore,
+      getAuthStoreRevision,
       getSkillsVersion,
       getPluginRegistryVersion,
       buildCommands,
@@ -77,6 +84,7 @@ function createHarness(
     buildCommands,
     buildProjection,
     getPluginRegistryVersion,
+    getAuthStoreRevision,
     getPreparedAuthStore,
     getPreparedOwner,
     getSkillsVersion,
@@ -84,6 +92,12 @@ function createHarness(
     setConfig(next: OpenClawConfig) {
       config = next;
     },
+    setAuthStore(next: AuthProfileStore | undefined) {
+      authStore = next;
+    },
+    setAuthStoreRevision(next: number) {
+      authStoreRevision = next;
+    },
     setOwner(next: PreparedModelRuntimeSnapshot) {
       owner = next;
     },
@@ -130,6 +144,7 @@ describe("gateway chat metadata runtime", () => {
     await harness.runtime.refresh();
     harness.getPreparedOwner.mockClear();
     harness.getPreparedAuthStore.mockClear();
+    harness.getAuthStoreRevision.mockClear();
     harness.getSkillsVersion.mockClear();
     harness.getPluginRegistryVersion.mockClear();
 
@@ -139,6 +154,7 @@ describe("gateway chat metadata runtime", () => {
     expect(first).toBe(second);
     expect(harness.getPreparedOwner).not.toHaveBeenCalled();
     expect(harness.getPreparedAuthStore).not.toHaveBeenCalled();
+    expect(harness.getAuthStoreRevision).not.toHaveBeenCalled();
     expect(harness.getSkillsVersion).not.toHaveBeenCalled();
     expect(harness.getPluginRegistryVersion).not.toHaveBeenCalled();
   });
@@ -148,6 +164,7 @@ describe("gateway chat metadata runtime", () => {
     await harness.runtime.refresh();
     harness.getPreparedOwner.mockClear();
     harness.getPreparedAuthStore.mockClear();
+    harness.getAuthStoreRevision.mockClear();
     harness.getSkillsVersion.mockClear();
     harness.getPluginRegistryVersion.mockClear();
 
@@ -169,6 +186,7 @@ describe("gateway chat metadata runtime", () => {
     expect(harness.buildProjection).toHaveBeenCalledTimes(1);
     expect(harness.getPreparedOwner).not.toHaveBeenCalled();
     expect(harness.getPreparedAuthStore).not.toHaveBeenCalled();
+    expect(harness.getAuthStoreRevision).not.toHaveBeenCalled();
     expect(harness.getSkillsVersion).not.toHaveBeenCalled();
     expect(harness.getPluginRegistryVersion).not.toHaveBeenCalled();
   });
@@ -322,6 +340,50 @@ describe("gateway chat metadata runtime", () => {
     expect(harness.buildProjection).toHaveBeenCalledTimes(1);
   });
 
+  test("rebuilds after an auth store publishes a newer revision", async () => {
+    const harness = createHarness();
+    harness.setAuthStore(undefined);
+    harness.buildProjection.mockImplementation(async ({ facts }) => ({
+      modelCatalog: facts.owner.modelCatalog.entries,
+      models: facts.owner.modelCatalog.entries.map((model) => ({
+        ...model,
+        available: Object.keys(facts.authStore.profiles).length > 0,
+      })),
+    }));
+
+    await harness.runtime.refresh();
+    await expect(harness.runtime.read({ agentId: "main" })).resolves.toMatchObject({
+      models: [expect.objectContaining({ available: false })],
+    });
+
+    harness.setAuthStore({
+      version: 1,
+      profiles: { "test:default": { type: "api_key", provider: "test" } },
+    });
+    harness.setAuthStoreRevision(2);
+    await harness.runtime.refresh();
+
+    await expect(harness.runtime.read({ agentId: "main" })).resolves.toMatchObject({
+      models: [expect.objectContaining({ available: true })],
+    });
+    expect(harness.buildProjection).toHaveBeenCalledTimes(2);
+  });
+
+  test("retains a generation while auth store revisions are unchanged", async () => {
+    const harness = createHarness();
+    harness.getPreparedAuthStore.mockImplementation(() => ({ version: 1, profiles: {} }));
+    await harness.runtime.refresh();
+    const first = await harness.runtime.read({ agentId: "main" });
+
+    await harness.runtime.refresh();
+    const second = await harness.runtime.read({ agentId: "main" });
+
+    expect(second).toBe(first);
+    expect(harness.buildProjection).toHaveBeenCalledTimes(1);
+    expect(harness.getAuthStoreRevision).toHaveBeenCalledWith("/tmp/first/agent");
+    expect(harness.getAuthStoreRevision).toHaveBeenCalledWith(undefined);
+  });
+
   test("refreshes config, catalog-auth, skills, and plugin generations", async () => {
     const harness = createHarness();
     await harness.runtime.refresh();

src/gateway/server-methods/chat-metadata-runtime.ts

@@ -5,6 +5,7 @@ import {
 } from "../../agents/agent-scope.js";
 import {
   getPreparedRuntimeAuthProfileStoreSnapshot,
+  getRuntimeAuthProfileStoreSnapshotRevision,
   type AuthProfileStore,
 } from "../../agents/auth-profiles.js";
 import type { ModelCatalogEntry } from "../../agents/model-catalog.types.js";
@@ -40,6 +41,7 @@ type PreparedAgentFacts = {
   agentId: string;
   owner: PreparedModelRuntimeSnapshot;
   authStore: AuthProfileStore;
+  authStoreRevision: string;
   skillsVersion: number;
 };
 
@@ -85,6 +87,7 @@ type ChatMetadataRuntimeDeps = {
     agentDir?: string,
     inheritedAuthDir?: string,
   ) => AuthProfileStore | undefined;
+  getAuthStoreRevision: (agentDir?: string) => number;
   getSkillsVersion: (workspaceDir?: string) => number;
   getPluginRegistryVersion: () => number;
   buildCommands: (params: {
@@ -150,6 +153,7 @@ function captureGenerationFacts(deps: ChatMetadataRuntimeDeps): PreparedGenerati
         version: 1,
         profiles: {},
       },
+      authStoreRevision: `${deps.getAuthStoreRevision(owner.agentDir)}:${deps.getAuthStoreRevision(owner.inheritedAuthDir)}`,
       skillsVersion: deps.getSkillsVersion(workspaceDir),
     };
   });
@@ -177,6 +181,7 @@ function generationFactsMatch(
     return (
       candidate?.agentId === agent.agentId &&
       candidate.owner === agent.owner &&
+      candidate.authStoreRevision === agent.authStoreRevision &&
       candidate.skillsVersion === agent.skillsVersion
     );
   });
@@ -277,6 +282,7 @@ export function createGatewayChatMetadataRuntime(params: {
     getContext: params.getContext,
     getPreparedOwner: getPreparedModelCatalogOwnerSnapshot,
     getPreparedAuthStore: getPreparedRuntimeAuthProfileStoreSnapshot,
+    getAuthStoreRevision: getRuntimeAuthProfileStoreSnapshotRevision,
     getSkillsVersion: getSkillsSnapshotVersion,
     getPluginRegistryVersion: getActivePluginRegistryVersion,
     buildCommands: defaultBuildCommands,

src/gateway/worker-environments/service.test.ts

@@ -1164,6 +1164,7 @@ describe("worker environment service", () => {
       workerService.create("development", "request-preparation-failure"),
     ).rejects.toMatchObject({
       code: "bootstrap_failure",
+      message: expect.stringContaining("npm install requires a released gateway package"),
     } satisfies Partial<WorkerEnvironmentServiceError>);
 
     expect(provision).not.toHaveBeenCalled();
@@ -1589,6 +1590,7 @@ describe("worker environment service", () => {
 
     await expect(workerService.create("development", "request-malformed")).rejects.toMatchObject({
       code: "provider_failure",
+      message: expect.stringContaining(error),
     } satisfies Partial<WorkerEnvironmentServiceError>);
     expect(store.list()[0]).toMatchObject({
       state: "provisioning",
@@ -1621,6 +1623,7 @@ describe("worker environment service", () => {
 
     await expect(workerService.create("development", "request-invalid")).rejects.toMatchObject({
       code: "invalid_profile",
+      message: expect.stringContaining("region is required"),
     } satisfies Partial<WorkerEnvironmentServiceError>);
     const record = expectDefined(store.list()[0], "store.list()[0] test invariant");
     expect(record).toMatchObject({ state: "failed", lastError: "region is required" });
@@ -2258,6 +2261,45 @@ describe("worker environment service", () => {
     expect(order).toEqual(["tunnel-stop", "provider-destroy"]);
   });
 
+  it("stops a poisoned tunnel start and returns a typed deadline error", async () => {
+    vi.useFakeTimers();
+    seedReady("worker-tunnel-timeout");
+    let signalStarted!: () => void;
+    const started = new Promise<void>((resolve) => {
+      signalStarted = resolve;
+    });
+    let rejectStart!: (error: Error) => void;
+    const pendingStart = new Promise<never>((_resolve, reject) => {
+      rejectStart = reject;
+    });
+    const tunnelManager = {
+      status: () => "connecting" as const,
+      start: vi.fn(() => {
+        signalStarted();
+        return pendingStart;
+      }),
+      stop: vi.fn(async () => {
+        rejectStart(new Error("tunnel stopped"));
+      }),
+      stopAll: vi.fn(async () => {}),
+    } as unknown as WorkerTunnelManager;
+    const workerService = createService(createProvider(), { tunnelManager });
+
+    const starting = workerService.startTunnel({
+      environmentId: "worker-tunnel-timeout",
+      ownerEpoch: 1,
+    });
+    const rejected = expect(starting).rejects.toMatchObject({
+      code: "provider_failure",
+      message: expect.stringContaining("did not connect within 3 minutes"),
+    } satisfies Partial<WorkerEnvironmentServiceError>);
+    await started;
+    await vi.advanceTimersByTimeAsync(3 * 60_000);
+
+    await rejected;
+    expect(tunnelManager.stop).toHaveBeenCalledWith("worker-tunnel-timeout", 1);
+  });
+
   it("adopts an unpersisted provision result before destroying", async () => {
     const intent = store.createIntent({
       environmentId: "worker-pending-destroy",

src/gateway/worker-environments/service.ts

@@ -92,6 +92,8 @@ class WorkerEnvironmentServiceError extends Error {
 const serviceError = (code: WorkerEnvironmentServiceErrorCode, message: string) =>
   new WorkerEnvironmentServiceError(code, message);
 const ORPHANED_LEASE_ERROR = "Worker provider no longer recognizes the lease";
+// One poisoned SSH attempt must not hold every later dispatch on the same owner epoch forever.
+const TUNNEL_START_TIMEOUT_MS = 3 * 60_000;
 
 function workerEnvironmentIdempotencyDigest(idempotencyKey: string): string {
   return createHash("sha256").update(idempotencyKey).digest("hex");
@@ -614,15 +616,16 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
         ),
       );
     } catch (error) {
+      const detail = boundedError(error);
       if (
         error instanceof WorkerProviderError ||
         (error instanceof WorkerEnvironmentServiceError && error.code === "invalid_profile")
       ) {
-        move(record, "failed", { lastError: boundedError(error) });
-        throw serviceError("invalid_profile", "Worker provider rejected profile");
+        move(record, "failed", { lastError: detail });
+        throw serviceError("invalid_profile", `Worker provider rejected profile: ${detail}`);
       }
       saveError(record, error);
-      throw serviceError("provider_failure", "Worker provider operation failed");
+      throw serviceError("provider_failure", `Worker provider operation failed: ${detail}`);
     }
     // A timeout can happen after allocation; retain the same operation id for safe replay.
     const patch = { leaseId: lease.leaseId, sshEndpoint: lease.ssh };
@@ -654,8 +657,12 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
         // must happen first because the previous response may have been lost after allocation.
         installation = await prepareInstallation(record);
       } catch (error) {
-        move(record, "failed", { lastError: boundedError(error) });
-        throw serviceError("bootstrap_failure", "Worker installation preparation failed");
+        const detail = boundedError(error);
+        move(record, "failed", { lastError: detail });
+        throw serviceError(
+          "bootstrap_failure",
+          `Worker installation preparation failed: ${detail}`,
+        );
       }
     }
     const provisioning = record.state === "requested" ? move(record, "provisioning") : record;
@@ -1101,7 +1108,24 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService
     if (!startup) {
       throw serviceError("invalid_state", "Worker tunnel failed to start");
     }
-    return await startup;
+    const timeoutError = serviceError(
+      "provider_failure",
+      "Worker tunnel did not connect within 3 minutes; check worker SSH reachability and retry",
+    );
+    try {
+      return await withTimeout(startup, TUNNEL_START_TIMEOUT_MS, {
+        createError: () => timeoutError,
+      });
+    } catch (error) {
+      if (error !== timeoutError) {
+        throw error;
+      }
+      // Stop can itself block on an unkillable SSH child; detach it (rejection observed,
+      // entry stays manager-tracked) so the deadline error is returned on time. Epoch-fenced
+      // so a stale timed-out attempt can never tear down a newer owner's tunnel.
+      void tunnels.stop(request.environmentId, request.ownerEpoch).catch(() => undefined);
+      throw timeoutError;
+    }
   };
 
   const stopTunnel = async (environmentId: string, ownerEpoch?: number): Promise<void> => {

src/gateway/worker-environments/tunnel-ssh-runner.test.ts

@@ -0,0 +1,169 @@
+import type { SpawnOptions } from "node:child_process";
+import { EventEmitter } from "node:events";
+import { PassThrough } from "node:stream";
+import { afterEach, beforeEach, describe, expect, it, type Mock, vi } from "vitest";
+
+const spawnMock = vi.hoisted(() => vi.fn());
+
+vi.mock("node:child_process", () => ({ spawn: spawnMock }));
+
+import { createWorkerSshRunner } from "./tunnel-ssh-runner.js";
+
+// Returned as the fake-typed union (not ChildProcessWithoutNullStreams) so `child.kill`
+// stays a plain vi.fn property; casting to the real type makes kill an unbound method for lint.
+function createChild() {
+  const child = new EventEmitter() as EventEmitter & {
+    stdin: PassThrough;
+    stdout: PassThrough;
+    stderr: PassThrough;
+    kill: Mock<(signal?: NodeJS.Signals | number) => boolean>;
+  };
+  child.stdin = new PassThrough();
+  child.stdout = new PassThrough();
+  child.stderr = new PassThrough();
+  child.kill = vi.fn(() => true);
+  return child;
+}
+
+describe("worker SSH process runner", () => {
+  beforeEach(() => {
+    spawnMock.mockReset();
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("settles readiness and exit when spawn emits an error without close", async () => {
+    const child = createChild();
+    spawnMock.mockReturnValue(child);
+    const process = createWorkerSshRunner().start(["missing-ssh"], { timeoutMs: 10_000 });
+
+    child.emit("error", new Error("spawn failed"));
+
+    await expect(process.ready).rejects.toThrow("Worker SSH tunnel failed");
+    await expect(process.exited).resolves.toEqual({ code: null, signal: null });
+  });
+
+  it("settles with the real exit when close lags after SIGKILL", async () => {
+    vi.useFakeTimers();
+    const child = createChild();
+    spawnMock.mockReturnValue(child);
+    const process = createWorkerSshRunner().start(["ssh"], { timeoutMs: 10_000 });
+
+    const stopping = process.stop();
+    expect(child.kill).toHaveBeenCalledWith("SIGTERM");
+    await vi.advanceTimersByTimeAsync(1_500);
+    expect(child.kill).toHaveBeenCalledWith("SIGKILL");
+    child.emit("exit", null, "SIGKILL");
+    await vi.advanceTimersByTimeAsync(2_000);
+
+    await expect(stopping).resolves.toBeUndefined();
+    await expect(process.exited).resolves.toEqual({ code: null, signal: "SIGKILL" });
+    await expect(process.ready).rejects.toThrow("Worker SSH tunnel failed");
+  });
+
+  it("fails stop when a SIGKILLed child never reports exit", async () => {
+    vi.useFakeTimers();
+    const child = createChild();
+    spawnMock.mockReturnValue(child);
+    const process = createWorkerSshRunner().start(["ssh"], { timeoutMs: 10_000 });
+
+    const stopping = process.stop();
+    const rejection = expect(stopping).rejects.toThrow("did not exit after SIGKILL");
+    await vi.advanceTimersByTimeAsync(1_500);
+    await vi.advanceTimersByTimeAsync(2_000);
+    await rejection;
+  });
+
+  it("treats a post-exit kill failure as terminal when close is delayed", async () => {
+    vi.useFakeTimers();
+    const child = createChild();
+    child.kill = vi.fn(() => false);
+    spawnMock.mockReturnValue(child);
+    const process = createWorkerSshRunner().start(["ssh"], { timeoutMs: 10_000 });
+
+    child.emit("exit", 255, null);
+    const stopping = process.stop();
+    await vi.advanceTimersByTimeAsync(1_500);
+    await vi.advanceTimersByTimeAsync(2_000);
+
+    await expect(stopping).resolves.toBeUndefined();
+    await expect(process.exited).resolves.toEqual({ code: 255, signal: null });
+    await expect(process.ready).rejects.toThrow("Worker SSH tunnel failed");
+  });
+
+  it("propagates a stop failure when SIGKILL delivery fails", async () => {
+    vi.useFakeTimers();
+    const child = createChild();
+    child.kill = vi.fn(() => false);
+    spawnMock.mockReturnValue(child);
+    const process = createWorkerSshRunner().start(["ssh"], { timeoutMs: 10_000 });
+
+    const stopping = process.stop();
+    const rejection = expect(stopping).rejects.toThrow("SIGKILL delivery failed");
+    await vi.advanceTimersByTimeAsync(1_500);
+    await vi.advanceTimersByTimeAsync(2_000);
+    await rejection;
+
+    let exitedSettled = false;
+    void process.exited.then(() => {
+      exitedSettled = true;
+    });
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(exitedSettled).toBe(false);
+  });
+
+  it("keeps exited pending when a live child emits error without close", async () => {
+    const child = createChild();
+    (child as { pid?: number }).pid = 4242;
+    spawnMock.mockReturnValue(child);
+    const process = createWorkerSshRunner().start(["ssh"], { timeoutMs: 10_000 });
+
+    child.emit("error", new Error("kill delivery failed"));
+
+    await expect(process.ready).rejects.toThrow("Worker SSH tunnel failed");
+    let exitedSettled = false;
+    void process.exited.then(() => {
+      exitedSettled = true;
+    });
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(exitedSettled).toBe(false);
+
+    child.emit("close", 0, null);
+    await expect(process.exited).resolves.toEqual({ code: 0, signal: null });
+  });
+
+  it("passes the owner signal to spawn so abort terminates and settles the child", async () => {
+    const child = createChild();
+    spawnMock.mockImplementation((_command: string, _args: string[], options: SpawnOptions) => {
+      options.signal?.addEventListener(
+        "abort",
+        () => {
+          child.kill("SIGTERM");
+          child.emit("error", Object.assign(new Error("aborted"), { name: "AbortError" }));
+        },
+        { once: true },
+      );
+      return child;
+    });
+    const controller = new AbortController();
+    const process = createWorkerSshRunner().start(["ssh"], {
+      timeoutMs: 10_000,
+      signal: controller.signal,
+    });
+
+    controller.abort();
+
+    expect(child.kill).toHaveBeenCalledWith("SIGTERM");
+    await expect(process.ready).rejects.toThrow("Worker SSH tunnel failed");
+    await expect(process.exited).resolves.toEqual({ code: null, signal: null });
+    expect(spawnMock).toHaveBeenCalledWith(
+      "ssh",
+      [],
+      expect.objectContaining({ signal: controller.signal }),
+    );
+  });
+});

src/gateway/worker-environments/tunnel-ssh-runner.ts

@@ -10,6 +10,7 @@ import {
 export const WORKER_TUNNEL_READY_MARKER = "OPENCLAW_WORKER_TUNNEL_READY";
 
 const STOP_GRACE_MS = 1_500;
+const STOP_KILL_WAIT_MS = 2_000;
 const STDERR_LIMIT = 4_096;
 
 type WorkerSshProcessExit = {
@@ -44,10 +45,12 @@ export function createWorkerSshRunner(): WorkerSshRunner {
       }
       const child = spawn(command, args, {
         env: options.baseEnv,
+        signal: options.signal,
         stdio: ["pipe", "pipe", "pipe"],
         windowsHide: true,
       });
       let closed = false;
+      let exitedSettled = false;
       let readySettled = false;
       let resolveReady!: () => void;
       let rejectReady!: (error: Error) => void;
@@ -56,6 +59,9 @@ export function createWorkerSshRunner(): WorkerSshRunner {
         resolveReady = resolve;
         rejectReady = reject;
       });
+      // Readiness can reject after its awaiter timed out and moved on (stop()/late close);
+      // observe it here so lifecycle settles never become unhandled rejections.
+      void ready.catch(() => {});
       const exited = new Promise<WorkerSshProcessExit>((resolve) => {
         resolveExited = resolve;
       });
@@ -68,6 +74,13 @@ export function createWorkerSshRunner(): WorkerSshRunner {
         readySettled = true;
         rejectReady(workerSshProcessError(stderr));
       };
+      const settleExited = (exit: WorkerSshProcessExit) => {
+        if (exitedSettled) {
+          return;
+        }
+        exitedSettled = true;
+        resolveExited(exit);
+      };
       child.stdout.setEncoding("utf8");
       child.stdout.on("error", () => {});
       child.stdout.on("data", (chunk: string) => {
@@ -85,11 +98,34 @@ export function createWorkerSshRunner(): WorkerSshRunner {
       child.stderr.on("data", (chunk: string) => {
         stderr = sliceUtf16Safe(`${stderr}${chunk}`, -STDERR_LIMIT);
       });
-      child.once("error", settleReadyError);
+      child.once("error", () => {
+        settleReadyError();
+        // "error" also fires for abort/kill-delivery failures on a live child; only a child
+        // that never spawned (no pid) gets a synthesized exit, otherwise close/stop() settle it.
+        // The no-pid case is terminal: mark it closed so stop() never signals an unspawned child.
+        if (child.pid === undefined) {
+          closed = true;
+          settleExited({ code: null, signal: null });
+        }
+      });
+      // "exit" fires before "close", and "close" can be delayed indefinitely while a
+      // descendant holds a piped stdio descriptor; settle on the real exit so connected
+      // tunnels awaiting `exited` observe termination without depending on stream closure.
+      let exitEventResult: WorkerSshProcessExit | undefined;
+      child.once("exit", (code, signal) => {
+        exitEventResult = { code, signal };
+        settleReadyError();
+        settleExited(exitEventResult);
+        // Release our pipe ends so a descendant holding the other side cannot pin local
+        // descriptors across retries; this also lets "close" fire promptly.
+        child.stdin.destroy();
+        child.stdout.destroy();
+        child.stderr.destroy();
+      });
       child.once("close", (code, signal) => {
         closed = true;
         settleReadyError();
-        resolveExited({ code, signal });
+        settleExited({ code, signal });
       });
       child.stdin.on("error", () => {});
       if (options.input !== undefined) {
@@ -117,9 +153,32 @@ export function createWorkerSshRunner(): WorkerSshRunner {
               }),
             ]);
             clearTimeout(timer);
-            if (!closed) {
-              child.kill("SIGKILL");
-              await exited;
+            if (!closed && !exitedSettled) {
+              // A false return can also mean the child died a moment ago with its "exit"
+              // event still queued; always take the bounded wait before judging.
+              const killDelivered = child.kill("SIGKILL");
+              let killTimer: ReturnType<typeof setTimeout> | undefined;
+              let killWaitExpired = false;
+              await Promise.race([
+                exited,
+                new Promise<void>((resolve) => {
+                  killTimer = setTimeout(() => {
+                    killWaitExpired = true;
+                    resolve();
+                  }, STOP_KILL_WAIT_MS);
+                  killTimer.unref?.();
+                }),
+              ]);
+              clearTimeout(killTimer);
+              if (killWaitExpired) {
+                // Neither delivered SIGKILL nor failed delivery proves termination without
+                // an exit event; fail the stop so the owner keeps tracking the live child.
+                throw workerSshProcessError(
+                  killDelivered
+                    ? "SSH child did not exit after SIGKILL; it may still be running"
+                    : "SIGKILL delivery failed; SSH child may still be running",
+                );
+              }
             }
           })());
         },

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

@@ -93,6 +93,27 @@ describe("worker tunnel manager", () => {
     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({

src/gateway/worker-environments/tunnel.ts

@@ -1,8 +1,11 @@
 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 { createDeferred, type Deferred } from "../../shared/deferred.js";
+import { boundedWorkerError } from "./service-validation.js";
 import {
   advanceWorkerSshAfterTransportExit,
   prepareWorkerSsh,
@@ -30,6 +33,9 @@ import { createWorkerWorkspaceActions, stableWorkerPathComponent } from "./works
 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,
@@ -37,6 +43,7 @@ const DEFAULT_BACKOFF: BackoffPolicy = {
   factor: 2,
   jitter: 0,
 };
+const tunnelLog = createSubsystemLogger("gateway/worker-tunnel");
 
 const REMOTE_SOCKET_SETUP_SCRIPT = String.raw`set -eu
 directory=$1
@@ -50,7 +57,7 @@ if [ -e "$directory" ] || [ -L "$directory" ]; then
 else
   mkdir -- "$directory"
 fi
-chmod 700 -- "$directory"
+chmod 700 "$directory"  # no "--": BSD/macOS chmod treats it as a filename; path is script-owned and absolute
 rm -f -- "$socket"
 `;
 
@@ -281,7 +288,9 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions =
         child = connection.process;
         childPort = connection.port;
         entry.process = child;
-        await child.ready;
+        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;
@@ -306,14 +315,41 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions =
         if (now() - connectedAtMs >= stableConnectionMs) {
           reconnectSupervisor.reset();
         }
-      } catch {
+      } catch (error) {
         if (child && childPort !== undefined) {
-          const exit = await child.exited.catch(() => 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);
           }
         }
-        await child?.stop().catch(() => undefined);
+        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;

src/gateway/worker-environments/workspace-quiescence-scripts.ts

@@ -92,6 +92,10 @@ function persistLease(targetPath, lease, verifyCurrent) {
   fs.renameSync(temporary, targetPath);
 }`;
 
+// Signal sites tolerate ESRCH (gone) without aborting the protocol. EPERM (exists but
+// unsignalable, e.g. macOS SIP-protected same-uid processes on shared static-ssh dev hosts)
+// must not crash cleanup/resume paths, but a freeze target that returns EPERM stays counted
+// as live so quiescence fails closed instead of reporting a still-running process as frozen.
 export const REMOTE_WORKSPACE_QUIESCE_JS = String.raw`const childProcess = require("node:child_process");
 const crypto = require("node:crypto");
 const fs = require("node:fs");
@@ -123,13 +127,15 @@ function writeLease(expiresAtMs = Date.now() + watchdogTimeoutMs) {
     expiresAtMs,
   });
 }
+// EPERM on SIGCONT implies the target was never ours to freeze: kill permission checks are
+// identical for SIGSTOP and SIGCONT, so any process this uid successfully stopped can be resumed.
 function resumeProcesses(entries) {
   for (const entry of entries) {
     if (processIdentity(entry.pid) !== entry.start) continue;
     try {
       process.kill(entry.pid, "SIGCONT");
     } catch (error) {
-      if (!error || error.code !== "ESRCH") throw error;
+      if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error;
     }
   }
 }
@@ -143,7 +149,7 @@ for (const name of orphanNames) {
   const orphanPath = path.join(leaseDirectory, name);
   const lease = parseLease(fs.readFileSync(orphanPath, "utf8"), match[1]);
   if (lease.watchdog !== null && processIdentity(lease.watchdog.pid) === lease.watchdog.start) {
-    try { process.kill(lease.watchdog.pid, "SIGTERM"); } catch (error) { if (!error || error.code !== "ESRCH") throw error; }
+    try { process.kill(lease.watchdog.pid, "SIGTERM"); } catch (error) { if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error; }
   }
   resumeProcesses(lease.processes);
   fs.unlinkSync(orphanPath);
@@ -194,6 +200,11 @@ try {
         }
         process.kill(pid, "SIGSTOP");
       } catch (error) {
+        if (error && error.code === "EPERM") {
+          frozen.delete(pid);
+          writeLease();
+          continue;
+        }
         if (!error || error.code !== "ESRCH") throw error;
       }
     }
@@ -210,7 +221,7 @@ try {
   }
 } catch (error) {
   if (processIdentity(watchdog.pid) === watchdogStart) {
-    try { process.kill(watchdog.pid, "SIGTERM"); } catch (killError) { if (!killError || killError.code !== "ESRCH") throw killError; }
+    try { process.kill(watchdog.pid, "SIGTERM"); } catch (killError) { if (!killError || (killError.code !== "ESRCH" && killError.code !== "EPERM")) throw killError; }
   }
   resumeProcesses([...frozen].map(([pid, start]) => ({ pid, start })));
   try { fs.unlinkSync(leasePath); } catch (unlinkError) { if (!unlinkError || unlinkError.code !== "ENOENT") throw unlinkError; }
@@ -254,7 +265,7 @@ function watchdogMain(watchedLeasePath, watchedNonce) {
           typeof entry.start !== "string" ||
           processIdentity(entry.pid) !== entry.start
         ) continue;
-        try { process.kill(entry.pid, "SIGCONT"); } catch (error) { if (!error || error.code !== "ESRCH") throw error; }
+        try { process.kill(entry.pid, "SIGCONT"); } catch (error) { if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error; }
       }
       watchdogFs.unlinkSync(watchedLeasePath);
     } catch (error) {
@@ -345,7 +356,9 @@ if (validationMode === "final") {
         if (input.expiresAtMs - Date.now() < 2500) refreshLease(frozenEntries);
         process.kill(pid, "SIGSTOP");
       } catch (error) {
-        if (!error || error.code !== "ESRCH") throw error;
+        if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error;
+        // Fail-closed either way: the candidate scan below runs without the frozen filter,
+        // so an EPERM-live process re-registers as a candidate and blocks quiescence.
         frozen.delete(pid);
       }
     }
@@ -393,11 +406,11 @@ ${REMOTE_QUIESCENCE_PS_JS}
 ${REMOTE_QUIESCENCE_LEASE_JS}
 const input = parseLease(raw, nonce);
 if (input.watchdog !== null && processIdentity(input.watchdog.pid) === input.watchdog.start) {
-  try { process.kill(input.watchdog.pid, "SIGTERM"); } catch (error) { if (!error || error.code !== "ESRCH") throw error; }
+  try { process.kill(input.watchdog.pid, "SIGTERM"); } catch (error) { if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error; }
 }
 for (const entry of input.processes) {
   if (processIdentity(entry.pid) !== entry.start) continue;
-  try { process.kill(entry.pid, "SIGCONT"); } catch (error) { if (!error || error.code !== "ESRCH") throw error; }
+  try { process.kill(entry.pid, "SIGCONT"); } catch (error) { if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error; }
 }
 fs.unlinkSync(leasePath);
 `;