diff --git a/docs/gateway/cloud-workers.md b/docs/gateway/cloud-workers.md index 18cbbf05c76b..0248983c7ea7 100644 --- a/docs/gateway/cloud-workers.md +++ b/docs/gateway/cloud-workers.md @@ -9,7 +9,7 @@ doc-schema-version: 1 Cloud workers let a session run its agent loop on a throwaway cloud machine while everything about the session stays where it always was: visible in the sidebar, streaming live, with the transcript owned by the Gateway. The Gateway leases a box, installs a pinned copy of OpenClaw on it, syncs the session's workspace over, and hands the turn loop to a restricted `openclaw worker` process. Model calls are proxied back through the Gateway, so provider credentials never leave your machine, and prompt caching keeps working because the provider sees one continuous stream. -When the work is done (or the box dies), the machine is discarded. The durable state — transcript, workspace commits, placement records — lives with the Gateway. +When the work is done (or the box dies), the machine is discarded. The durable state — transcript, last-reconciled workspace files, and placement records — lives with the Gateway. Cloud workers are opt-in. Until you configure a profile, clients hide the Cloud destination and the Gateway does not advertise `sessions.dispatch`. The `cloudWorkers` config schema and the read-only `environments.list` and `environments.status` methods remain available for configuration and environment discovery. @@ -17,13 +17,13 @@ Cloud workers are opt-in. Until you configure a profile, clients hide the Cloud ## What runs where -| Concern | Location | -| ------------------------------------------------------- | -------------------------------------------------------------------------------- | -| Agent loop + tools (`exec`, `read`, `write`, `edit`, …) | Cloud worker box | -| Model inference and provider credentials | Gateway (proxied by `{provider, model}` reference) | -| Transcript (durable, session store) | Gateway | -| Live streaming into the sidebar | Gateway fanout, fed by the worker's replayable event stream | -| Workspace git history | Authored on the box credential-free; the Gateway adopts commits and owns push/PR | +| Concern | Location | +| ------------------------------------------------------- | --------------------------------------------------------------------------------- | +| Agent loop + tools (`exec`, `read`, `write`, `edit`, …) | Cloud worker box | +| Model inference and provider credentials | Gateway (proxied by `{provider, model}` reference) | +| Transcript (durable, session store) | Gateway | +| Live streaming into the sidebar | Gateway fanout, fed by the worker's replayable event stream | +| Workspace file state | Changed on the box credential-free; the Gateway reconciles files and owns push/PR | The box needs no inbound ports except `sshd`: the Gateway connects out via pinned SSH, and a reverse tunnel carries the worker's WebSocket back. The bundled Crabbox provider forces the public SSH route and disables managed Tailscale enrollment. Outbound internet access is provider policy; the default AWS profile can reach the internet unless you restrict its network or security group. @@ -188,6 +188,14 @@ openclaw gateway call sessions.reclaim \ Placement moves through a durable state machine (`local → requested → provisioning → syncing → starting → active`), so a Gateway restart mid-dispatch reconciles instead of leaking machines. A failed model turn keeps the active placement available for a retry. Workspace path conflicts keep the local version, apply the rest of the cloud result, and preserve the staged cloud ref for inspection; other reconciliation or lifecycle failures retain their durable recovery fence and diagnostic tail until recovery can safely retry or reclaim the environment. +## What survives a dead machine + +The Gateway commits each complete user, assistant, and tool-result message to the canonical session transcript before the worker's session write settles. Commits are ordered and idempotent against the exact transcript leaf. If the machine disappears mid-message, durable history ends at the last committed message. Partial text or tool progress already shown by the live stream may disappear; the failed turn remains visible, and the failed placement records a bounded terminal reason above the composer. + +Workspace state has a wider loss window. A completed turn reconciles worker files before releasing its claim, and **Stop cloud worker…** performs one final reconciliation before destroying the machine. Changes made between reconciliations exist only on the worker and can be lost. Session deletion does not synchronize a live worker: active placements must first be stopped or archived. Deletion then snapshots the already-reconciled managed worktree under `refs/openclaw/snapshots/` before removing it. + +After a failed placement, redispatch the session and retry the turn. A reclaimed placement redispatches automatically on the next turn. The new worker rebuilds its inference context from the Gateway transcript, so it continues from the messages that crossed the durability boundary. + ## Desktop (interactive) Cloud Worker Desktop is an experimental Labs feature and is off by default. Enable **Cloud Worker Desktop** in **Settings → Agents & Tools → Labs**, or set `cloudWorkers.desktop: true`, then restart the Gateway for the Desktop panel to appear. diff --git a/qa/scenarios/runtime/cloud-worker-midturn-loss.yaml b/qa/scenarios/runtime/cloud-worker-midturn-loss.yaml new file mode 100644 index 000000000000..c304adec314d --- /dev/null +++ b/qa/scenarios/runtime/cloud-worker-midturn-loss.yaml @@ -0,0 +1,33 @@ +title: Cloud worker mid-turn machine loss + +scenario: + id: cloud-worker-midturn-loss + surface: gateway + category: gateway.session-apis + 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. + successCriteria: + - A managed-worktree qa-channel session dispatches to a real static-SSH 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. + - 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. + docsRefs: + - docs/gateway/cloud-workers.md + - docs/concepts/qa-e2e-automation.md + - docs/channels/qa-channel.md + codeRefs: + - src/worker/embedded-agent-transcript.runtime.ts + - src/gateway/worker-environments/transcript-commit.ts + - src/gateway/worker-environments/worker-turn-launcher.ts + - test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-proof.ts + 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. + timeoutMs: 900000 + args: + - --artifact-base + - ${outputDir} diff --git a/src/gateway/worker-environments/bundle-staging.ts b/src/gateway/worker-environments/bundle-staging.ts index 115156f19c7b..8f461d0a3e01 100644 --- a/src/gateway/worker-environments/bundle-staging.ts +++ b/src/gateway/worker-environments/bundle-staging.ts @@ -49,7 +49,7 @@ function serializePackageManifest(parsed: Record): Buffer { // at their vendored copies so `npm install` on the box resolves them without a registry. function pruneWorkerPackageManifest( contents: Buffer, - vendoredDirsByName: ReadonlyMap, + vendoredDirsByName: ReadonlyMap = new Map(), ): Buffer { const parsed = JSON.parse(contents.toString("utf8")) as Record; const dependencies = readManifestDependencies(parsed); @@ -75,14 +75,6 @@ function pruneWorkerPackageManifest( return serializePackageManifest(pruned); } -// Vendored workspace manifests keep their registry dependencies but never ship -// lifecycle scripts or dev-only fields. -function pruneVendoredPackageManifest(contents: Buffer): Buffer { - const parsed = JSON.parse(contents.toString("utf8")) as Record; - const { pruned, prunedFieldCount } = withoutLifecycleFields(parsed); - return prunedFieldCount === 0 ? contents : serializePackageManifest(pruned); -} - function normalizePortableMode(mode: number, relativePath: string): number { return relativePath === "openclaw.mjs" || (mode & 0o111) !== 0 ? 0o700 : 0o600; } @@ -184,6 +176,22 @@ function collectOpenclawImportSpecifiers( } } +function pruneVendoredPackageManifest( + packageName: string, + referencedPackages: ReadonlySet, + contents: Buffer, +): Buffer { + const parsed = JSON.parse(contents.toString("utf8")) as Record; + for (const [dependencyName, spec] of Object.entries(readManifestDependencies(parsed))) { + if (spec.startsWith("workspace:") && referencedPackages.has(dependencyName)) { + throw new Error( + `Vendored workspace dependency ${dependencyName} remains referenced by ${packageName} dist; bundle it into the package build or add explicit worker bundle support`, + ); + } + } + return pruneWorkerPackageManifest(contents); +} + async function readWorkspaceDependencyNames(sourceRoot: string): Promise> { const raw = await fs.readFile(path.join(sourceRoot, "package.json"), "utf8"); const dependencies = readManifestDependencies(JSON.parse(raw) as Record); @@ -248,15 +256,25 @@ async function stageVendoredWorkspacePackages(params: { ); } const vendorDir = `vendor/${packageName.replace(/^@/u, "").replaceAll("/", "-")}`; - for (const relativePath of await collectVendoredPackageFiles(packageName, vendorRealRoot)) { - const { entry } = await stageFileEntry(params.stagingRoot, { + const files = await collectVendoredPackageFiles(packageName, vendorRealRoot); + const referencedPackages = new Set(); + for (const relativePath of files.filter((candidate) => candidate !== "package.json")) { + const { entry, contents } = await stageFileEntry(params.stagingRoot, { sourcePath: path.join(vendorRealRoot, ...relativePath.split("/")), expectedRealPath: path.resolve(vendorRealRoot, ...relativePath.split("/")), stagedPath: `${vendorDir}/${relativePath}`, - transform: relativePath === "package.json" ? pruneVendoredPackageManifest : undefined, }); + collectOpenclawImportSpecifiers(relativePath, contents, referencedPackages); entries.push(entry); } + const { entry: packageManifestEntry } = await stageFileEntry(params.stagingRoot, { + sourcePath: path.join(vendorRealRoot, "package.json"), + expectedRealPath: path.resolve(vendorRealRoot, "package.json"), + stagedPath: `${vendorDir}/package.json`, + transform: (contents) => + pruneVendoredPackageManifest(packageName, referencedPackages, contents), + }); + entries.push(packageManifestEntry); vendoredDirsByName.set(packageName, vendorDir); } return { entries, vendoredDirsByName }; diff --git a/src/gateway/worker-environments/bundle.test.ts b/src/gateway/worker-environments/bundle.test.ts index 3b58bf1feab0..34a48fe75ce5 100644 --- a/src/gateway/worker-environments/bundle.test.ts +++ b/src/gateway/worker-environments/bundle.test.ts @@ -184,7 +184,10 @@ describe("worker bundle producer", () => { version: "1.2.3", type: "module", main: "./dist/index.js", - dependencies: { "partial-json": "0.1.7" }, + dependencies: { + "@openclaw/fake-nested": "workspace:*", + "partial-json": "0.1.7", + }, scripts: { build: "tsdown" }, devDependencies: { vitest: "4.0.0" }, })}\n`, @@ -225,6 +228,21 @@ describe("worker bundle producer", () => { expect(vendored.dependencies).toEqual({ "partial-json": "0.1.7" }); expect(vendored).not.toHaveProperty("scripts"); expect(vendored).not.toHaveProperty("devDependencies"); + + await fs.writeFile( + path.join(vendorSource, "dist/index.js"), + 'import { nested } from "@openclaw/fake-nested";\nexport const fake = nested;\n', + "utf8", + ); + await expect( + createWorkerBundleProducer({ + packageRoot, + cacheDir: path.join(root, "cache-with-runtime-workspace-dep"), + openclawVersion: "1.2.3", + }).prepare(), + ).rejects.toThrow( + "Vendored workspace dependency @openclaw/fake-nested remains referenced by @openclaw/fake-pkg dist", + ); }); }); 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 new file mode 100644 index 000000000000..90061aff84b6 --- /dev/null +++ b/test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-fixture.ts @@ -0,0 +1,538 @@ +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"; +export const BASELINE_REPLY = "CLOUD-MIDTURN-BASELINE"; +export const MIDTURN_PROMPT = + "CLOUD-MIDTURN-KILL: persist two checkpoints, then stream the final reply."; +export const CONTEXT_PROMPT = + "CLOUD-MIDTURN-CONTEXT: prove the committed checkpoints are in context."; +export const CONTEXT_REPLY = "CLOUD-MIDTURN-CONTEXT-OK"; +export const VOLATILE_TEXT = "CLOUD-MIDTURN-VOLATILE-PARTIAL"; +export const COMMITTED_MARKERS = [ + "CLOUD-MIDTURN-ASSISTANT-1", + "CLOUD-MIDTURN-TOOL-1", + "CLOUD-MIDTURN-ASSISTANT-2", + "CLOUD-MIDTURN-TOOL-2", +] 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); + }); +} + +export async function waitFor( + label: string, + read: () => T | undefined | Promise, +) { + const deadline = Date.now() + PROOF_TIMEOUT_MS; + while (Date.now() < deadline) { + const value = await read(); + if (value !== undefined) { + return value; + } + await delay(50); + } + throw new Error(`timed out waiting for ${label}`); +} + +function createDeferred() { + let resolve = () => {}; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} + +function writeSseEvent(response: ServerResponse, event: unknown): void { + response.write(`data: ${JSON.stringify(event)}\n\n`); +} + +function assistantItem(id: string, text: string, phase: "commentary" | "final_answer") { + return { + type: "message", + id, + role: "assistant", + phase, + status: "completed", + content: [{ type: "output_text", text, annotations: [] }], + }; +} + +function toolCallItem(index: number, file: string) { + const args = JSON.stringify({ path: file }); + return { + args, + item: { + type: "function_call", + id: `fc_cloud_midturn_${index}`, + call_id: `call_cloud_midturn_${index}`, + name: "read", + arguments: args, + }, + }; +} + +function writeCompletedAssistant(response: ServerResponse, text: string, id: string): void { + const item = assistantItem(id, text, "final_answer"); + response.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-store", + connection: "keep-alive", + }); + writeSseEvent(response, { + type: "response.output_item.added", + output_index: 0, + item: { ...item, status: "in_progress", content: [] }, + }); + writeSseEvent(response, { + type: "response.output_text.delta", + item_id: id, + output_index: 0, + content_index: 0, + delta: text, + }); + writeSseEvent(response, { + type: "response.output_text.done", + item_id: id, + output_index: 0, + content_index: 0, + text, + }); + writeSseEvent(response, { type: "response.output_item.done", output_index: 0, item }); + writeSseEvent(response, { + type: "response.completed", + response: { + id: `resp_${id}`, + status: "completed", + output: [item], + usage: { input_tokens: 32, output_tokens: 8, total_tokens: 40 }, + }, + }); + response.end("data: [DONE]\n\n"); +} + +function writeCheckpointToolCall(response: ServerResponse, index: 1 | 2): void { + const text = `CLOUD-MIDTURN-ASSISTANT-${index}`; + const message = assistantItem(`msg_cloud_midturn_${index}`, text, "commentary"); + const call = toolCallItem(index, `checkpoint-${index}.txt`); + response.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-store", + connection: "keep-alive", + }); + writeSseEvent(response, { + type: "response.output_item.added", + output_index: 0, + item: { ...message, status: "in_progress", content: [] }, + }); + writeSseEvent(response, { + type: "response.output_text.delta", + item_id: message.id, + output_index: 0, + content_index: 0, + delta: text, + }); + writeSseEvent(response, { + type: "response.output_text.done", + item_id: message.id, + output_index: 0, + content_index: 0, + text, + }); + writeSseEvent(response, { type: "response.output_item.done", output_index: 0, item: message }); + writeSseEvent(response, { + type: "response.output_item.added", + output_index: 1, + item: { ...call.item, arguments: "" }, + }); + writeSseEvent(response, { + type: "response.function_call_arguments.delta", + item_id: call.item.id, + output_index: 1, + delta: call.args, + }); + writeSseEvent(response, { type: "response.output_item.done", output_index: 1, item: call.item }); + writeSseEvent(response, { + type: "response.completed", + response: { + id: `resp_cloud_midturn_${index}`, + status: "completed", + output: [message, call.item], + usage: { input_tokens: 64, output_tokens: 24, total_tokens: 88 }, + }, + }); + response.end("data: [DONE]\n\n"); +} + +async function readRequestBody(request: AsyncIterable): Promise { + const chunks: Buffer[] = []; + for await (const chunk of request) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk))); + } + return Buffer.concat(chunks).toString("utf8"); +} + +export async function startMidturnProvider() { + let midturnRequestCount = 0; + let contextRequest = ""; + const partialStarted = createDeferred(); + const releasePartial = createDeferred(); + const requests: string[] = []; + const server = createServer((request, response) => { + void (async () => { + if (request.method === "GET" && request.url === "/v1/models") { + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ data: [{ id: "gpt-5.6-luna", object: "model" }] })); + return; + } + if (request.method !== "POST" || request.url !== "/v1/responses") { + response.writeHead(404).end(); + return; + } + const raw = await readRequestBody(request); + requests.push(raw); + if (raw.includes(CONTEXT_PROMPT)) { + contextRequest = raw; + const missing = COMMITTED_MARKERS.filter((marker) => !raw.includes(marker)); + writeCompletedAssistant( + response, + missing.length === 0 ? CONTEXT_REPLY : `MISSING-CONTEXT:${missing.join(",")}`, + "msg_cloud_midturn_context", + ); + return; + } + if (raw.includes(MIDTURN_PROMPT)) { + midturnRequestCount += 1; + if (midturnRequestCount <= 2) { + writeCheckpointToolCall(response, midturnRequestCount as 1 | 2); + return; + } + response.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-store", + connection: "keep-alive", + }); + const item = assistantItem("msg_cloud_midturn_volatile", VOLATILE_TEXT, "final_answer"); + writeSseEvent(response, { + type: "response.output_item.added", + output_index: 0, + item: { ...item, status: "in_progress", content: [] }, + }); + for (const deltaText of ["CLOUD-MIDTURN-", "VOLATILE-", "PARTIAL"]) { + writeSseEvent(response, { + type: "response.output_text.delta", + item_id: item.id, + output_index: 0, + content_index: 0, + delta: deltaText, + }); + await delay(50); + } + partialStarted.resolve(); + await releasePartial.promise; + if (!response.destroyed) { + writeSseEvent(response, { + type: "response.output_text.done", + item_id: item.id, + output_index: 0, + content_index: 0, + text: VOLATILE_TEXT, + }); + writeSseEvent(response, { type: "response.output_item.done", output_index: 0, item }); + writeSseEvent(response, { + type: "response.completed", + response: { + id: "resp_cloud_midturn_volatile", + status: "completed", + output: [item], + usage: { input_tokens: 64, output_tokens: 12, total_tokens: 76 }, + }, + }); + response.end("data: [DONE]\n\n"); + } + return; + } + writeCompletedAssistant(response, BASELINE_REPLY, "msg_cloud_midturn_baseline"); + })().catch((error: unknown) => { + if (!response.headersSent) { + response.writeHead(500); + } + response.end(error instanceof Error ? error.message : String(error)); + }); + }); + 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("mid-turn provider did not bind"); + } + return { + baseUrl: `http://127.0.0.1:${address.port}`, + partialStarted: partialStarted.promise, + get contextRequest() { + return contextRequest; + }, + get requestCount() { + return requests.length; + }, + async stop() { + releasePartial.resolve(); + server.closeAllConnections(); + await new Promise((resolve) => { + server.close(() => resolve()); + }); + }, + }; +} + +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 new file mode 100644 index 000000000000..c3976f60d04c --- /dev/null +++ b/test/e2e/qa-lab/runtime/cloud-worker-midturn-loss-proof.ts @@ -0,0 +1,524 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; +import { isDeepStrictEqual } from "node:util"; +import { GatewayClient } from "openclaw/plugin-sdk/gateway-runtime"; +import { + createQaBusState, + createQaChannelTransport, + QA_EVIDENCE_FILENAME, + startQaBusServer, + startQaGatewayChild, + type QaEvidenceSummaryJson, +} from "../../../../extensions/qa-lab/api.js"; +import { + GATEWAY_CLIENT_MODES, + GATEWAY_CLIENT_NAMES, +} from "../../../../packages/gateway-protocol/src/client-info.js"; +import { loadOrCreateDeviceIdentity } from "../../../../src/infra/device-identity.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 { 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>; +type GatewayEvent = { event: string; payload?: unknown }; +type GatewayRunResult = { runId?: string; status?: string; summary?: string }; +type ChatHistory = { messages?: unknown[] }; + +function requireRecord(value: unknown, label: string): Record { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new Error(`${label} was not an object`); + } + return value as Record; +} + +function parseOptions(argv: readonly string[]): ProducerOptions { + const index = argv.indexOf("--artifact-base"); + const artifactBase = index >= 0 ? argv[index + 1] : undefined; + if (!artifactBase) { + throw new Error("--artifact-base is required"); + } + return { artifactBase: path.resolve(artifactBase), repoRoot: process.cwd() }; +} + +async function connectOperator( + gateway: Gateway, + events: GatewayEvent[], + deviceIdentity: NonNullable[0]["deviceIdentity"]>, +): Promise { + return await new Promise((resolve, reject) => { + let settled = false; + const finish = (error?: Error) => { + if (settled) { + return; + } + settled = true; + clearTimeout(timeout); + if (error) { + client.stop(); + reject(error); + } else { + resolve(client); + } + }; + const timeout = setTimeout(() => finish(new Error("operator connection timed out")), 30_000); + timeout.unref(); + const client = new GatewayClient({ + url: gateway.wsUrl, + origin: "http://127.0.0.1", + token: gateway.token, + env: gateway.runtimeEnv, + role: "operator", + clientName: GATEWAY_CLIENT_NAMES.CONTROL_UI, + clientDisplayName: "Cloud mid-turn loss QA operator", + clientVersion: "1.0.0", + platform: process.platform, + mode: GATEWAY_CLIENT_MODES.WEBCHAT, + scopes: ["operator.admin", "operator.read", "operator.write"], + deviceIdentity, + requestTimeoutMs: PROOF_TIMEOUT_MS, + onEvent: (event) => events.push(event), + onHelloOk: () => finish(), + onConnectError: (error) => finish(error), + onClose: (code, reason) => finish(new Error(`Gateway closed (${code}): ${reason}`)), + }); + client.start(); + }); +} + +function messageRole(message: unknown): string { + return String(requireRecord(message, "history message").role ?? ""); +} + +function messageText(message: unknown): string { + const content = requireRecord(message, "history message").content; + if (typeof content === "string") { + return content; + } + if (!Array.isArray(content)) { + return ""; + } + return content + .flatMap((part) => { + const record = part && typeof part === "object" ? (part as Record) : {}; + return typeof record.text === "string" ? [record.text] : []; + }) + .join(""); +} + +async function readHistory(client: GatewayClient): Promise { + const history = await client.request("chat.history", { + sessionKey: SESSION_KEY, + limit: 100, + }); + return history.messages ?? []; +} + +function markerCounts(messages: readonly unknown[]) { + const text = messages.map(messageText).join("\n"); + return Object.fromEntries( + COMMITTED_MARKERS.map((marker) => [marker, text.split(marker).length - 1]), + ); +} + +async function waitForOutbound( + state: ReturnType, + cursor: number, + marker: string, +): Promise { + await waitFor(`qa-channel outbound ${marker}`, () => + state + .getSnapshot() + .messages.slice(cursor) + .some((message) => message.direction === "outbound" && message.text.includes(marker)) + ? true + : undefined, + ); +} + +async function waitForFailedPlacement(gateway: Gateway) { + return await waitFor("failed worker placement", async () => { + const payload = requireRecord( + await gateway.call("sessions.describe", { key: SESSION_KEY }), + "sessions.describe", + ); + const session = requireRecord(payload.session, "described session"); + const placement = requireRecord(session.placement, "session placement"); + return placement.state === "failed" ? placement : undefined; + }); +} + +function waitForVolatilePreview(events: readonly GatewayEvent[], runId: string) { + return waitFor("volatile sidebar preview", () => { + const agentVisible = events.some((event) => { + if (event.event !== "agent") { + return false; + } + const payload = requireRecord(event.payload, "agent event"); + return payload.runId === runId && JSON.stringify(payload.data ?? {}).includes(VOLATILE_TEXT); + }); + const chatText = events + .filter((event) => event.event === "chat") + .map((event) => requireRecord(event.payload, "chat event")) + .filter((payload) => payload.runId === runId && payload.state === "delta") + .map((payload) => (typeof payload.deltaText === "string" ? payload.deltaText : "")) + .join(""); + return agentVisible || chatText.includes(VOLATILE_TEXT) ? true : undefined; + }); +} + +function waitForChatError(events: readonly GatewayEvent[], runId: string) { + return waitFor("operator-visible chat error", () => { + const found = events.find((event) => { + if (event.event !== "chat") { + return false; + } + const payload = requireRecord(event.payload, "chat event"); + return payload.runId === runId && payload.state === "error"; + }); + return found ? requireRecord(found.payload, "chat error") : undefined; + }); +} + +async function runProof(options: ProducerOptions) { + // openclaw-temp-dir: allow standalone QA producer owns and removes this fixture root. + const fixtureRoot = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-cloud-midturn-loss-")); + const state = createQaBusState(); + const bus = await startQaBusServer({ state }); + const provider = await startMidturnProvider(); + const ssh = await createSshdFixture(fixtureRoot); + let sshd = await ssh.start(); + let gateway: Gateway | undefined; + let operator: GatewayClient | undefined; + let proofError: unknown; + let verdict: Record | undefined; + try { + const repo = await initializeRepository(fixtureRoot); + const sshPrivateKey = await fs.readFile(ssh.clientKeyPath, "utf8"); + const transport = createQaChannelTransport(state); + gateway = await startQaGatewayChild({ + repoRoot: options.repoRoot, + useRepoCli: true, + providerBaseUrl: `${provider.baseUrl}/v1`, + providerMode: "mock-openai", + primaryModel: MODEL_REF, + alternateModel: MODEL_REF, + transport, + transportBaseUrl: bus.baseUrl, + 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", + }, + }, + }, + }, + }, + }), + }); + const events: GatewayEvent[] = []; + const deviceIdentity = loadOrCreateDeviceIdentity({ + path: path.join(fixtureRoot, "operator-identity.sqlite"), + }); + operator = await connectOperator(gateway, events, deviceIdentity); + await operator.request("sessions.create", { + key: SESSION_KEY, + agentId: "qa", + worktree: true, + worktreeName: `cloud-midturn-${randomUUID().slice(0, 8)}`, + worktreeBaseRef: "main", + cwd: repo, + }); + await operator.request("sessions.messages.subscribe", { key: SESSION_KEY }); + + const baselineCursor = state.getSnapshot().messages.length; + state.addInboundMessage({ + conversation: { id: SENDER_ID, kind: "direct" }, + senderId: SENDER_ID, + senderName: SENDER_ID, + text: BASELINE_PROMPT, + }); + await waitForOutbound(state, baselineCursor, BASELINE_REPLY); + + await gateway.call( + "sessions.dispatch", + { key: SESSION_KEY, profileId: PROFILE_ID }, + { timeoutMs: PROOF_TIMEOUT_MS }, + ); + const runId = `cloud-midturn-loss-${randomUUID()}`; + const started = await operator.request("chat.send", { + sessionKey: SESSION_KEY, + message: MIDTURN_PROMPT, + deliver: false, + idempotencyKey: runId, + }); + if (started.status !== "started" || started.runId !== runId) { + throw new Error(`chat.send did not start the worker turn: ${JSON.stringify(started)}`); + } + await provider.partialStarted; + const committedBeforeKill = await waitFor("four committed worker messages", async () => { + const messages = await readHistory(operator as GatewayClient); + const counts = markerCounts(messages); + return Object.values(counts).every((count) => count === 1) ? messages : undefined; + }); + await waitForVolatilePreview(events, runId); + + const killed = await killSshdProcessTree(sshd); + const waitResult = await operator.request( + "agent.wait", + { runId, timeoutMs: PROOF_TIMEOUT_MS }, + { timeoutMs: PROOF_TIMEOUT_MS + 5_000 }, + ); + const chatError = await waitForChatError(events, runId); + const failedPlacement = await waitForFailedPlacement(gateway); + const terminalReason = String(failedPlacement.terminalReason ?? ""); + if (!terminalReason || terminalReason.length > 1_024) { + throw new Error(`placement terminal reason was missing or unbounded: ${terminalReason}`); + } + const historyAfterFailure = await readHistory(operator); + const countsAfterFailure = markerCounts(historyAfterFailure); + const committedSequence = historyAfterFailure.flatMap((message) => { + const text = messageText(message); + const marker = COMMITTED_MARKERS.find((candidate) => text.includes(candidate)); + return marker ? [{ role: messageRole(message), marker }] : []; + }); + if ( + !isDeepStrictEqual(historyAfterFailure, committedBeforeKill) || + committedSequence.length !== COMMITTED_MARKERS.length || + committedSequence.some((entry, index) => entry.marker !== COMMITTED_MARKERS[index]) || + historyAfterFailure.some((message) => messageText(message).includes(VOLATILE_TEXT)) || + Object.values(countsAfterFailure).some((count) => count !== 1) + ) { + throw new Error(`unexpected durable cutoff: ${JSON.stringify(committedSequence)}`); + } + + sshd = await ssh.start(); + const redispatched = requireRecord( + await gateway.call( + "sessions.dispatch", + { key: SESSION_KEY, profileId: PROFILE_ID }, + { timeoutMs: PROOF_TIMEOUT_MS }, + ), + "sessions.dispatch redispatch", + ); + const recoveryRunId = `cloud-midturn-recovery-${randomUUID()}`; + const recoveryStarted = await operator.request("chat.send", { + sessionKey: SESSION_KEY, + message: CONTEXT_PROMPT, + deliver: false, + idempotencyKey: recoveryRunId, + }); + if (recoveryStarted.status !== "started" || recoveryStarted.runId !== recoveryRunId) { + throw new Error(`recovery chat.send did not start: ${JSON.stringify(recoveryStarted)}`); + } + const recoveryResult = await operator.request( + "agent.wait", + { runId: recoveryRunId, timeoutMs: PROOF_TIMEOUT_MS }, + { timeoutMs: PROOF_TIMEOUT_MS + 5_000 }, + ); + if (recoveryResult.status !== "ok") { + throw new Error(`recovery turn failed: ${JSON.stringify(recoveryResult)}`); + } + const historyAfterRecovery = await waitFor("durable recovery reply", async () => { + const messages = await readHistory(operator as GatewayClient); + return messages.some((message) => messageText(message).includes(CONTEXT_REPLY)) + ? messages + : undefined; + }); + const recoveryCounts = markerCounts(historyAfterRecovery); + if ( + !COMMITTED_MARKERS.every((marker) => provider.contextRequest.includes(marker)) || + provider.contextRequest.includes(VOLATILE_TEXT) || + Object.values(recoveryCounts).some((count) => count !== 1) + ) { + throw new Error( + "redispatched inference did not preserve exactly one copy of each checkpoint", + ); + } + + verdict = { + status: "pass", + providerMode: "mock-openai", + channel: "qa-channel", + workerProvider: "static-ssh", + sessionKey: SESSION_KEY, + killedWorker: killed, + durableTranscript: { + cutoff: COMMITTED_MARKERS.length, + exactPreKillSnapshotRetained: true, + historyMessageCount: historyAfterFailure.length, + exactMarkers: COMMITTED_MARKERS, + sequence: committedSequence, + markerCounts: countsAfterFailure, + volatileMessagePersisted: false, + }, + livePreview: { + text: VOLATILE_TEXT, + deliveredBeforeDeath: true, + absentFromDurableTranscript: true, + visibleFailureAfterDeath: true, + }, + turnFailure: { + agentWaitStatus: waitResult.status, + chatError: String(chatError.errorMessage ?? chatError.error ?? "worker turn failed"), + terminalReason, + terminalReasonLength: terminalReason.length, + }, + redispatch: { + placementState: requireRecord(redispatched.placement, "redispatched placement").state, + contextContainedCutoff: true, + contextExcludedVolatilePreview: true, + reply: CONTEXT_REPLY, + turnStatus: recoveryResult.status, + markerCounts: recoveryCounts, + }, + providerRequestCount: provider.requestCount, + historyMessageCountBeforeKill: committedBeforeKill.length, + }; + await fs.mkdir(options.artifactBase, { recursive: true }); + await fs.writeFile( + path.join(options.artifactBase, VERDICT_FILE), + `${JSON.stringify(verdict, null, 2)}\n`, + "utf8", + ); + } catch (error) { + proofError = error; + } + + const cleanup = await Promise.allSettled([ + operator?.stopAndWait({ timeoutMs: 1_000 }) ?? Promise.resolve(), + gateway?.stop() ?? Promise.resolve(), + stopSshd(sshd), + provider.stop(), + bus.stop(), + fs.rm(fixtureRoot, { recursive: true, force: true }), + ]); + const cleanupFailures = cleanup.flatMap((result) => + result.status === "rejected" ? [result.reason] : [], + ); + if (cleanupFailures.length > 0) { + proofError = new AggregateError( + proofError ? [proofError, ...cleanupFailures] : cleanupFailures, + "cloud worker mid-turn loss cleanup failed", + proofError ? { cause: proofError } : undefined, + ); + } + if (proofError) { + throw proofError; + } + if (!verdict) { + throw new Error("cloud worker mid-turn loss proof produced no verdict"); + } + return verdict; +} + +async function runProducer(options: ProducerOptions): Promise { + const writer = createQaScriptEvidenceWriter({ + artifactBase: options.artifactBase, + logFileName: `${SCENARIO_ID}.log`, + primaryModel: MODEL_REF, + providerMode: "mock-openai", + repoRoot: options.repoRoot, + target: { + id: SCENARIO_ID, + title: "Cloud worker mid-turn machine loss", + sourcePath: `qa/scenarios/runtime/${SCENARIO_ID}.yaml`, + docsRefs: ["docs/gateway/cloud-workers.md", "docs/concepts/qa-e2e-automation.md"], + codeRefs: [ + "src/worker/embedded-agent-transcript.runtime.ts", + "src/gateway/worker-environments/transcript-commit.ts", + "src/gateway/worker-environments/worker-turn-launcher.ts", + ], + }, + }); + const startedAt = Date.now(); + try { + const verdict = await runProof(options); + writer.appendLog(`pass: ${JSON.stringify(verdict)}\n`); + 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", + durationMs: Math.max(1, Date.now() - startedAt), + status: "pass", + }); + } catch (error) { + const details = error instanceof Error ? error.message : String(error); + writer.appendLog(`fail: ${details}\n`); + return await writer.write({ + details, + durationMs: Math.max(1, Date.now() - startedAt), + status: "fail", + }); + } +} + +async function main(argv: readonly string[]) { + const options = parseOptions(argv); + const evidence = await runProducer(options); + const status = evidence.entries[0]?.result.status; + console.log(`Cloud worker mid-turn loss evidence: ${QA_EVIDENCE_FILENAME}`); + console.log( + `Cloud worker mid-turn loss verdict: ${path.join(options.artifactBase, VERDICT_FILE)}`, + ); + if (status === "pass") { + console.log((await fs.readFile(path.join(options.artifactBase, VERDICT_FILE), "utf8")).trim()); + } + return status === "pass" ? 0 : 1; +} + +if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) { + main(process.argv.slice(2)) + .then((exitCode) => { + process.exitCode = exitCode; + }) + .catch((error: unknown) => { + console.error(error instanceof Error ? error.message : String(error)); + process.exitCode = 1; + }); +}