From 1ba243c88ed800986909bc50e4ce7b8139891b94 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 25 Aug 2026 03:54:50 -0700 Subject: [PATCH] fix(qa): restore paired-node worker crash recovery proof (#129241) --- .../runtime/cloud-worker-midturn-loss.yaml | 12 +- .../cloud-worker-midturn-loss-fixture.ts | 247 ------------------ .../cloud-worker-midturn-loss-proof.ts | 110 +++++--- 3 files changed, 73 insertions(+), 296 deletions(-) diff --git a/qa/scenarios/runtime/cloud-worker-midturn-loss.yaml b/qa/scenarios/runtime/cloud-worker-midturn-loss.yaml index 7215abb055b9..b3840de0db29 100644 --- a/qa/scenarios/runtime/cloud-worker-midturn-loss.yaml +++ b/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 diff --git a/test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-fixture.ts b/test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-fixture.ts index 933849b36a5c..e5f2d24fe751 100644 --- a/test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-fixture.ts +++ b/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 { return new Promise((resolve) => { setTimeout(resolve, ms); @@ -441,222 +413,3 @@ export async function startMidturnProvider() { }, }; } - -async function reserveLoopbackPort(): Promise { - const server = createNetServer(); - await new Promise((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((resolve, reject) => { - server.close((error) => { - if (error) { - reject(error); - } else { - resolve(); - } - }); - }); - return address.port; -} - -async function resolveSshdPath(): Promise { - 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; - 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 => { - 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((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 { - 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 { - 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); -} diff --git a/test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-proof.ts b/test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-proof.ts index 4f4d5f012710..e0b2ef1ef389 100644 --- a/test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-proof.ts +++ b/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>; @@ -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> | undefined; let provider: Awaited> | undefined; - let sshd: Parameters[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 | 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( "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("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