fix(qa): restore paired-node worker crash recovery proof (#129241)

qa/scenarios/runtime/cloud-worker-midturn-loss.yaml

@@ -7,14 +7,14 @@ scenario:
   coverage:
     secondary:
       - gateway.session-apis-sessions-list
-  objective: Prove a static-SSH worker can disappear during a streamed turn without losing or duplicating its already committed transcript prefix, silently hanging the turn, or breaking redispatch context; prove that real worker disk-space projections are evicted after reclaim.
+  objective: Prove a paired-node worker can disappear during a streamed turn without losing or duplicating its already committed transcript prefix, silently hanging the turn, or breaking redispatch context; prove that real worker disk-space projections are evicted after reclaim.
   successCriteria:
-    - A managed-worktree qa-channel session dispatches to a real static-SSH worker through an isolated Gateway.
+    - A managed-worktree qa-channel session dispatches to a real paired-node worker through an isolated Gateway.
     - The mock model persists two assistant/tool-result checkpoints, then pauses during a fifth streamed message.
-    - Killing the proof-owned SSH and worker process tree leaves exactly the four completed checkpoint messages in Gateway history.
+    - Killing the proof-owned worker process leaves exactly the four completed checkpoint messages in Gateway history before its paired node disconnects.
     - The volatile streamed message is absent from durable history while the turn emits a visible chat error and the placement records a bounded terminal reason.
-    - Restarting the static-SSH host and redispatching the same session produces one successful recovery turn whose inference context contains each checkpoint exactly once.
-    - The real static-SSH worker runs the disk-space probe and exposes its projection through `sessions.list`.
+    - Reconnecting the same paired node and redispatching the session produces one successful recovery turn whose inference context contains each checkpoint exactly once.
+    - The real paired-node worker runs the disk-space probe and exposes its projection through `sessions.list`.
     - Reclaiming the worker causes a later `sessions.list` read to return the reclaimed placement without a stale disk-space projection.
   docsRefs:
     - docs/gateway/cloud-workers.md
@@ -30,7 +30,7 @@ scenario:
   execution:
     kind: script
     path: test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-proof.ts
-    summary: Dispatches a managed-worktree qa-channel session to a proof-owned static-SSH worker, kills its process tree during a streamed message, and verifies the durable cutoff, visible failure, and redispatch context.
+    summary: Dispatches a managed-worktree qa-channel session to a proof-owned paired-node worker, kills its process and disconnects its node during a streamed message, and verifies the durable cutoff, visible failure, and redispatch context.
     timeoutMs: 900000
     args:
       - --artifact-base

test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-fixture.ts

@@ -1,12 +1,4 @@
-import { execFile, spawn, type ChildProcess } from "node:child_process";
-import fs from "node:fs/promises";
 import { createServer, type ServerResponse } from "node:http";
-import { createServer as createNetServer } from "node:net";
-import os from "node:os";
-import path from "node:path";
-import { promisify } from "node:util";
-
-const execFileAsync = promisify(execFile);
 
 export const MODEL_REF = "mock-openai/gpt-5.6-luna";
 export const BASELINE_PROMPT = "Reply exactly: CLOUD-MIDTURN-BASELINE";
@@ -28,26 +20,6 @@ export const COMMITTED_MARKERS = [
 ] as const;
 export const PROOF_TIMEOUT_MS = 180_000;
 
-function privilegedInvocation(command: string, args: readonly string[]) {
-  if (typeof process.getuid !== "function" || process.getuid() === 0) {
-    return { command, args: [...args] };
-  }
-  return { command: "/usr/bin/sudo", args: ["-n", "--", command, ...args] };
-}
-
-async function runChecked(command: string, args: readonly string[]) {
-  return await execFileAsync(command, [...args], {
-    encoding: "utf8",
-    maxBuffer: 1024 * 1024,
-    timeout: 10_000,
-  });
-}
-
-async function runPrivileged(command: string, args: readonly string[]) {
-  const invocation = privilegedInvocation(command, args);
-  return await runChecked(invocation.command, invocation.args);
-}
-
 function delay(ms: number): Promise<void> {
   return new Promise((resolve) => {
     setTimeout(resolve, ms);
@@ -441,222 +413,3 @@ export async function startMidturnProvider() {
     },
   };
 }
-
-async function reserveLoopbackPort(): Promise<number> {
-  const server = createNetServer();
-  await new Promise<void>((resolve, reject) => {
-    server.once("error", reject);
-    server.listen(0, "127.0.0.1", resolve);
-  });
-  const address = server.address();
-  if (!address || typeof address === "string") {
-    throw new Error("could not reserve SSH port");
-  }
-  await new Promise<void>((resolve, reject) => {
-    server.close((error) => {
-      if (error) {
-        reject(error);
-      } else {
-        resolve();
-      }
-    });
-  });
-  return address.port;
-}
-
-async function resolveSshdPath(): Promise<string> {
-  for (const candidate of ["/usr/sbin/sshd", "/usr/local/sbin/sshd", "/opt/homebrew/sbin/sshd"]) {
-    try {
-      await fs.access(candidate);
-      return candidate;
-    } catch {
-      // Try the next platform path.
-    }
-  }
-  throw new Error("sshd is required for the static-SSH mid-turn proof");
-}
-
-type SshdProcess = {
-  child: ChildProcess;
-  daemonPid: number;
-  exit: Promise<void>;
-  stderr: () => string;
-};
-
-export async function createSshdFixture(root: string) {
-  const sshdPath = await resolveSshdPath();
-  const port = await reserveLoopbackPort();
-  const hostKeyPath = path.join(root, "ssh-host-key");
-  const clientKeyPath = path.join(root, "ssh-client-key");
-  const authorizedKeysPath = path.join(root, "authorized_keys");
-  const knownHostsPath = path.join(root, "known_hosts");
-  const configPath = path.join(root, "sshd_config");
-  if (typeof process.getuid === "function" && process.getuid() !== 0) {
-    await runChecked("/usr/bin/sudo", ["-n", "true"]);
-  }
-  await execFileAsync("ssh-keygen", ["-q", "-t", "ed25519", "-N", "", "-f", hostKeyPath]);
-  await execFileAsync("ssh-keygen", ["-q", "-t", "ed25519", "-N", "", "-f", clientKeyPath]);
-  const hostKey = (await fs.readFile(`${hostKeyPath}.pub`, "utf8"))
-    .trim()
-    .split(/\s+/u)
-    .slice(0, 2)
-    .join(" ");
-  const clientPublicKey = await fs.readFile(`${clientKeyPath}.pub`, "utf8");
-  await fs.writeFile(authorizedKeysPath, clientPublicKey, { mode: 0o600 });
-  await fs.writeFile(knownHostsPath, `[127.0.0.1]:${port} ${hostKey}\n`, "utf8");
-  const user = os.userInfo().username;
-  await fs.writeFile(
-    configPath,
-    [
-      `Port ${port}`,
-      "ListenAddress 127.0.0.1",
-      `HostKey ${hostKeyPath}`,
-      `PidFile ${path.join(root, "sshd.pid")}`,
-      `AuthorizedKeysFile ${authorizedKeysPath}`,
-      "StrictModes no",
-      "AuthenticationMethods publickey",
-      "PubkeyAuthentication yes",
-      "PasswordAuthentication no",
-      "KbdInteractiveAuthentication no",
-      "ChallengeResponseAuthentication no",
-      "PermitEmptyPasswords no",
-      // Testbox runner accounts are password-locked; PAM still permits generated-key auth.
-      "UsePAM yes",
-      "PermitRootLogin prohibit-password",
-      `AllowUsers ${user}`,
-      "AllowTcpForwarding yes",
-      "AllowStreamLocalForwarding yes",
-      "StreamLocalBindUnlink yes",
-      "PrintMotd no",
-      "LogLevel VERBOSE",
-      "Subsystem sftp internal-sftp",
-      "",
-    ].join("\n"),
-    "utf8",
-  );
-  await runPrivileged(sshdPath, ["-t", "-f", configPath]);
-
-  const start = async (): Promise<SshdProcess> => {
-    const invocation = privilegedInvocation(sshdPath, ["-D", "-e", "-f", configPath]);
-    const child = spawn(invocation.command, invocation.args, {
-      stdio: ["ignore", "ignore", "pipe"],
-    });
-    let stderrText = "";
-    child.stderr?.on("data", (chunk: Buffer) => {
-      stderrText = `${stderrText}${chunk.toString("utf8")}`.slice(-8_000);
-    });
-    const exit = new Promise<void>((resolve) => {
-      child.once("exit", () => resolve());
-    });
-    await waitFor("proof SSH server", async () => {
-      if (child.exitCode !== null || child.signalCode !== null) {
-        throw new Error(`proof sshd exited early: ${stderrText}`);
-      }
-      try {
-        await execFileAsync(
-          "ssh",
-          [
-            "-F",
-            "/dev/null",
-            "-i",
-            clientKeyPath,
-            "-p",
-            String(port),
-            "-o",
-            "BatchMode=yes",
-            "-o",
-            "IdentitiesOnly=yes",
-            "-o",
-            "StrictHostKeyChecking=yes",
-            "-o",
-            `UserKnownHostsFile=${knownHostsPath}`,
-            `${user}@127.0.0.1`,
-            "true",
-          ],
-          { timeout: 2_000 },
-        );
-        return true;
-      } catch {
-        return undefined;
-      }
-    });
-    const daemonPidText = (await fs.readFile(path.join(root, "sshd.pid"), "utf8")).trim();
-    if (!/^[1-9]\d*$/u.test(daemonPidText)) {
-      throw new Error(`proof sshd did not write a valid pid: ${daemonPidText}`);
-    }
-    return { child, daemonPid: Number(daemonPidText), exit, stderr: () => stderrText };
-  };
-  return { clientKeyPath, hostKey, port, start, user };
-}
-
-async function processTree(rootPid: number) {
-  const { stdout } = await execFileAsync("ps", ["-axww", "-o", "pid=,ppid=,command="], {
-    encoding: "utf8",
-  });
-  const rows = stdout.split("\n").flatMap((line) => {
-    const match = /^\s*(\d+)\s+(\d+)\s+(.*)$/u.exec(line);
-    return match
-      ? [{ pid: Number(match[1]), ppid: Number(match[2]), command: match[3] ?? "" }]
-      : [];
-  });
-  const descendants = [] as typeof rows;
-  const parents = new Set([rootPid]);
-  while (true) {
-    const found = rows.filter((row) => parents.has(row.ppid) && !parents.has(row.pid));
-    if (found.length === 0) {
-      break;
-    }
-    for (const row of found) {
-      parents.add(row.pid);
-      descendants.push(row);
-    }
-  }
-  return descendants;
-}
-
-export async function killSshdProcessTree(process: SshdProcess) {
-  const pid = process.daemonPid;
-  const descendants = await processTree(pid);
-  const worker = descendants.find((entry) =>
-    /(?:^|\/)(?:openclaw-worker|openclaw\.mjs\s+worker)\b/u.test(entry.command),
-  );
-  if (!worker) {
-    throw new Error(`proof sshd tree had no worker process: ${JSON.stringify(descendants)}`);
-  }
-  for (const entry of descendants.toReversed()) {
-    await runPrivileged("/bin/kill", ["-KILL", String(entry.pid)]).catch(() => undefined);
-  }
-  await runPrivileged("/bin/kill", ["-KILL", String(pid)]).catch(() => undefined);
-  await Promise.race([process.exit, delay(5_000)]);
-  if (process.child.exitCode === null && process.child.signalCode === null) {
-    process.child.kill("SIGKILL");
-    await process.exit;
-  }
-  return { killedProcessCount: descendants.length + 1, workerPid: worker.pid };
-}
-
-export async function stopSshd(process: SshdProcess | undefined): Promise<void> {
-  if (!process) {
-    return;
-  }
-  await runPrivileged("/bin/kill", ["-KILL", String(process.daemonPid)]).catch(() => undefined);
-  await Promise.race([process.exit, delay(5_000)]);
-  if (process.child.exitCode === null && process.child.signalCode === null) {
-    process.child.kill("SIGKILL");
-    await process.exit;
-  }
-}
-
-export async function initializeRepository(root: string): Promise<string> {
-  const repo = path.join(root, "workspace-source");
-  await fs.mkdir(repo, { recursive: true });
-  const git = (...args: string[]) => execFileAsync("git", ["-C", repo, ...args]);
-  await git("init", "-b", "main");
-  await git("config", "user.name", "OpenClaw QA");
-  await git("config", "user.email", "openclaw-qa@example.invalid");
-  await fs.writeFile(path.join(repo, "checkpoint-1.txt"), "CLOUD-MIDTURN-TOOL-1\n");
-  await fs.writeFile(path.join(repo, "checkpoint-2.txt"), "CLOUD-MIDTURN-TOOL-2\n");
-  await git("add", ".");
-  await git("commit", "-m", "initialize cloud mid-turn proof workspace");
-  return await fs.realpath(repo);
-}

test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-proof.ts

@@ -18,30 +18,33 @@ import {
   GATEWAY_CLIENT_NAMES,
 } from "../../../../packages/gateway-protocol/src/client-info.js";
 import { loadOrCreateDeviceIdentity } from "../../../../src/infra/device-identity.js";
+import { NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND } from "../../../../src/infra/node-commands.js";
 import {
   BASELINE_PROMPT,
   BASELINE_REPLY,
   COMMITTED_MARKERS,
   CONTEXT_PROMPT,
   CONTEXT_REPLY,
-  createSshdFixture,
-  initializeRepository,
-  killSshdProcessTree,
   MIDTURN_PROMPT,
   MODEL_REF,
   PROOF_TIMEOUT_MS,
   startMidturnProvider,
-  stopSshd,
   VOLATILE_TEXT,
   waitFor,
 } from "./cloud-worker-midturn-loss-fixture.js";
+import {
+  closeWireServer,
+  createPairedNodeWorkerHost,
+  createPublishedWireWorkspace,
+  type PairedNodeWorkerHost,
+  type PublishedWireWorkspace,
+} from "./paired-node-worker-wire-fixture.js";
 import { createQaScriptEvidenceWriter } from "./script-evidence.js";
 
 const SCENARIO_ID = "cloud-worker-midturn-loss";
 const VERDICT_FILE = `${SCENARIO_ID}-verdict.json`;
 const SESSION_KEY = "agent:qa:qa-channel:direct:cloud-midturn-loss";
 const SENDER_ID = "cloud-midturn-loss";
-const PROFILE_ID = "development";
 
 type ProducerOptions = { artifactBase: string; repoRoot: string };
 type Gateway = Awaited<ReturnType<typeof startQaGatewayChild>>;
@@ -105,7 +108,7 @@ async function connectOperator(
       clientVersion: "1.0.0",
       platform: process.platform,
       mode: GATEWAY_CLIENT_MODES.WEBCHAT,
-      scopes: ["operator.admin", "operator.read", "operator.write"],
+      scopes: ["operator.admin", "operator.pairing", "operator.read", "operator.write"],
       deviceIdentity,
       requestTimeoutMs: PROOF_TIMEOUT_MS,
       onEvent: (event) => events.push(event),
@@ -274,18 +277,17 @@ async function runProof(options: ProducerOptions) {
   const fixtureRoot = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-cloud-midturn-loss-"));
   let bus: Awaited<ReturnType<typeof startQaBusServer>> | undefined;
   let provider: Awaited<ReturnType<typeof startMidturnProvider>> | undefined;
-  let sshd: Parameters<typeof stopSshd>[0] = undefined;
   let gateway: Gateway | undefined;
   let operator: GatewayClient | undefined;
+  let workerNode: PairedNodeWorkerHost | undefined;
+  let published: PublishedWireWorkspace | undefined;
+  let workerLaunchId: string | undefined;
   let proofError: unknown;
   let verdict: Record<string, unknown> | undefined;
   try {
     bus = await startQaBusServer({ state });
     provider = await startMidturnProvider();
-    const ssh = await createSshdFixture(fixtureRoot);
-    sshd = await ssh.start();
-    const repo = await initializeRepository(fixtureRoot);
-    const sshPrivateKey = await fs.readFile(ssh.clientKeyPath, "utf8");
+    published = await createPublishedWireWorkspace(fixtureRoot);
     const transport = createQaChannelTransport(state);
     gateway = await startQaGatewayChild({
       repoRoot: options.repoRoot,
@@ -299,32 +301,12 @@ async function runProof(options: ProducerOptions) {
       enabledPluginIds: ["qa-lab"],
       controlUiEnabled: false,
       controlUiAllowedOrigins: ["http://127.0.0.1"],
-      runtimeEnvPatch: { OPENCLAW_QA_STATIC_SSH_KEY: sshPrivateKey },
       mutateConfig: (config) => ({
         ...config,
         session: { ...config.session, dmScope: "per-peer" },
-        secrets: {
-          ...config.secrets,
-          providers: { ...config.secrets?.providers, default: { source: "env" } },
-        },
-        cloudWorkers: {
-          profiles: {
-            [PROFILE_ID]: {
-              provider: "static-ssh",
-              install: "bundle",
-              settings: {
-                host: "127.0.0.1",
-                port: ssh.port,
-                user: ssh.user,
-                hostKey: ssh.hostKey,
-                keyRef: {
-                  source: "env",
-                  provider: "default",
-                  id: "OPENCLAW_QA_STATIC_SSH_KEY",
-                },
-              },
-            },
-          },
+        nodeHost: {
+          ...config.nodeHost,
+          workerRuns: { enabled: true },
         },
       }),
     });
@@ -333,14 +315,38 @@ async function runProof(options: ProducerOptions) {
       path: path.join(fixtureRoot, "operator-identity.sqlite"),
     });
     operator = await connectOperator(gateway, events, deviceIdentity);
+    workerNode = await createPairedNodeWorkerHost({
+      gateway,
+      operator,
+      root: fixtureRoot,
+      label: "midturn-worker",
+      onInvoke: (frame) => {
+        if (frame.command === NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND && frame.paramsJSON) {
+          workerLaunchId = (JSON.parse(frame.paramsJSON) as { launchId?: string }).launchId;
+        }
+      },
+    });
     await operator.request("sessions.create", {
       key: SESSION_KEY,
       agentId: "qa",
       worktree: true,
       worktreeName: `cloud-midturn-${randomUUID().slice(0, 8)}`,
       worktreeBaseRef: "main",
-      cwd: repo,
+      cwd: published.source,
     });
+    const created = requireRecord(
+      await gateway.call("sessions.describe", { key: SESSION_KEY }),
+      "created session",
+    );
+    const session = requireRecord(created.session, "created session details");
+    const localWorkspaceDir = session.execCwd ?? session.spawnedCwd;
+    if (typeof localWorkspaceDir !== "string") {
+      throw new Error("created session did not expose its managed workspace");
+    }
+    await Promise.all([
+      fs.writeFile(path.join(localWorkspaceDir, "checkpoint-1.txt"), "CLOUD-MIDTURN-TOOL-1\n"),
+      fs.writeFile(path.join(localWorkspaceDir, "checkpoint-2.txt"), "CLOUD-MIDTURN-TOOL-2\n"),
+    ]);
     await operator.request("sessions.messages.subscribe", { key: SESSION_KEY });
 
     const baselineCursor = state.getSnapshot().messages.length;
@@ -354,7 +360,7 @@ async function runProof(options: ProducerOptions) {
 
     await gateway.call(
       "sessions.dispatch",
-      { key: SESSION_KEY, profileId: PROFILE_ID },
+      { key: SESSION_KEY, deviceId: workerNode.identity.deviceId },
       { timeoutMs: PROOF_TIMEOUT_MS },
     );
     const runId = `cloud-midturn-loss-${randomUUID()}`;
@@ -375,7 +381,22 @@ async function runProof(options: ProducerOptions) {
     });
     await waitForVolatilePreview(events, runId);
 
-    const killed = await killSshdProcessTree(sshd);
+    const node = workerNode;
+    const activeWorker = await waitFor("proof-owned active node worker", async () => {
+      if (!workerLaunchId) {
+        return undefined;
+      }
+      const receipt = await node.supervisor.status(workerLaunchId);
+      return receipt?.state === "running" && receipt.runId === runId && receipt.worker
+        ? receipt.worker
+        : undefined;
+    });
+    process.kill(activeWorker.pid, "SIGKILL");
+    const killed = {
+      killedProcessCount: 1,
+      nodeDeviceId: node.identity.deviceId,
+      workerPid: activeWorker.pid,
+    };
     const waitResult = await operator.request<GatewayRunResult>(
       "agent.wait",
       { runId, timeoutMs: PROOF_TIMEOUT_MS },
@@ -404,11 +425,13 @@ async function runProof(options: ProducerOptions) {
       throw new Error(`unexpected durable cutoff: ${JSON.stringify(committedSequence)}`);
     }
 
-    sshd = await ssh.start();
+    // Keep the node connected until its supervisor delivers the worker's terminal receipt.
+    await node.disconnect();
+    await node.connect();
     const redispatched = requireRecord(
       await gateway.call(
         "sessions.dispatch",
-        { key: SESSION_KEY, profileId: PROFILE_ID },
+        { key: SESSION_KEY, deviceId: workerNode.identity.deviceId },
         { timeoutMs: PROOF_TIMEOUT_MS },
       ),
       "sessions.dispatch redispatch",
@@ -454,7 +477,7 @@ async function runProof(options: ProducerOptions) {
       throw new Error("operator was unavailable after recovery");
     }
     const activeDiskSpace = await waitFor(
-      "real static-SSH worker disk-space projection",
+      "real paired-node worker disk-space projection",
       async () =>
         readActiveWorkerDiskSpace(await qaOperator.request<SessionsList>("sessions.list", {})),
     );
@@ -474,7 +497,7 @@ async function runProof(options: ProducerOptions) {
       status: "pass",
       providerMode: "mock-openai",
       channel: "qa-channel",
-      workerProvider: "static-ssh",
+      workerProvider: "device",
       sessionKey: SESSION_KEY,
       killedWorker: killed,
       durableTranscript: {
@@ -524,8 +547,9 @@ async function runProof(options: ProducerOptions) {
   } finally {
     const cleanup = await Promise.allSettled([
       operator?.stopAndWait({ timeoutMs: 1_000 }) ?? Promise.resolve(),
+      workerNode?.stop() ?? Promise.resolve(),
       gateway?.stop() ?? Promise.resolve(),
-      stopSshd(sshd),
+      published ? closeWireServer(published.server) : Promise.resolve(),
       provider?.stop() ?? Promise.resolve(),
       bus?.stop() ?? Promise.resolve(),
       fs.rm(fixtureRoot, { recursive: true, force: true }),
@@ -578,7 +602,7 @@ async function runProducer(options: ProducerOptions): Promise<QaEvidenceSummaryJ
     return await writer.write({
       artifacts: [{ filePath: VERDICT_FILE, kind: "verdict" }],
       details:
-        "static-SSH process-tree loss preserved the exact committed transcript prefix, surfaced an error, and redispatched with continuous context",
+        "paired-node worker loss preserved the exact committed transcript prefix, surfaced an error, and redispatched with continuous context",
       durationMs: Math.max(1, Date.now() - startedAt),
       status: "pass",
     });